diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index 813e53ce6e..05486f1d28 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 639fc223b2..ef1bb5ab6c 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 bcd2ddeda2..7a418c6bdb 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 0052716eaa..f7e572cb81 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -389,6 +389,14 @@ 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. `BaseChatClient` snapshots accumulate only observed outgoing model-call tools after chat middleware; + other clients fall back to prepared tool configuration. Same-name input-schema changes retain a `schema_changed` + marker through capture and aggregation. + Automatic evaluation rejects called tools with that marker rather than applying the latest schema to earlier calls; + description-only changes and uncalled tools do not block evaluation. + `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 a6e7ccddf8..38e1064d18 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,8 @@ ResponseStream, _append_instructions, # pyright: ignore[reportPrivateUsage] _build_agent_response_from_chat_response, # pyright: ignore[reportPrivateUsage] + _capture_evaluation_tools, # pyright: ignore[reportPrivateUsage] + _copy_operation_state, # pyright: ignore[reportPrivateUsage] map_chat_to_agent_update, normalize_messages, ) @@ -275,6 +278,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 +400,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 +1411,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 +1448,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 +1489,28 @@ 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() + metadata_update = AgentResponseUpdate( + continuation_token=response.continuation_token, + additional_properties={ + _EVALUATION_TOOLS_METADATA_KEY: deepcopy( + response.additional_properties[_EVALUATION_TOOLS_METADATA_KEY] + ) + }, + ) + _copy_operation_state(response, metadata_update) + yield metadata_update + + 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 +1788,12 @@ async def _prepare_run_context( per_service_call_history_middleware, ] + evaluation_tools: dict[str, dict[str, Any]] = {} + if isinstance(self.client, BaseChatClient): + effective_client_kwargs["_evaluation_tool_capture"] = partial(_capture_evaluation_tools, evaluation_tools) + else: + _capture_evaluation_tools(evaluation_tools, co.get("tools")) + return { "session": active_session, "session_context": session_context, @@ -1768,6 +1802,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 d353b0a552..e02b227b9c 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -581,6 +581,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 1c57b9f72d..8ee060870c 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,44 @@ 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: + called_tools = { + content.name for message in all_msgs for content in message.contents if content.type == "function_call" + } + 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." + ) + schema_changed = definition.get("schema_changed", False) + if not isinstance(schema_changed, bool): + raise ValueError( + f"Invalid agent_framework.evaluation definition for {name!r}: schema_changed must be bool." + ) + if schema_changed and name in called_tools: + raise ValueError( + f"Tool {name!r} changed input schema during the captured conversation; " + "automatic evaluation cannot represent multiple schemas for the same called tool. " + "Use distinct tool names or supply an explicit tools= override." + ) + 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 +1644,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 +1662,16 @@ 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. + + Automatic discovery rejects a called tool whose captured input schema changed + under the same name, even if the schemas might be compatible. Use distinct + names for different schemas or supply an explicit ``tools`` override. Args: agent: An agent-framework agent instance. @@ -1642,6 +1691,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 +1798,7 @@ async def evaluate_agent( query=q, response=r, agent=agent, + tools=tools, context=context, ) ) @@ -1766,6 +1819,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 0a14782433..6ab54d6dae 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 2232093ea3..8a6d22606d 100644 --- a/python/packages/core/agent_framework/_harness/_tool_approval.py +++ b/python/packages/core/agent_framework/_harness/_tool_approval.py @@ -562,6 +562,14 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]: remaining_update = copy.copy(update) remaining_update.contents = remaining_contents yield remaining_update + elif update.additional_properties: + metadata_update = copy.copy(update) + metadata_update.contents = [] + metadata_update.role = None + metadata_update.author_name = None + metadata_update.message_id = None + metadata_update.merge_mode = None + yield metadata_update 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 b7febaf425..9338032cb7 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,37 @@ from ._tools import ToolTypes +def _capture_evaluation_tools( # pyright: ignore[reportUnusedFunction] + captured: dict[str, dict[str, Any]], tools: Any +) -> None: + """Retain exposed declarations and flag same-name schema changes.""" + from ._tools import FunctionTool, normalize_tools + + for tool in normalize_tools(tools or []): + if not isinstance(tool, FunctionTool): + continue + captured[tool.name] = _merge_evaluation_tool_definition( + captured.get(tool.name), + { + "description": tool.description, + "input_schema": json.loads(json.dumps(tool.parameters(), allow_nan=False)), + }, + ) + + +def _merge_evaluation_tool_definition( + previous: Mapping[str, Any] | None, incoming: Mapping[str, Any] +) -> dict[str, Any]: + definition = dict(incoming) + if previous is not None and ( + previous.get("schema_changed") is True + or json.dumps(previous.get("input_schema"), sort_keys=True, allow_nan=False) + != json.dumps(incoming.get("input_schema"), sort_keys=True, allow_nan=False) + ): + definition["schema_changed"] = True + return definition + + # region Content Parsing Utilities @@ -618,7 +650,17 @@ def merge_additional_properties( for key, value in (second or {}).items(): current = merged.get(key) if isinstance(current, Mapping) and isinstance(value, Mapping): - merged[key] = {**current, **value} + current = cast("Mapping[str, Any]", current) + value = cast("Mapping[str, Any]", value) + namespace = {**current, **value} + if key == _EVALUATION_TOOLS_METADATA_KEY: + for name, definition in value.items(): + previous = current.get(name) + if isinstance(previous, Mapping) and isinstance(definition, Mapping): + namespace[name] = _merge_evaluation_tool_definition( + cast("Mapping[str, Any]", previous), cast("Mapping[str, Any]", definition) + ) + merged[key] = namespace else: merged[key] = dict(cast("Mapping[str, Any]", value)) if isinstance(value, Mapping) else value return merged @@ -2546,63 +2588,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 @@ -3617,6 +3666,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 8d6ec0033a..12cfbdd744 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -1599,7 +1599,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 @@ -1688,7 +1691,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 178528510a..46743d5a9c 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 f32e216bd9..98cafa95aa 100644 --- a/python/packages/core/tests/core/test_evaluation.py +++ b/python/packages/core/tests/core/test_evaluation.py @@ -4,15 +4,55 @@ from __future__ import annotations -from typing import Any, cast -from unittest.mock import MagicMock +import asyncio +import pickle +from collections.abc import Awaitable, Callable +from copy import deepcopy +from typing import Any, Literal, cast +from unittest.mock import AsyncMock, MagicMock, patch 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 ChatContext, 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, + _get_operation_state, + _set_operation_state, +) + + +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 +169,995 @@ 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 AgentResponse.from_json(response.to_json()).messages[0].contents[0].name == "load_skill" + restored = pickle.loads(pickle.dumps(response)) + 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) + + @pytest.mark.parametrize("stream", [False, True]) + @pytest.mark.parametrize("replace_tool", [False, True]) + async def test_chat_middleware_exposure_overrides_prepared_tools( + self, chat_client_base: Any, stream: bool, replace_tool: bool + ) -> None: + original = FunctionTool(name="search", input_model={"type": "object"}, func=lambda **kwargs: "Seattle") + replacement = FunctionTool( + name="search", + input_model={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + func=lambda city: city, + ) + exposed_tools = [replacement] if replace_tool else [] + + async def change_tools(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + context.options = {**(context.options or {}), "tools": exposed_tools} + await call_next() + + agent = Agent(client=chat_client_base, tools=[original], middleware=[change_tools]) + contents = [Content.from_text("Done")] + if replace_tool: + contents.insert(0, Content.from_function_call("call_1", "search", arguments={"city": "Seattle"})) + chat_client_base.run_responses = [ChatResponse(messages=[Message("assistant", contents)])] + chat_client_base.streaming_responses = [[ChatResponseUpdate(role="assistant", contents=contents)]] + reconstructed_responses: list[AgentResponse[Any]] = [] + with patch.object(chat_client_base, "_inner_get_response", wraps=chat_client_base._inner_get_response) as model: + if stream: + 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() + reconstructed_responses.append(AgentResponse.from_updates(updates)) + else: + response = await agent.run("Test") + assert model.call_count >= 1 + assert all(call.kwargs["options"]["tools"] == exposed_tools for call in model.call_args_list) + + for result in ( + response, + *reconstructed_responses, + AgentResponse.from_dict(response.to_dict()), + AgentResponse.from_json(response.to_json()), + pickle.loads(pickle.dumps(response)), + ): + item = _to_eval_item(query="Test", response=result, agent=agent) + if replace_tool: + assert item.tools is not None + assert [(tool.name, tool.parameters()) for tool in item.tools] == [("search", replacement.parameters())] + else: + assert item.tools is None + + @pytest.mark.parametrize("stream", [False, True]) + async def test_chat_middleware_removal_retains_previously_exposed_tools( + self, chat_client_base: Any, stream: bool + ) -> None: + search = FunctionTool(name="search", func=lambda: "Seattle") + exposed_tools: list[list[Any]] = [] + + async def remove_after_first_call(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + if exposed_tools: + context.options = {**(context.options or {}), "tools": []} + exposed_tools.append(list((context.options or {}).get("tools") or [])) + await call_next() + + call = Content.from_function_call("call_1", "search", arguments={}) + chat_client_base.run_responses = [ + ChatResponse(messages=[Message("assistant", [call])]), + ChatResponse(messages=[Message("assistant", ["Done"])]), + ] + chat_client_base.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[call])], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Done")])], + ] + agent = Agent(client=chat_client_base, tools=[search], middleware=[remove_after_first_call]) + if stream: + response_stream = agent.run("Test", stream=True) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in response_stream] + response = AgentResponse.from_updates(updates) + else: + response = await agent.run("Test") + + assert exposed_tools == [[search], []] + for result in (response, AgentResponse.from_json(response.to_json())): + item = _to_eval_item(query="Test", response=result) + assert item.tools is not None + assert [tool.name for tool in item.tools] == ["search"] + + @pytest.mark.parametrize("stream", [False, True]) + async def test_custom_client_retains_prepared_tool_fallback(self, client: Any, stream: bool) -> None: + search = FunctionTool(name="search", func=lambda: "Seattle") + agent = Agent(client=client, tools=[search]) + if stream: + response_stream = agent.run("Test", stream=True) + updates = [AgentResponseUpdate.from_json(update.to_json()) async for update in response_stream] + response = AgentResponse.from_updates(updates) + else: + response = await agent.run("Test") + + for result in (response, AgentResponse.from_json(response.to_json())): + item = _to_eval_item(query="Test", response=result) + assert item.tools is not None + assert [(tool.name, tool.parameters()) for tool in item.tools] == [("search", search.parameters())] + + 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_rejects_ambiguous_evaluation( + 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"}), + ] + reconstructed_responses: list[AgentResponse[Any]] = [] + 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() + reconstructed_responses.append(AgentResponse.from_updates(updates)) + else: + chat_client_base.run_responses = [ChatResponse(messages=[Message("assistant", [call])]) for call in calls] + response = await agent.run("Test") + + for result in ( + response, + *reconstructed_responses, + AgentResponse.from_dict(response.to_dict()), + AgentResponse.from_json(response.to_json()), + pickle.loads(pickle.dumps(response)), + ): + with pytest.raises(ValueError, match="search.*changed input schema"): + _to_eval_item(query="Test", response=result) + assert _to_eval_item(query="Test", response=result, tools=[replacement]).tools == [replacement] + assert _to_eval_item(query="Test", response=result, tools=[]).tools is None + assert any( + content.type == "function_result" and content.result == "Seattle" + for message in response.messages + for content in message.contents + ) + + @pytest.mark.parametrize("constraint", ["const", "enum"]) + async def test_boolean_numeric_schema_change_rejects_evaluation( + self, chat_client_base: Any, constraint: str + ) -> None: + def schema(value: bool | int) -> dict[str, Any]: + return { + "type": "object", + "properties": {"value": {constraint: [value] if constraint == "enum" else value}}, + "required": ["value"], + } + + replacement = FunctionTool(name="search", input_model=schema(1), func=lambda value: str(value)) + + def replace(value: bool, ctx: FunctionInvocationContext) -> str: + ctx.remove_tools(["search"]) + ctx.add_tools([replacement]) + return "Replaced" + + original = FunctionTool(name="search", input_model=schema(True), func=replace) + chat_client_base.run_responses = [ + ChatResponse( + messages=[ + Message("assistant", [Content.from_function_call("call_1", "search", arguments={"value": True})]) + ] + ), + ChatResponse( + messages=[ + Message("assistant", [Content.from_function_call("call_2", "search", arguments={"value": 1})]) + ] + ), + ] + response = await Agent(client=chat_client_base, tools=[original]).run("Test") + + for result in (response, AgentResponse.from_json(response.to_json())): + with pytest.raises(ValueError, match="search.*changed input schema"): + _to_eval_item(query="Test", response=result) + + @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): + if replace_same_name: + with pytest.raises(ValueError, match="load.*changed input schema"): + _to_eval_item(query="Test", response=response) + continue + 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", "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 ( + 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()}}}, + {"weather": {"description": "", "input_schema": {}, "schema_changed": "yes"}}, + ], + ) + 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] + + +class TestEvaluateAgentTools: + @pytest.mark.parametrize("with_approval_middleware", [False, True]) + @pytest.mark.parametrize("operation_state", [None, "in_progress", "terminal"]) + async def test_stream_tool_metadata_preserves_continuation( + self, + chat_client_base: Any, + with_approval_middleware: bool, + operation_state: Literal["in_progress", "terminal"] | None, + ) -> None: + token = {"response_id": "response_1"} + update = ChatResponseUpdate( + role="assistant", + contents=[Content.from_text("Working")], + continuation_token=cast(Any, token), + ) + _set_operation_state(update, "in_progress" if operation_state == "terminal" else operation_state) + provider_updates = [update] + if operation_state == "terminal": + terminal_update = ChatResponseUpdate(finish_reason="stop") + _set_operation_state(terminal_update, "terminal") + provider_updates.append(terminal_update) + chat_client_base.streaming_responses = [provider_updates] + + if with_approval_middleware: + agent = create_harness_agent( + chat_client_base, + disable_compaction=True, + disable_todo=True, + disable_mode=True, + disable_file_memory=True, + disable_web_search=True, + ) + else: + agent = Agent(client=chat_client_base) + + stream = agent.run("Test", stream=True, session=agent.create_session()) + updates = [update async for update in stream] + final = await stream.get_final_response() + restored_updates = [AgentResponseUpdate.from_json(update.to_json()) for update in updates] + expected_token = None if operation_state == "terminal" else token + + for result in (final, AgentResponse.from_updates(updates), AgentResponse.from_updates(restored_updates)): + assert result.continuation_token == expected_token + assert _get_operation_state(result) == operation_state + assert result.text == "Working" + assert len(result.messages) == 1 + assert result.additional_properties["agent_framework.evaluation"] == {} + + @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_description(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("stream", [False, True]) + async def test_loop_rejects_called_tool_with_changing_schema(self, chat_client_base: Any, stream: 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 + context.extend_tools( + self.source_id, + [ + FunctionTool( + name="search", + input_model={ + "type": "object", + "properties": {"value": {"type": "integer", "enum": [min(iteration, 2)]}}, + "required": ["value"], + }, + func=lambda value: str(value), + ) + ], + ) + + calls = [ + Content.from_function_call(f"call_{iteration}", "search", arguments={"value": min(iteration, 2)}) + for iteration in (1, 2, 3) + ] + agent = Agent( + client=chat_client_base, + context_providers=[ChangingToolsProvider("changing")], + middleware=[AgentLoopMiddleware(lambda iteration, **kwargs: iteration < 3)], + ) + if stream: + chat_client_base.streaming_responses = [ + updates + for call in calls + for updates in ( + [ChatResponseUpdate(role="assistant", contents=[call])], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Done")])], + ) + ] + response_stream = agent.run("Test", stream=True, session=agent.create_session()) + updates = [update async for update in response_stream] + response = AgentResponse.from_updates(updates) + else: + chat_client_base.run_responses = [ + result + for call in calls + for result in ( + ChatResponse(messages=[Message("assistant", [call])]), + ChatResponse(messages=[Message("assistant", ["Done"])]), + ) + ] + response = await agent.run("Test", session=agent.create_session()) + + for result in (response, AgentResponse.from_json(response.to_json())): + assert [ + content.arguments + for message in result.messages + for content in message.contents + if content.type == "function_call" + ] == [{"value": 1}, {"value": 2}, {"value": 2}] + with pytest.raises(ValueError, match="search.*changed input schema"): + _to_eval_item(query="Test", response=result) + + @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 + + @pytest.mark.parametrize("serialization", [None, "json", "pickle"]) + async def test_evaluate_existing_response_discovers_skill_tools( + self, + chat_client_base: Any, + serialization: str | None, + ) -> 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") + if serialization == "json": + response = AgentResponse.from_json(response.to_json()) + elif serialization == "pickle": + response = 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=response, + 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 40c85e9ac1..1ad4663027 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 a0d1c54bfa..d2170fbff4 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -11,7 +11,7 @@ from datetime import timedelta from enum import Enum from pathlib import Path -from typing import Any +from typing import Any, cast from uuid import UUID import pytest @@ -20,6 +20,8 @@ from agent_framework import ( DEFAULT_TOOL_APPROVAL_SOURCE_ID, Agent, + AgentContext, + AgentMiddleware, AgentResponse, AgentResponseUpdate, AgentSession, @@ -44,6 +46,7 @@ tool, ) from agent_framework._feature_stage import ExperimentalWarning +from agent_framework._types import _get_operation_state, _set_operation_state from agent_framework.security import ( PRINCIPAL_METADATA_KEY, ConfidentialityLabel, @@ -470,7 +473,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 +589,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 +632,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 +726,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 +752,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 +1222,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 +2895,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 +3289,89 @@ 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"] + + +@pytest.mark.parametrize("terminal", [False, True]) +async def test_streaming_buffered_hosted_approval_preserves_continuation( + chat_client_base: MockBaseChatClient, terminal: bool +) -> None: + chat_client_base.function_invocation_configuration["enabled"] = False + call = Content.from_function_call("call_1", "hosted", arguments={}, hosted=True) + update = ChatResponseUpdate( + role="assistant", + contents=[Content.from_function_approval_request("approval_1", call)], + continuation_token=cast(Any, {"resume": "token"}), + ) + _set_operation_state(update, "in_progress") + provider_updates = [update] + if terminal: + terminal_update = ChatResponseUpdate(finish_reason="stop") + _set_operation_state(terminal_update, "terminal") + provider_updates.append(terminal_update) + chat_client_base.streaming_responses = [provider_updates] + agent = Agent(client=chat_client_base, middleware=[ToolApprovalMiddleware()]) + + stream = agent.run("Test", stream=True, session=agent.create_session()) + updates = [update async for update in stream] + final = await stream.get_final_response() + restored_updates = [AgentResponseUpdate.from_json(update.to_json()) for update in updates] + + for response in (final, AgentResponse.from_updates(updates), AgentResponse.from_updates(restored_updates)): + assert response.continuation_token == (None if terminal else {"resume": "token"}) + assert _get_operation_state(response) == ("terminal" if terminal else "in_progress") + assert len(_approval_requests(response.messages)) == 1 + assert all(message.contents for message in response.messages) + + 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 75ebc9e220..3ddf51e21a 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 d04767c904..92906ac0c4 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 626353676e..d047c92bed 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -2269,6 +2269,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 d7cb0913d8..a547f7dd57 100644 --- a/python/packages/foundry/tests/foundry/test_foundry_agent.py +++ b/python/packages/foundry/tests/foundry/test_foundry_agent.py @@ -1448,7 +1448,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 diff --git a/skills/agent-framework-builder-python/references/evaluation.md b/skills/agent-framework-builder-python/references/evaluation.md index 3b50930074..7b8af0f593 100644 --- a/skills/agent-framework-builder-python/references/evaluation.md +++ b/skills/agent-framework-builder-python/references/evaluation.md @@ -6,6 +6,17 @@ behavior is acceptable; observability explains what happened during execution. Evaluation APIs are experimental. Check current feature maturity and API docstrings before making them a long-term application contract. +Agent responses retain schema-only snapshots of tools exposed during execution, including context-provider tools. +These snapshots survive response serialization. For `BaseChatClient` clients, only tools sent to model calls after +chat middleware are captured; tools exposed in earlier calls remain included even if later removed. Other clients +fall back to prepared tool configuration. `evaluate_agent(tools=...)` overrides automatic discovery for evaluation +only; it does not change agent execution, and `tools=[]` disables tool definitions. + +Automatic discovery rejects a called tool whose input schema changed under the same name during the captured run or +aggregated conversation, even if the schemas might be compatible. Use distinct names for different schemas, evaluate +the segments separately, or explicitly supply `tools=` when you can choose appropriate definitions. Agent execution +itself is unchanged. + ## Capture the case Record the input, current and expected output, expected tool calls, relevant session or workflow state, provider/model,