Python: fix tools from context providers

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

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

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(

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
 

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,

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(

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,
                     )
                 )

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,

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 = []