diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index bd79ecda92a..f2e4797f550 100644 --- a/python/packages/ag-ui/AGENTS.md +++ b/python/packages/ag-ui/AGENTS.md @@ -26,7 +26,8 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard. ## Protocol Notes -- Outbound custom events are emitted as AG-UI `CUSTOM`. +- Outbound custom events are emitted as AG-UI `CUSTOM`. Content-free `AgentResponseUpdate` metadata does not + produce a workflow-output event. - Usage metadata from `Content(type="usage")` is surfaced as `CUSTOM` events with `name="usage"`. - Text annotations are surfaced as `CUSTOM` events with `name="annotations"` and `value={"messageId": ..., "annotations": [...]}`. Annotation-only updates announce a text message if needed diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py b/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py index 639fc223b25..ef1bb5ab6ca 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py @@ -1539,6 +1539,8 @@ def _drain_open_blocks() -> list[BaseEvent]: # message. "output" keeps the terminal-message behavior. is_intermediate = event_type in {"intermediate", "data"} output_payload = getattr(event, "data", None) + if isinstance(output_payload, AgentResponseUpdate) and not output_payload.contents: + continue if isinstance(output_payload, BaseEvent): yield output_payload continue diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py index bcd2ddeda23..7a418c6bdb6 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py +++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py @@ -1326,6 +1326,34 @@ async def scripted_stream(messages: Any, options: Any, **kwargs: Any) -> AsyncIt ] +@pytest.mark.parametrize("intermediate", [False, True]) +async def test_workflow_metadata_only_updates_do_not_emit_custom_output(intermediate: bool) -> None: + @executor(id="participant") + async def participant(message: Any, ctx: WorkflowContext[Any, AgentResponseUpdate]) -> None: + del message + update = AgentResponseUpdate(additional_properties={"custom": {"metadata": "only"}}) + await ctx.yield_output(update) + + workflow = WorkflowBuilder( + start_executor=participant, + output_from=[] if intermediate else [participant], + intermediate_output_from=[participant] if intermediate else [], + ).build() + events = [ + event + async for event in run_workflow_stream( + {"messages": [{"role": "user", "content": "Test"}]}, + workflow, + ) + ] + + assert not [ + event for event in events if event.type == "CUSTOM" and getattr(event, "name", None) == "workflow_output" + ] + assert not [event for event in events if event.type == "RUN_ERROR"] + assert events[-1].type == "RUN_FINISHED" + + async def test_workflow_stream_reconciles_result_from_final_response() -> None: """A final response should complete, but not repeat, its exposed tool call.""" function_call = Content.from_function_call( diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index ad8a87b1080..9917995c2a4 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -359,6 +359,9 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID the shim. - Core orchestration builds `EvalItem` instances through private helpers; callers needing manual control construct the public `EvalItem` directly. +- Agent responses retain schema-only tool snapshots keyed by tool name, including context-provider tools, under + `additional_properties["agent_framework.evaluation"]` for evaluation after serialization; executable callbacks are + not persisted. `evaluate_agent(tools=...)` overrides discovery without changing agent execution. ## Built-in Providers diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py index a6e7ccddf88..e993a057744 100644 --- a/python/packages/core/agent_framework/_agents.py +++ b/python/packages/core/agent_framework/_agents.py @@ -5,7 +5,7 @@ import logging import re import sys -from collections.abc import Awaitable, Callable, Mapping, MutableMapping, Sequence +from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, MutableMapping, Sequence from contextlib import AbstractAsyncContextManager, AsyncExitStack from copy import deepcopy from functools import partial @@ -51,6 +51,7 @@ ) from ._telemetry import FeatureIndex, mark_feature_used from ._types import ( + _EVALUATION_TOOLS_METADATA_KEY, # pyright: ignore[reportPrivateUsage] AgentResponse, AgentResponseUpdate, AgentRunInputs, @@ -60,6 +61,7 @@ ResponseStream, _append_instructions, # pyright: ignore[reportPrivateUsage] _build_agent_response_from_chat_response, # pyright: ignore[reportPrivateUsage] + _capture_evaluation_tools, # pyright: ignore[reportPrivateUsage] map_chat_to_agent_update, normalize_messages, ) @@ -275,6 +277,7 @@ class _RunContext(TypedDict): agent_name: str suppress_response_id: bool chat_options: MutableMapping[str, Any] + evaluation_tools: dict[str, dict[str, Any]] compaction_strategy: CompactionStrategy | None tokenizer: TokenizerProtocol | None client_kwargs: Mapping[str, Any] @@ -396,7 +399,8 @@ def run( messages: The message(s) to send to the agent. Keyword Args: - stream: Whether to stream the response. Defaults to False. + stream: Whether to stream the response. Defaults to False. Streams + may include response metadata updates without message contents. session: The conversation session associated with the message(s). function_invocation_kwargs: Keyword arguments forwarded to tool invocation. client_kwargs: Additional client-specific keyword arguments. @@ -1406,6 +1410,7 @@ async def _parse_non_streaming_response( response_format=context["chat_options"].get("response_format"), suppress_response_id=context["suppress_response_id"], ) + agent_response.additional_properties[_EVALUATION_TOOLS_METADATA_KEY] = context["evaluation_tools"] session_context = context["session_context"] session_context._response = agent_response # type: ignore[assignment] await self._run_after_providers(session=session, context=session_context) @@ -1442,6 +1447,7 @@ async def _post_hook(response: AgentResponse) -> None: session_context = context["session_context"] if context["suppress_response_id"]: response.response_id = None + response.additional_properties[_EVALUATION_TOOLS_METADATA_KEY] = context["evaluation_tools"] session_context._response = response # type: ignore[assignment] # Result hooks run during finalization, outside the per-pull identity # scope registered below, so re-stamp this run's identity around its @@ -1482,7 +1488,25 @@ def _finalizer(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]: # carries this run's identity; nested runs re-stamp their own within theirs. stream = stream.with_pull_context_manager(partial(_run_identity_scope, run_identity)) - return stream.with_transform_hook(_propagate_conversation_id).with_result_hook(_post_hook) + stream = stream.with_transform_hook(_propagate_conversation_id).with_result_hook(_post_hook) + + async def _with_evaluation_metadata() -> AsyncIterable[AgentResponseUpdate]: + async with stream: + async for update in stream: + yield update + response = await stream.get_final_response() + yield AgentResponseUpdate( + additional_properties={ + _EVALUATION_TOOLS_METADATA_KEY: deepcopy( + response.additional_properties[_EVALUATION_TOOLS_METADATA_KEY] + ) + } + ) + + async def _finalize_with_metadata(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]: + return await stream.get_final_response() + + return ResponseStream(_with_evaluation_metadata(), finalizer=_finalize_with_metadata) def _finalize_response_updates( self, @@ -1760,6 +1784,11 @@ async def _prepare_run_context( per_service_call_history_middleware, ] + evaluation_tools: dict[str, dict[str, Any]] = {} + _capture_evaluation_tools(evaluation_tools, co.get("tools")) + if isinstance(self.client, BaseChatClient): + effective_client_kwargs["_evaluation_tool_capture"] = partial(_capture_evaluation_tools, evaluation_tools) + return { "session": active_session, "session_context": session_context, @@ -1768,6 +1797,7 @@ async def _prepare_run_context( "agent_name": agent_name, "suppress_response_id": bool(per_service_call_history_providers) and not service_stores_history, "chat_options": co, + "evaluation_tools": evaluation_tools, "compaction_strategy": compaction_strategy or self.compaction_strategy, "tokenizer": tokenizer or self.tokenizer, "client_kwargs": effective_client_kwargs, diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index 564d78aa660..9613a3d3028 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -516,6 +516,11 @@ def get_response( tokenizer=tokenizer, ) merged_client_kwargs = dict(client_kwargs) if client_kwargs is not None else {} + evaluation_tool_capture = cast( + Callable[[Any], None] | None, merged_client_kwargs.pop("_evaluation_tool_capture", None) + ) + if evaluation_tool_capture is not None: + evaluation_tool_capture((options or {}).get("tools")) if not compaction_overrides: return self._inner_get_response( diff --git a/python/packages/core/agent_framework/_evaluation.py b/python/packages/core/agent_framework/_evaluation.py index 1c57b9f72d8..e8473ee7f9a 100644 --- a/python/packages/core/agent_framework/_evaluation.py +++ b/python/packages/core/agent_framework/_evaluation.py @@ -56,7 +56,11 @@ from ._feature_stage import ExperimentalFeature, experimental from ._tools import FunctionTool, normalize_tools -from ._types import AgentResponse, Message +from ._types import ( + _EVALUATION_TOOLS_METADATA_KEY, # pyright: ignore[reportPrivateUsage] + AgentResponse, + Message, +) if TYPE_CHECKING: from ._agents import SupportsAgentRun @@ -872,8 +876,30 @@ def _to_eval_item( all_msgs = list(input_msgs) + list(response.messages or []) typed_tools: list[FunctionTool] = [] - if tools: + if tools is not None: typed_tools = [tool for tool in normalize_tools(tools) if isinstance(tool, FunctionTool)] + elif _EVALUATION_TOOLS_METADATA_KEY in response.additional_properties: + definitions = response.additional_properties[_EVALUATION_TOOLS_METADATA_KEY] + if not isinstance(definitions, dict): + raise ValueError("Invalid agent_framework.evaluation metadata: expected tool definitions keyed by name.") + for name, definition in cast(dict[str, Any], definitions).items(): + if not isinstance(name, str) or not name or not isinstance(definition, dict): + raise ValueError("Invalid agent_framework.evaluation metadata: expected named tool definition objects.") + definition = cast(dict[str, Any], definition) + description = definition.get("description") + schema = definition.get("input_schema") + if not isinstance(description, str) or not isinstance(schema, dict): + raise ValueError( + f"Invalid agent_framework.evaluation definition for {name!r}: " + "expected a string description and an input_schema object." + ) + try: + schema = json.loads(json.dumps(schema, allow_nan=False)) + except (TypeError, ValueError) as exc: + raise ValueError( + f"Invalid agent_framework.evaluation input_schema for {name!r}: expected JSON-serializable data." + ) from exc + typed_tools.append(FunctionTool(name=name, description=description, input_model=schema)) elif agent: raw_tools = getattr(agent, "default_options", {}).get("tools", []) typed_tools = [tool for tool in normalize_tools(raw_tools) if isinstance(tool, FunctionTool)] @@ -1604,6 +1630,7 @@ async def evaluate_agent( expected_output: str | Sequence[str] | None = None, expected_tool_calls: Sequence[ExpectedToolCall] | Sequence[Sequence[ExpectedToolCall]] | None = None, responses: AgentResponse[Any] | Sequence[AgentResponse[Any]] | None = None, + tools: FunctionTool | Callable[..., Any] | Sequence[FunctionTool | Callable[..., Any]] | None = None, evaluators: Evaluator | Callable[..., Any] | Sequence[Evaluator | Callable[..., Any]], eval_name: str | None = None, context: str | None = None, @@ -1621,8 +1648,12 @@ async def evaluate_agent( or a list for convenience. If ``responses`` is provided, skips running the agent and evaluates those - responses directly — but still extracts tool definitions from the agent. - In this mode ``queries`` is required to construct the conversation. + responses directly. Tool definitions come from the original run when + available, including schema-only snapshots retained through JSON or pickle + serialization. Responses without captured definitions fall back to the + agent's configured tools; supply ``tools`` when those definitions are + unavailable or need to be overridden. In this mode ``queries`` is required + to construct the conversation. Args: agent: An agent-framework agent instance. @@ -1642,6 +1673,9 @@ async def evaluate_agent( running the agent. A single response is wrapped into a one-element list. When provided, ``queries`` must also be provided to construct the conversation for evaluation. + tools: Tool definitions for evaluation, overriding automatic discovery + for every item. These are not passed to ``agent.run``. An empty + sequence disables tool definitions. evaluators: One or more ``Evaluator`` instances. eval_name: Display name (defaults to agent name). context: Optional context for groundedness evaluation. @@ -1746,6 +1780,7 @@ async def evaluate_agent( query=q, response=r, agent=agent, + tools=tools, context=context, ) ) @@ -1766,6 +1801,7 @@ async def evaluate_agent( query=query, response=response, agent=agent, + tools=tools, context=context, ) ) diff --git a/python/packages/core/agent_framework/_harness/_loop.py b/python/packages/core/agent_framework/_harness/_loop.py index 0a147824335..6ab54d6daee 100644 --- a/python/packages/core/agent_framework/_harness/_loop.py +++ b/python/packages/core/agent_framework/_harness/_loop.py @@ -50,6 +50,7 @@ ResponseStream, UsageDetails, add_usage_details, + merge_additional_properties, normalize_messages, ) @@ -544,6 +545,7 @@ async def _process_non_streaming( # injected "nudge" messages, used to build the combined response when return_final_only=False. aggregated: list[Message] = [] aggregated_usage: UsageDetails | None = None + additional_properties: dict[str, Any] = {} final_result: AgentResponse | None = None stamped_options = dict(context.options) if context.options is not None else {} stamped_options[_LOOP_ITERATION_TOKEN_KEY] = object() @@ -562,6 +564,7 @@ async def _process_non_streaming( final_result = result aggregated.extend(result.messages) + additional_properties = merge_additional_properties(additional_properties, result.additional_properties) if result.usage_details is not None: aggregated_usage = add_usage_details(aggregated_usage, result.usage_details) @@ -619,6 +622,7 @@ async def _process_non_streaming( if not self.return_final_only: context.result = self._aggregate_response(final_result, aggregated, aggregated_usage) + context.result.additional_properties = additional_properties await self._fire_turn_scoped_after_providers( context, context.result if isinstance(context.result, AgentResponse) else final_result, diff --git a/python/packages/core/agent_framework/_harness/_tool_approval.py b/python/packages/core/agent_framework/_harness/_tool_approval.py index 2232093ea3f..f65cd610de3 100644 --- a/python/packages/core/agent_framework/_harness/_tool_approval.py +++ b/python/packages/core/agent_framework/_harness/_tool_approval.py @@ -562,6 +562,8 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]: remaining_update = copy.copy(update) remaining_update.contents = remaining_contents yield remaining_update + elif update.additional_properties: + yield AgentResponseUpdate(additional_properties=update.additional_properties) if not all_auto_approved or has_other_user_input: return context.messages = [] diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 3106e91ba40..b2977b9d13e 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -55,6 +55,7 @@ _SERIALIZED_EXCEPTION_MARKER: Final[str] = "FunctionInvocationError" _OperationState: TypeAlias = Literal["in_progress", "terminal"] _OPERATION_STATE_SERIALIZATION_KEY: Final[str] = "_operation_state" +_EVALUATION_TOOLS_METADATA_KEY: Final[str] = "agent_framework.evaluation" if TYPE_CHECKING: from pydantic import BaseModel @@ -62,6 +63,21 @@ from ._tools import ToolTypes +def _capture_evaluation_tools( # pyright: ignore[reportUnusedFunction] + captured: dict[str, dict[str, Any]], tools: Any +) -> None: + """Retain exposed declarations, refreshing same-name schemas at each model call.""" + from ._tools import FunctionTool, normalize_tools + + for tool in normalize_tools(tools or []): + if not isinstance(tool, FunctionTool): + continue + captured[tool.name] = { + "description": tool.description, + "input_schema": json.loads(json.dumps(tool.parameters(), allow_nan=False)), + } + + # region Content Parsing Utilities @@ -2501,63 +2517,70 @@ def prepend_instructions_to_messages( def _process_update(response: ChatResponse | AgentResponse, update: ChatResponseUpdate | AgentResponseUpdate) -> None: - """Processes a single update and modifies the response in place.""" - is_new_message = False + """Merge an update, creating a message only when it carries message-level data.""" if ( - not response.messages - or ( - update.message_id - and response.messages[-1].message_id - and response.messages[-1].message_id != update.message_id - ) - or ( - update.author_name - and response.messages[-1].author_name - and response.messages[-1].author_name != update.author_name - ) - or (update.role and response.messages[-1].role != update.role) + update.contents + or update.role is not None + or update.author_name is not None + or update.message_id is not None + or update.merge_mode == "replace" ): - is_new_message = True + is_new_message = False + if ( + not response.messages + or ( + update.message_id + and response.messages[-1].message_id + and response.messages[-1].message_id != update.message_id + ) + or ( + update.author_name + and response.messages[-1].author_name + and response.messages[-1].author_name != update.author_name + ) + or (update.role and response.messages[-1].role != update.role) + ): + is_new_message = True - if is_new_message: - message = Message("assistant", []) - response.messages.append(message) - else: - message = response.messages[-1] - if getattr(update, "merge_mode", None) == "replace": - message.contents.clear() - # Incorporate the update's properties into the message. - if update.author_name is not None: - message.author_name = update.author_name - if update.role is not None: - message.role = update.role - if update.message_id: - message.message_id = update.message_id - for content in update.contents: - # Fast path: get type attribute (most content will have it) - content_type = getattr(content, "type", None) - # Slow path: only check for dict if type is None - if content_type is None and isinstance(content, (dict, MutableMapping)): - try: - content = Content.from_dict(content) - content_type = content.type - except ContentError as exc: - logger.warning(f"Skipping unknown content type or invalid content: {exc}") - continue - match content_type: - case "function_call": - _merge_function_call_content(message, content) - case "usage": - if content.merge_mode == "replace": - response.usage_details = deepcopy(content.usage_details) - elif response.usage_details is None: - response.usage_details = UsageDetails() - response.usage_details = add_usage_details(response.usage_details, content.usage_details) - else: - # mypy doesn't narrow type based on match/case, but we know this is UsageContent - response.usage_details = add_usage_details(response.usage_details, content.usage_details) - case _: - message.contents.append(content) + if is_new_message: + message = Message("assistant", []) + response.messages.append(message) + else: + message = response.messages[-1] + if update.merge_mode == "replace": + message.contents.clear() + # Incorporate the update's properties into the message. + if update.author_name is not None: + message.author_name = update.author_name + if update.role is not None: + message.role = update.role + if update.message_id: + message.message_id = update.message_id + for content in update.contents: + # Fast path: get type attribute (most content will have it) + content_type = getattr(content, "type", None) + # Slow path: only check for dict if type is None + if content_type is None and isinstance(content, (dict, MutableMapping)): + try: + content = Content.from_dict(content) + content_type = content.type + except ContentError as exc: + logger.warning(f"Skipping unknown content type or invalid content: {exc}") + continue + match content_type: + case "function_call": + _merge_function_call_content(message, content) + case "usage": + if content.merge_mode == "replace": + response.usage_details = deepcopy(content.usage_details) + elif response.usage_details is None: + response.usage_details = UsageDetails() + response.usage_details = add_usage_details(response.usage_details, content.usage_details) + else: + # mypy doesn't narrow type based on match/case, but we know this is UsageContent + response.usage_details = add_usage_details(response.usage_details, content.usage_details) + case _: + message.contents.append(content) # Incorporate the update's properties into the response. if update.response_id: response.response_id = update.response_id @@ -3569,6 +3592,10 @@ class AgentResponse(SerializationMixin, Generic[ResponseModelT]): not on the `AgentResponse` itself. Use `response.messages[0].author_name` to access the author name of individual messages. + Agent runs retain schema-only tool snapshots, keyed by tool name, under + ``additional_properties["agent_framework.evaluation"]``. These snapshots survive + JSON and pickle serialization and contain no executable implementations. + Examples: .. code-block:: python diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index 1481ce8a77b..dfb41567ae1 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -1544,7 +1544,10 @@ async def after_run( assert response.usage_details == {"input_token_count": 3, "output_token_count": 2, "total_token_count": 5} assert response.value == {"answer": "ok"} assert response.continuation_token == {"token": "next"} - assert response.additional_properties == {"provider": "test"} + assert response.additional_properties == { + "provider": "test", + "agent_framework.evaluation": {}, + } assert response.raw_representation is raw_response @@ -1633,7 +1636,10 @@ async def after_run( assert response.usage_details == {"input_token_count": 4, "output_token_count": 3, "total_token_count": 7} assert response.value == {"answer": "ok"} assert response.continuation_token == {"token": "stream-next"} - assert response.additional_properties == {"provider": "stream-test"} + assert response.additional_properties == { + "provider": "stream-test", + "agent_framework.evaluation": {}, + } assert response.raw_representation == [raw_update] diff --git a/python/packages/core/tests/core/test_as_tool_kwargs_propagation.py b/python/packages/core/tests/core/test_as_tool_kwargs_propagation.py index 178528510a4..46743d5a9c6 100644 --- a/python/packages/core/tests/core/test_as_tool_kwargs_propagation.py +++ b/python/packages/core/tests/core/test_as_tool_kwargs_propagation.py @@ -213,7 +213,7 @@ async def stream_callback(update: Any) -> None: assert captured_kwargs == {} assert captured_function_invocation_kwargs["api_key"] == "streaming-key-999" - assert len(captured_updates) == 1 + assert len([update for update in captured_updates if update.contents]) == 1 async def test_as_tool_empty_kwargs_still_works(self, client: MockChatClient) -> None: """Test that as_tool works correctly when no extra kwargs are provided.""" diff --git a/python/packages/core/tests/core/test_evaluation.py b/python/packages/core/tests/core/test_evaluation.py index f32e216bd9b..edbe621cadb 100644 --- a/python/packages/core/tests/core/test_evaluation.py +++ b/python/packages/core/tests/core/test_evaluation.py @@ -4,15 +4,52 @@ from __future__ import annotations +import asyncio +import pickle +from copy import deepcopy from typing import Any, cast -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest +from opentelemetry.sdk.metrics import MeterProvider from agent_framework import AgentEvalConverter as ExportedAgentEvalConverter -from agent_framework._evaluation import AgentEvalConverter, ConversationSplit, EvalItem, _to_eval_item -from agent_framework._tools import FunctionTool -from agent_framework._types import AgentResponse, Content, Message +from agent_framework._agents import Agent, SupportsAgentRun +from agent_framework._clients import BaseChatClient +from agent_framework._evaluation import ( + AgentEvalConverter, + ConversationSplit, + EvalItem, + LocalEvaluator, + _to_eval_item, + evaluate_agent, + evaluator, +) +from agent_framework._harness._agent import create_harness_agent +from agent_framework._harness._loop import AgentLoopMiddleware +from agent_framework._mcp import MCPTool +from agent_framework._middleware import FunctionInvocationContext +from agent_framework._sessions import AgentSession, ContextProvider, SessionContext +from agent_framework._skills import InlineSkill, SkillFrontmatter, SkillsProvider +from agent_framework._tools import FunctionInvocationLayer, FunctionTool +from agent_framework._types import ( + AgentResponse, + AgentResponseUpdate, + ChatResponse, + ChatResponseUpdate, + Content, + Message, +) + + +class _FunctionOnlyChatClient(FunctionInvocationLayer[Any], BaseChatClient[Any]): + def __init__(self, backend: Any) -> None: + super().__init__() + self.backend = backend + + def _inner_get_response(self, *, messages: Any, stream: bool, options: Any, **kwargs: Any) -> Any: + assert "_evaluation_tool_capture" not in kwargs + return self.backend._inner_get_response(messages=messages, stream=stream, options=options, **kwargs) class TestAgentEvalConverterCompatibility: @@ -129,6 +166,767 @@ def test_explicit_tools_override_agent(self) -> None: assert item.tools == [explicit_tool] + def test_explicit_empty_tools_override_agent(self) -> None: + agent = MagicMock() + agent.default_options = {"tools": [FunctionTool(name="unused", func=lambda: "")]} + + item = _to_eval_item(query="Test", response=AgentResponse(), agent=agent, tools=[]) + + assert item.tools is None + + @pytest.mark.parametrize("stream", [False, True]) + async def test_with_agent_context_provider_tools(self, chat_client_base: Any, stream: bool) -> None: + skill = InlineSkill( + frontmatter=SkillFrontmatter(name="test-skill", description="A test skill."), + instructions="Skill instructions.", + ) + provider = SkillsProvider(skill, disable_load_skill_approval=True) + provider.before_run = AsyncMock(wraps=provider.before_run) # type: ignore[method-assign] + default_tool = FunctionTool(name="calculate", description="Calculate", func=lambda: "42") + mcp_tool = FunctionTool(name="search", description="Search", func=lambda: "Found it") + mcp = MagicMock(spec=MCPTool) + mcp.functions = [mcp_tool] + mcp.is_connected = True + mcp._prepare_for_run = AsyncMock() + agent = Agent(client=chat_client_base, tools=[default_tool, mcp], context_providers=[provider]) + call = Content.from_function_call("call_1", "load_skill", arguments={"skill_name": "test-skill"}) + if stream: + chat_client_base.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[call])], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Done")])], + ] + response = await agent.run("Use the skill", stream=True).get_final_response() + else: + chat_client_base.run_responses = [ + ChatResponse(messages=[Message("assistant", [call])]), + ChatResponse(messages=[Message("assistant", ["Done"])]), + ] + response = await agent.run("Use the skill") + + item = _to_eval_item(query="Use the skill", response=response, agent=agent) + + assert item.tools is not None + definitions = {tool.name: tool for tool in item.tools} + assert set(definitions) == { + "calculate", + "search", + "load_skill", + "read_skill_resource", + "run_skill_script", + } + assert len(item.tools) == len(definitions) + assert definitions["calculate"].parameters() == default_tool.parameters() + assert definitions["search"].parameters() == mcp_tool.parameters() + assert all(tool.declaration_only for tool in item.tools) + assert definitions["load_skill"].parameters()["required"] == ["skill_name"] + expected_content = await skill.get_content() + assert any( + content.type == "function_result" and content.result == expected_content + for message in response.messages + for content in message.contents + ) + provider.before_run.assert_awaited_once() + + # Post-hoc evaluation must use the original run, not the agent's current configuration. + agent.context_providers.clear() + agent.default_options["tools"] = [] + agent.mcp_tools.clear() + for supplied_agent in (None, agent): + captured = _to_eval_item(query="Use the skill", response=response, agent=supplied_agent).tools + assert captured is not None + assert {tool.name: tool.parameters() for tool in captured} == { + tool.name: tool.parameters() for tool in item.tools + } + explicit_tool = FunctionTool(name="explicit", func=lambda: "") + assert _to_eval_item(query="Test", response=response, tools=[explicit_tool]).tools == [explicit_tool] + assert _to_eval_item(query="Test", response=response, tools=[]).tools is None + serialized = response.to_dict() + assert "tools" not in serialized + assert "_evaluation_tools" not in serialized + assert AgentResponse.from_json(response.to_json()).messages[0].contents[0].name == "load_skill" + restored = pickle.loads(pickle.dumps(response)) + assert not hasattr(restored, "_evaluation_tools") + for restored_response in [ + AgentResponse.from_dict(serialized), + AgentResponse.from_json(response.to_json()), + restored, + ]: + restored_item = _to_eval_item(query="Use the skill", response=restored_response) + assert restored_item.tools is not None + assert {tool.name: (tool.description, tool.parameters()) for tool in restored_item.tools} == { + tool.name: (tool.description, tool.parameters()) for tool in item.tools + } + assert all(tool.declaration_only for tool in restored_item.tools) + + async def test_context_provider_tools_are_isolated_per_run(self, chat_client_base: Any) -> None: + class QueryToolsProvider(ContextProvider): + async def before_run( + self, + *, + agent: SupportsAgentRun, + session: AgentSession, + context: SessionContext, + state: dict[str, Any], + ) -> None: + name = context.input_messages[0].text + context.extend_tools(self.source_id, [FunctionTool(name=name, func=lambda: name)]) + await asyncio.sleep(0) + + provider = QueryToolsProvider("query-tools") + provider.before_run = AsyncMock(wraps=provider.before_run) # type: ignore[method-assign] + agent = Agent(client=chat_client_base, context_providers=[provider]) + first, second = await asyncio.gather(agent.run("first"), agent.run("second")) + + for name, response in [("first", first), ("second", second)]: + item = _to_eval_item(query=name, response=response, agent=agent) + assert item.tools is not None + assert [tool.name for tool in item.tools] == [name] + assert provider.before_run.await_count == 2 + + @pytest.mark.parametrize("stream", [False, True]) + @pytest.mark.parametrize("chat_middleware", [False, True]) + async def test_progressively_exposed_tools_are_retained( + self, chat_client_base: Any, stream: bool, chat_middleware: bool + ) -> None: + def finish(ctx: FunctionInvocationContext) -> str: + ctx.remove_tools(["finish"]) + return "Finished" + + finish_tool = FunctionTool(name="finish", func=finish) + + def load(ctx: FunctionInvocationContext) -> str: + ctx.add_tools([finish_tool]) + return "Loaded" + + loader = FunctionTool(name="load", func=load) + calls = [ + Content.from_function_call("call_1", "load", arguments={}), + Content.from_function_call("call_2", "finish", arguments={}), + ] + client = chat_client_base if chat_middleware else _FunctionOnlyChatClient(chat_client_base) + agent = Agent(client=client, tools=[loader]) + if stream: + chat_client_base.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[call])] for call in calls + ] + response_stream = agent.run("Test", stream=True) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in response_stream] + response = await response_stream.get_final_response() + assert sum("agent_framework.evaluation" in (update.additional_properties or {}) for update in updates) == 1 + assert "agent_framework.evaluation" in (updates[-1].additional_properties or {}) + restored_item = _to_eval_item(query="Test", response=AgentResponse.from_updates(updates)) + assert restored_item.tools is not None + assert [tool.name for tool in restored_item.tools] == ["load", "finish"] + else: + chat_client_base.run_responses = [ChatResponse(messages=[Message("assistant", [call])]) for call in calls] + response = await agent.run("Test") + + item = _to_eval_item(query="Test", response=response) + + assert item.tools is not None + assert [tool.name for tool in item.tools] == ["load", "finish"] + assert all(tool.declaration_only for tool in item.tools) + assert any( + content.type == "function_result" and content.result == "Finished" + for message in response.messages + for content in message.contents + ) + + @pytest.mark.parametrize("stream", [False, True]) + @pytest.mark.parametrize("chat_middleware", [False, True]) + async def test_progressively_replaced_tool_uses_latest_schema( + self, chat_client_base: Any, stream: bool, chat_middleware: bool + ) -> None: + replacement = FunctionTool( + name="search", + description="Search by city", + input_model={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + func=lambda city: city, + ) + + def replace(ctx: FunctionInvocationContext) -> str: + ctx.remove_tools(["search"]) + ctx.add_tools([replacement]) + return "Replaced" + + original = FunctionTool(name="search", description="Load city search", func=replace) + client = chat_client_base if chat_middleware else _FunctionOnlyChatClient(chat_client_base) + agent = Agent(client=client, tools=[original]) + calls = [ + Content.from_function_call("call_1", "search", arguments={}), + Content.from_function_call("call_2", "search", arguments={"city": "Seattle"}), + ] + if stream: + chat_client_base.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[call])] for call in calls + ] + response_stream = agent.run("Test", stream=True) + updates = [update async for update in response_stream] + response = await response_stream.get_final_response() + restored = AgentResponse.from_updates(updates) + else: + chat_client_base.run_responses = [ChatResponse(messages=[Message("assistant", [call])]) for call in calls] + response = await agent.run("Test") + restored = AgentResponse.from_json(response.to_json()) + + for result in (response, restored): + item = _to_eval_item(query="Test", response=result) + assert item.tools is not None + assert len(item.tools) == 1 + assert item.tools[0].description == replacement.description + assert item.tools[0].parameters() == replacement.parameters() + assert any( + content.type == "function_result" and content.result == "Seattle" + for message in response.messages + for content in message.contents + ) + + @pytest.mark.parametrize("serialization", ["dict", "json", "pickle"]) + async def test_empty_snapshot_survives_agent_reconfiguration( + self, chat_client_base: Any, serialization: str + ) -> None: + agent = Agent(client=chat_client_base) + response = await agent.run("No tools") + agent.default_options["tools"] = [FunctionTool(name="unused", func=lambda: "")] + if serialization == "dict": + restored = AgentResponse.from_dict(response.to_dict()) + elif serialization == "json": + restored = AgentResponse.from_json(response.to_json()) + else: + restored = pickle.loads(pickle.dumps(response)) + + assert restored.additional_properties["agent_framework.evaluation"] == {} + assert _to_eval_item(query="No tools", response=restored, agent=agent).tools is None + + @pytest.mark.parametrize("replace_same_name", [False, True]) + async def test_empty_followup_stream_publishes_changed_tools( + self, chat_client_base: Any, replace_same_name: bool + ) -> None: + added_tool = FunctionTool( + name="load" if replace_same_name else "finish", + description="New declaration", + input_model={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + func=lambda city: city, + ) + + def load(ctx: FunctionInvocationContext) -> str: + if replace_same_name: + ctx.remove_tools(["load"]) + ctx.add_tools([added_tool]) + return "Loaded" + + agent = Agent(client=chat_client_base, tools=[FunctionTool(name="load", func=load)]) + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_function_call("call_1", "load", arguments={})], + ) + ], + [], + ] + stream = agent.run("Test", stream=True) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in stream] + final = await stream.get_final_response() + restored = AgentResponse.from_updates(updates) + + assert chat_client_base.call_count == 2 + assert updates[-1].contents == [] + assert ( + restored.additional_properties["agent_framework.evaluation"] + == (final.additional_properties["agent_framework.evaluation"]) + ) + for response in (final, restored): + item = _to_eval_item(query="Test", response=response) + assert item.tools is not None + tools = {tool.name: tool for tool in item.tools} + assert set(tools) == ({"load"} if replace_same_name else {"load", "finish"}) + assert tools[added_tool.name].description == added_tool.description + assert tools[added_tool.name].parameters() == added_tool.parameters() + + @pytest.mark.parametrize("serialization", ["dict", "json", "pickle"]) + async def test_tuple_schema_constraints_survive_serialization( + self, chat_client_base: Any, serialization: str + ) -> None: + schema = { + "type": "object", + "properties": {"city": {"type": "string", "enum": ("Seattle", "Paris")}}, + "required": ("city",), + } + tool = FunctionTool(name="weather", input_model=schema, func=lambda city: city) + response = await Agent(client=chat_client_base, tools=[tool]).run("Weather?") + if serialization == "dict": + restored = AgentResponse.from_dict(response.to_dict()) + elif serialization == "json": + restored = AgentResponse.from_json(response.to_json()) + else: + restored = pickle.loads(pickle.dumps(response)) + + item = _to_eval_item(query="Weather?", response=restored) + assert item.tools is not None + assert item.tools[0].parameters()["properties"]["city"]["enum"] == ["Seattle", "Paris"] + assert item.tools[0].parameters()["required"] == ["city"] + assert tool.parameters()["properties"]["city"]["enum"] == ("Seattle", "Paris") + + @pytest.mark.parametrize("serialization", ["dict", "json", "pickle"]) + async def test_schema_snapshots_do_not_retain_runtime_objects( + self, + chat_client_base: Any, + serialization: str, + ) -> None: + schema = { + "type": "object", + "properties": {"city": {"type": "string", "enum": ["Seattle", "Paris"]}}, + "required": ["city"], + "additionalProperties": False, + } + tool = FunctionTool( + name="weather", + description="Original description", + func=lambda city: city, + input_model=schema, + additional_properties={"runtime_marker": "not-in-snapshot"}, + ) + agent = Agent(client=chat_client_base, tools=[tool]) + response = await agent.run("Weather?") + tool.description = "Changed description" + tool.parameters()["properties"]["city"]["enum"].append("London") + agent.default_options["tools"] = [] + if serialization == "dict": + restored = AgentResponse.from_dict(response.to_dict()) + elif serialization == "json": + restored = AgentResponse.from_json(response.to_json()) + else: + restored = pickle.loads(pickle.dumps(response)) + + item = _to_eval_item(query="Weather?", response=restored, agent=agent) + + assert item.tools is not None + assert len(item.tools) == 1 + captured = item.tools[0] + assert captured.name == "weather" + assert captured.description == "Original description" + assert captured.parameters()["properties"]["city"]["enum"] == ["Seattle", "Paris"] + assert captured.parameters()["additionalProperties"] is False + assert captured.func is None + assert captured.additional_properties is None + metadata = response.to_dict()["additional_properties"]["agent_framework.evaluation"] + assert metadata == { + "weather": { + "description": "Original description", + "input_schema": { + "type": "object", + "properties": {"city": {"type": "string", "enum": ["Seattle", "Paris"]}}, + "required": ["city"], + "additionalProperties": False, + }, + }, + } + assert "not-in-snapshot" not in response.to_json() + + async def test_deepcopy_preserves_snapshots_without_copying_telemetry( + self, chat_client_base: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + meter_provider = MeterProvider(shutdown_on_exit=False) + monkeypatch.setattr("agent_framework._tools.get_meter", lambda: meter_provider.get_meter("evaluation-test")) + monkeypatch.setattr("agent_framework.observability.OBSERVABILITY_SETTINGS.enable_instrumentation", True) + try: + tool = FunctionTool( + name="weather", + description="Get the weather", + input_model={"type": "object", "properties": {"city": {"type": "string"}}}, + func=lambda city: city, + ) + response = await Agent(client=chat_client_base, tools=[tool]).run("Weather?") + copied = deepcopy(response) + + assert not hasattr(copied, "_evaluation_tools") + assert ( + copied.additional_properties["agent_framework.evaluation"] + == (response.additional_properties["agent_framework.evaluation"]) + ) + item = _to_eval_item(query="Weather?", response=copied) + assert item.tools is not None + assert len(item.tools) == 1 + assert item.tools[0].name == tool.name + assert item.tools[0].description == tool.description + assert item.tools[0].parameters() == tool.parameters() + assert item.tools[0].declaration_only + item.tools[0].parameters()["properties"]["city"]["type"] = "number" + assert copied.additional_properties["agent_framework.evaluation"]["weather"]["input_schema"] == ( + tool.parameters() + ) + copied.additional_properties["agent_framework.evaluation"].clear() + assert len(response.additional_properties["agent_framework.evaluation"]) == 1 + finally: + meter_provider.shutdown() + + @pytest.mark.parametrize( + "metadata", + [ + None, + [], + {"weather": None}, + {"": {"description": "", "input_schema": {}}}, + {1: {"description": "", "input_schema": {}}}, + {"weather": {"description": None, "input_schema": {}}}, + {"weather": {"description": "", "input_schema": []}}, + {"weather": {"description": "", "input_schema": {"default": object()}}}, + ], + ) + def test_invalid_serialized_tool_metadata_raises(self, metadata: Any) -> None: + response = AgentResponse(additional_properties={"agent_framework.evaluation": metadata}) + + with pytest.raises(ValueError, match="agent_framework.evaluation"): + _to_eval_item(query="Weather?", response=response) + + explicit = FunctionTool(name="explicit", func=lambda: "") + assert _to_eval_item(query="Weather?", response=response, tools=[explicit]).tools == [explicit] + + def test_serialized_empty_definitions_do_not_fall_back_to_agent_tools(self) -> None: + response = AgentResponse( + additional_properties={ + "agent_framework.evaluation": {}, + } + ) + agent = MagicMock() + agent.default_options = {"tools": [FunctionTool(name="unused", func=lambda: "")]} + + assert _to_eval_item(query="Test", response=response, agent=agent).tools is None + + +class TestEvaluateAgentTools: + @pytest.mark.parametrize("auto_approve", [False, True]) + async def test_harness_stream_preserves_provider_tools(self, chat_client_base: Any, auto_approve: bool) -> None: + skill = InlineSkill( + frontmatter=SkillFrontmatter(name="test-skill", description="A test skill."), + instructions="Skill instructions.", + ) + provider = SkillsProvider(skill) + provider.before_run = AsyncMock(wraps=provider.before_run) # type: ignore[method-assign] + + class ChangingToolsProvider(ContextProvider): + async def before_run( + self, + *, + agent: SupportsAgentRun, + session: AgentSession, + context: SessionContext, + state: dict[str, Any], + ) -> None: + iteration = state["iteration"] = state.get("iteration", 0) + 1 + context.extend_tools( + self.source_id, + [ + FunctionTool(name=f"iteration_{iteration}", func=lambda: ""), + FunctionTool(name="versioned", description=f"Version {iteration}", func=lambda: ""), + ], + ) + + agent = create_harness_agent( + chat_client_base, + context_providers=[provider, ChangingToolsProvider("changing")], + disable_compaction=True, + disable_todo=True, + disable_mode=True, + disable_file_memory=True, + disable_web_search=True, + auto_approval_rules=[SkillsProvider.all_tools_auto_approval_rule] if auto_approve else [], + ) + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[ + Content.from_function_call("call_1", "load_skill", arguments={"skill_name": "test-skill"}) + ], + ) + ], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Done")])], + ] + response_stream = agent.run("Use the skill", stream=True, session=agent.create_session()) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in response_stream] + response = await response_stream.get_final_response() + + for result in (response, AgentResponse.from_updates(updates), AgentResponse.from_json(response.to_json())): + item = _to_eval_item(query="Use the skill", response=result) + assert item.tools is not None + tools = {tool.name: tool for tool in item.tools} + expected = {"load_skill", "read_skill_resource", "run_skill_script", "iteration_1", "versioned"} + if auto_approve: + expected.add("iteration_2") + assert set(tools) == expected + assert tools["versioned"].description == f"Version {2 if auto_approve else 1}" + assert provider.before_run.await_count == (2 if auto_approve else 1) + assert any( + content.type == ("function_result" if auto_approve else "function_approval_request") + for message in response.messages + for content in message.contents + ) + + @pytest.mark.parametrize("empty_stream", [False, True]) + async def test_tool_free_stream_updates_preserve_empty_snapshot( + self, chat_client_base: Any, empty_stream: bool + ) -> None: + if empty_stream: + chat_client_base.streaming_responses = [[]] + agent = Agent(client=chat_client_base) + stream = agent.run("No tools", stream=True) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in stream] + final = await stream.get_final_response() + restored = AgentResponse.from_updates(updates) + agent.default_options["tools"] = [FunctionTool(name="unrelated", func=lambda: "")] + + assert updates + assert restored.text == final.text + assert len(restored.messages) == len(final.messages) + if empty_stream: + assert restored.messages == final.messages == [] + assert _to_eval_item(query="No tools", response=restored, agent=agent).tools is None + assert restored.additional_properties["agent_framework.evaluation"] == {} + + async def test_loop_aggregation_uses_latest_same_name_definition(self, chat_client_base: Any) -> None: + class ChangingToolsProvider(ContextProvider): + async def before_run( + self, + *, + agent: SupportsAgentRun, + session: AgentSession, + context: SessionContext, + state: dict[str, Any], + ) -> None: + iteration = state["iteration"] = state.get("iteration", 0) + 1 + context.extend_tools( + self.source_id, + [ + FunctionTool(name="search", description=f"Version {iteration}", func=lambda: ""), + FunctionTool(name=f"iteration_{iteration}", func=lambda: ""), + ], + ) + + agent = Agent( + client=chat_client_base, + context_providers=[ChangingToolsProvider("changing")], + middleware=[AgentLoopMiddleware(lambda iteration, **kwargs: iteration < 2)], + ) + response = await agent.run("Test", session=agent.create_session()) + for result in (response, AgentResponse.from_json(response.to_json())): + item = _to_eval_item(query="Test", response=result) + assert item.tools is not None + tools = {tool.name: tool for tool in item.tools} + assert set(tools) == {"search", "iteration_1", "iteration_2"} + assert tools["search"].description == "Version 2" + + @pytest.mark.parametrize("tool_free_final", [False, True]) + async def test_streaming_loop_updates_retain_previous_tool_definitions( + self, chat_client_base: Any, tool_free_final: bool + ) -> None: + class ChangingToolsProvider(ContextProvider): + async def before_run( + self, + *, + agent: SupportsAgentRun, + session: AgentSession, + context: SessionContext, + state: dict[str, Any], + ) -> None: + iteration = state["iteration"] = state.get("iteration", 0) + 1 + if tool_free_final and iteration == 2: + return + context.extend_tools( + self.source_id, + [ + FunctionTool(name=f"iteration_{iteration}", func=lambda: "Done"), + FunctionTool( + name="search", + description=f"Version {iteration}", + input_model={ + "type": "object", + "properties": {"value": {"type": "integer", "enum": [iteration]}}, + "required": ["value"], + }, + func=lambda value: str(value), + ), + ], + ) + + agent = Agent( + client=chat_client_base, + context_providers=[ChangingToolsProvider("changing")], + middleware=[AgentLoopMiddleware(lambda iteration, **kwargs: iteration < 2)], + ) + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_function_call("call_1", "iteration_1", arguments={})], + ) + ], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("First done")])], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Second done")])], + ] + stream = agent.run("Test", stream=True, session=agent.create_session()) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in stream] + final = await stream.get_final_response() + restored = AgentResponse.from_updates(updates) + item = _to_eval_item(query="Test", response=restored) + + assert item.tools is not None + tools = {tool.name: tool for tool in item.tools} + expected = {"search", "iteration_1"} + if not tool_free_final: + expected.add("iteration_2") + assert set(tools) == expected + version = 1 if tool_free_final else 2 + assert tools["search"].description == f"Version {version}" + assert tools["search"].parameters()["properties"]["value"]["enum"] == [version] + assert [ + content.name + for message in restored.messages + for content in message.contents + if content.type == "function_call" + ] == ["iteration_1"] + assert final.text == "Second done" + final_tools = _to_eval_item(query="Test", response=final).tools + assert {tool.name for tool in final_tools or []} == (set() if tool_free_final else {"search", "iteration_2"}) + + @pytest.mark.parametrize("return_final_only", [False, True]) + async def test_loop_response_retains_skill_tools(self, chat_client_base: Any, return_final_only: bool) -> None: + skill = InlineSkill( + frontmatter=SkillFrontmatter(name="test-skill", description="A test skill."), + instructions="Skill instructions.", + ) + agent = Agent( + client=chat_client_base, + context_providers=[SkillsProvider(skill)], + middleware=[ + AgentLoopMiddleware( + lambda iteration, **kwargs: iteration < 2, + return_final_only=return_final_only, + ) + ], + ) + response = await agent.run("Test") + + item = _to_eval_item(query="Test", response=response) + + assert item.tools is not None + assert {tool.name for tool in item.tools} == {"load_skill", "read_skill_resource", "run_skill_script"} + assert len(item.tools) == 3 + assert chat_client_base.call_count == 2 + restored_item = _to_eval_item(query="Test", response=AgentResponse.from_json(response.to_json())) + assert restored_item.tools is not None + assert {tool.name for tool in restored_item.tools} == {"load_skill", "read_skill_resource", "run_skill_script"} + + async def test_evaluate_agent_includes_skill_tools(self, chat_client_base: Any) -> None: + skill = InlineSkill( + frontmatter=SkillFrontmatter(name="test-skill", description="A test skill."), + instructions="Skill instructions.", + ) + agent = Agent(client=chat_client_base, context_providers=[SkillsProvider(skill)]) + + @evaluator + def has_skill_tools(tools: list[FunctionTool]) -> bool: + return "load_skill" in {tool.name for tool in tools} + + results = await evaluate_agent( + agent=agent, + queries="Test", + evaluators=LocalEvaluator(has_skill_tools), + ) + + assert results[0].passed == 1 + + async def test_evaluate_existing_response_includes_skill_tools(self, chat_client_base: Any) -> None: + skill = InlineSkill( + frontmatter=SkillFrontmatter(name="test-skill", description="A test skill."), + instructions="Skill instructions.", + ) + agent = Agent(client=chat_client_base, context_providers=[SkillsProvider(skill)]) + response = await agent.run("Test") + + @evaluator + def has_skill_tools(tools: list[FunctionTool]) -> bool: + return "load_skill" in {tool.name for tool in tools} + + results = await evaluate_agent( + queries="Test", + responses=response, + evaluators=LocalEvaluator(has_skill_tools), + ) + + assert results[0].passed == 1 + assert chat_client_base.call_count == 1 + + @pytest.mark.parametrize("serialization", ["json", "pickle"]) + async def test_evaluate_restored_response_discovers_skill_tools( + self, + chat_client_base: Any, + serialization: str, + ) -> None: + skill = InlineSkill( + frontmatter=SkillFrontmatter(name="test-skill", description="A test skill."), + instructions="Skill instructions.", + ) + provider = SkillsProvider(skill) + provider.before_run = AsyncMock(wraps=provider.before_run) # type: ignore[method-assign] + agent = Agent(client=chat_client_base, context_providers=[provider]) + response = await agent.run("Test") + restored = ( + AgentResponse.from_json(response.to_json()) + if serialization == "json" + else pickle.loads(pickle.dumps(response)) + ) + + @evaluator + def has_skill_tools(tools: list[FunctionTool]) -> bool: + return all(tool.declaration_only for tool in tools) and "load_skill" in {tool.name for tool in tools} + + results = await evaluate_agent( + queries="Test", + responses=restored, + evaluators=LocalEvaluator(has_skill_tools), + ) + + assert results[0].passed == 1 + assert chat_client_base.call_count == 1 + provider.before_run.assert_awaited_once() + + async def test_explicit_tools_override_fresh_run(self, chat_client_base: Any) -> None: + configured_tool = FunctionTool(name="configured", func=lambda: "") + explicit_tool = FunctionTool(name="explicit", func=lambda: "") + agent = Agent(client=chat_client_base, tools=[configured_tool]) + + @evaluator + def has_explicit_tools(tools: list[FunctionTool]) -> bool: + return tools == [explicit_tool] + + results = await evaluate_agent( + agent=agent, + queries="Test", + tools=[explicit_tool], + evaluators=LocalEvaluator(has_explicit_tools), + ) + + assert results[0].passed == 1 + + async def test_explicit_tools_for_restored_response(self) -> None: + response = AgentResponse.from_json(AgentResponse(messages=[Message("assistant", ["Done"])]).to_json()) + tool = FunctionTool(name="load_skill", func=lambda skill_name: skill_name) + + @evaluator + def has_explicit_tools(tools: list[FunctionTool]) -> bool: + return tools == [tool] + + results = await evaluate_agent( + queries="Test", + responses=response, + tools=[tool], + evaluators=LocalEvaluator(has_explicit_tools), + ) + + assert results[0].passed == 1 + class TestEvalItemSplitting: def test_split_messages_format(self) -> None: diff --git a/python/packages/core/tests/core/test_harness_loop.py b/python/packages/core/tests/core/test_harness_loop.py index 40c85e9ac19..1ad4663027f 100644 --- a/python/packages/core/tests/core/test_harness_loop.py +++ b/python/packages/core/tests/core/test_harness_loop.py @@ -339,6 +339,32 @@ async def test_non_streaming_return_final_only_returns_last_response() -> None: assert "first answer" not in response.text +@pytest.mark.parametrize("return_final_only", [False, True]) +async def test_non_streaming_loop_preserves_namespaced_metadata( + chat_client_base: MockBaseChatClient, return_final_only: bool +) -> None: + first_metadata = {"custom": {"first": 1, "shared": {"old": True}}, "status": "first"} + final_metadata = {"custom": {"last": 2, "shared": {"new": True}}, "status": "last"} + chat_client_base.run_responses = [ + ChatResponse(messages=[Message("assistant", ["First"])], additional_properties=first_metadata), + ChatResponse(messages=[Message("assistant", ["Last"])], additional_properties=final_metadata), + ] + agent = Agent( + client=chat_client_base, + middleware=[AgentLoopMiddleware(always_continue, max_iterations=2, return_final_only=return_final_only)], + ) + + response = await agent.run("Test") + + expected = {"last": 2, "shared": {"new": True}} + if not return_final_only: + expected["first"] = 1 + assert response.additional_properties["custom"] == expected + assert response.additional_properties["status"] == "last" + assert first_metadata == {"custom": {"first": 1, "shared": {"old": True}}, "status": "first"} + assert final_metadata == {"custom": {"last": 2, "shared": {"new": True}}, "status": "last"} + + # region feedback loop diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index a0d1c54bfa0..46ab508185a 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -20,6 +20,8 @@ from agent_framework import ( DEFAULT_TOOL_APPROVAL_SOURCE_ID, Agent, + AgentContext, + AgentMiddleware, AgentResponse, AgentResponseUpdate, AgentSession, @@ -470,7 +472,11 @@ async def capture_response( resumed_stream = agent.run(request.to_function_approval_response(True), stream=True, session=session) resumed_updates = [update async for update in resumed_stream] resumed = await resumed_stream.get_final_response() - assert [(update.role, [content.type for content in update.contents]) for update in resumed_updates] == [ + assert [ + (update.role, [content.type for content in update.contents]) + for update in resumed_updates + if update.contents + ] == [ ("tool", ["function_result"]), ("assistant", ["text"]), ] @@ -582,7 +588,9 @@ async def capture_response( ) stale_updates = [update async for update in stale_stream] stale = await stale_stream.get_final_response() - assert [(update.role, [content.type for content in update.contents]) for update in stale_updates] == [ + assert [ + (update.role, [content.type for content in update.contents]) for update in stale_updates if update.contents + ] == [ ("assistant", ["function_approval_request"]), ] else: @@ -623,7 +631,11 @@ async def capture_response( ) approved_updates = [update async for update in approved_stream] approved = await approved_stream.get_final_response() - assert [(update.role, [content.type for content in update.contents]) for update in approved_updates] == [ + assert [ + (update.role, [content.type for content in update.contents]) + for update in approved_updates + if update.contents + ] == [ ("tool", ["function_result"]), ("assistant", ["text"]), ] @@ -713,7 +725,9 @@ def identity_sink(value: str) -> str: ) stale_updates = [update async for update in stale_stream] stale = await stale_stream.get_final_response() - assert [(update.role, [content.type for content in update.contents]) for update in stale_updates] == [ + assert [ + (update.role, [content.type for content in update.contents]) for update in stale_updates if update.contents + ] == [ ("assistant", ["function_approval_request"]), ] else: @@ -737,7 +751,11 @@ def identity_sink(value: str) -> str: ) approved_updates = [update async for update in approved_stream] approved = await approved_stream.get_final_response() - assert [(update.role, [content.type for content in update.contents]) for update in approved_updates] == [ + assert [ + (update.role, [content.type for content in update.contents]) + for update in approved_updates + if update.contents + ] == [ ("tool", ["function_result"]), ("assistant", ["text"]), ] @@ -1203,7 +1221,7 @@ def guarded_tool() -> str: second_stream = agent.run(approval_message, stream=True, session=session) second_updates = [update async for update in second_stream] second_response = await second_stream.get_final_response() - assert [[content.type for content in update.contents] for update in second_updates] == [ + assert [[content.type for content in update.contents] for update in second_updates if update.contents] == [ ["function_result"], ["text"], ] @@ -2876,7 +2894,7 @@ def second_streamed_tool() -> str: final_updates = [update async for update in final_stream] final_response = await final_stream.get_final_response() - assert final_updates[-1].text == "done" + assert "".join(update.text for update in final_updates) == "done" assert final_response.text == "done" assert calls == 2 @@ -3270,6 +3288,57 @@ def echo(text: str) -> str: assert response.value.answer == "42" +async def test_streaming_filtered_approval_preserves_response_metadata( + chat_client_base: MockBaseChatClient, +) -> None: + class StampApprovalMetadata(AgentMiddleware): + async def process(self, context: AgentContext, call_next: Callable[[], Awaitable[None]]) -> None: + await call_next() + assert isinstance(context.result, ResponseStream) + + def stamp(update: AgentResponseUpdate) -> AgentResponseUpdate: + if any(content.type == "function_approval_request" for content in update.contents): + update.additional_properties = {"custom": {"approval": "retained"}} + return update + + context.result = context.result.with_transform_hook(stamp) + + executions: list[str] = [] + + @tool(approval_mode="always_require") + def guarded() -> str: + executions.append("called") + return "Done" + + agent = Agent( + client=chat_client_base, + tools=[guarded], + middleware=[ + ToolApprovalMiddleware(auto_approval_rules=[lambda function_call: True]), + StampApprovalMetadata(), + ], + ) + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_function_call("call_1", "guarded", arguments={})], + ) + ], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Finished")])], + ] + + stream = agent.run("Test", stream=True, session=agent.create_session()) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in stream] + final = await stream.get_final_response() + + for response in (final, AgentResponse.from_updates(updates)): + assert response.additional_properties["custom"] == {"approval": "retained"} + assert not _approval_requests(response.messages) + assert all(message.contents for message in response.messages) + assert executions == ["called"] + + async def test_streaming_auto_approved_tool_preserves_value_when_preamble_text_coalesces( chat_client_base: MockBaseChatClient, ) -> None: diff --git a/python/packages/core/tests/core/test_middleware_context_result.py b/python/packages/core/tests/core/test_middleware_context_result.py index 75ebc9e2204..3ddf51e21a0 100644 --- a/python/packages/core/tests/core/test_middleware_context_result.py +++ b/python/packages/core/tests/core/test_middleware_context_result.py @@ -206,7 +206,7 @@ async def process(self, context: AgentContext, call_next: Callable[[], Awaitable async for update in agent.run(normal_messages, stream=True): normal_updates.append(update) - assert len(normal_updates) == 2 + assert len([update for update in normal_updates if update.contents]) == 2 assert normal_updates[0].text == "test streaming response " assert normal_updates[1].text == "another update" diff --git a/python/packages/core/tests/core/test_middleware_with_agent.py b/python/packages/core/tests/core/test_middleware_with_agent.py index d04767c904f..92906ac0c43 100644 --- a/python/packages/core/tests/core/test_middleware_with_agent.py +++ b/python/packages/core/tests/core/test_middleware_with_agent.py @@ -350,7 +350,7 @@ async def process(self, context: AgentContext, call_next: Callable[[], Awaitable updates.append(update) # Verify streaming response - assert len(updates) == 2 + assert len([update for update in updates if update.contents]) == 2 assert updates[0].text == "Streaming" assert updates[1].text == " response" assert client.call_count == 1 @@ -1425,7 +1425,7 @@ async def process(self, context: AgentContext, call_next: Callable[[], Awaitable updates.append(update) # Verify streaming responsecod - assert len(updates) == 2 + assert len([update for update in updates if update.contents]) == 2 assert updates[0].text == "Stream" assert updates[1].text == " response" diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index fe4f5dc11c5..58c17ca1f5a 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -2262,6 +2262,122 @@ def test_agent_run_response_from_updates(agent_response_update: AgentResponseUpd assert response.text == "Test contentTest content" +@mark.parametrize( + ("response_type", "update_type"), + [(ChatResponse, ChatResponseUpdate), (AgentResponse, AgentResponseUpdate)], +) +def test_response_only_update_does_not_create_message(response_type: Any, update_type: Any) -> None: + response = response_type() + raw = object() + update = update_type( + response_id="response-1", + created_at="2026-10-21T00:00:00Z", + finish_reason="stop", + additional_properties={"provider": {"key": "value"}}, + continuation_token={"token": "next"}, + raw_representation=raw, + ) + _set_operation_state(update, "in_progress") + if isinstance(update, ChatResponseUpdate): + update.conversation_id = "conversation-1" + update.model = "model-1" + else: + update.agent_id = "agent-1" + + _process_update(response, update) + + assert response.messages == [] + assert response.response_id == "response-1" + assert response.created_at == update.created_at + assert response.finish_reason == "stop" + assert response.additional_properties == {"provider": {"key": "value"}} + assert response.continuation_token == {"token": "next"} + assert response.raw_representation == [raw] + assert _get_operation_state(response) == "in_progress" + if isinstance(response, ChatResponse): + assert response.conversation_id == "conversation-1" + assert response.model == "model-1" + else: + assert response.agent_id == "agent-1" + + terminal = update_type() + _set_operation_state(terminal, "terminal") + _process_update(response, terminal) + assert response.messages == [] + assert response.continuation_token is None + assert _get_operation_state(response) == "terminal" + + +@mark.parametrize( + ("response_type", "update_type"), + [(ChatResponse, ChatResponseUpdate), (AgentResponse, AgentResponseUpdate)], +) +@mark.parametrize("with_content", [False, True]) +@mark.parametrize("from_generator", [False, True]) +async def test_response_only_updates_preserve_reconstructed_messages( + response_type: Any, update_type: Any, with_content: bool, from_generator: bool +) -> None: + updates = [update_type(additional_properties={"provider": {"before": True}})] + if with_content: + updates.extend([ + update_type(contents=[Content.from_text("Hello ")], role="assistant", message_id="m1"), + update_type(additional_properties={"provider": {"between": True}}), + update_type(contents=[Content.from_text("world")], message_id="m1"), + ]) + updates.append(update_type(additional_properties={"provider": {"after": True}}, finish_reason="stop")) + + async def generate() -> AsyncIterator[Any]: + for update in updates: + yield update + + if from_generator: + response = await response_type.from_update_generator(generate()) + else: + response = response_type.from_updates(updates) + + assert response.additional_properties == { + "provider": {"before": True, "after": True, **({"between": True} if with_content else {})} + } + assert response.finish_reason == "stop" + assert len(response.messages) == int(with_content) + if with_content: + assert response.messages[0].text == "Hello world" + assert response.messages[0].role == "assistant" + assert response.messages[0].message_id == "m1" + + +@mark.parametrize( + ("response_type", "update_type"), + [(ChatResponse, ChatResponseUpdate), (AgentResponse, AgentResponseUpdate)], +) +@mark.parametrize("header", [{"role": "tool"}, {"author_name": "writer"}, {"message_id": "m1"}]) +def test_empty_message_header_is_preserved(response_type: Any, update_type: Any, header: dict[str, str]) -> None: + response = response_type() + _process_update(response, update_type(**header)) + assert len(response.messages) == 1 + for key, value in header.items(): + assert getattr(response.messages[0], key) == value + + +@mark.parametrize( + ("response_type", "update_type"), + [(ChatResponse, ChatResponseUpdate), (AgentResponse, AgentResponseUpdate)], +) +def test_empty_replacement_still_clears_message(response_type: Any, update_type: Any) -> None: + response = response_type(messages=[Message("assistant", ["old"], message_id="m1")]) + _process_update(response, update_type(merge_mode="replace")) + assert len(response.messages) == 1 + assert response.messages[0].contents == [] + assert response.messages[0].message_id == "m1" + assert ( + response_type.from_updates([ + update_type(contents=[Content.from_text("old")], message_id="m1"), + update_type(merge_mode="replace"), + ]).messages + == [] + ) + + @pytest.mark.parametrize( ("response_type", "update_type"), [(ChatResponse, ChatResponseUpdate), (AgentResponse, AgentResponseUpdate)], diff --git a/python/packages/foundry/tests/foundry/test_foundry_agent.py b/python/packages/foundry/tests/foundry/test_foundry_agent.py index 5bba9eb78c0..631d9924c07 100644 --- a/python/packages/foundry/tests/foundry/test_foundry_agent.py +++ b/python/packages/foundry/tests/foundry/test_foundry_agent.py @@ -1388,7 +1388,7 @@ async def stream() -> AsyncIterator[ChatResponseUpdate]: assert len(observed_options) == 3 assert all("agent_session_id" not in options.get("extra_body", {}) for options in observed_options) assert all(options.get("conversation_id") is None for options in observed_options) - assert len(updates) == 3 + assert len([update for update in updates if update.contents]) == 3 expected_state = {"ordinary": "parent-value"} if parent_handle is not None: expected_state[FOUNDRY_HOSTED_AGENT_SESSION_ID_KEY] = parent_handle