From 3154e911a970f589fba2dbaf473ca58ef33004f1 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Fri, 10 Jul 2026 15:11:00 +0900 Subject: [PATCH 1/3] Python: emit participant tool calls in AG-UI workflows Decisions: - Pass function call, function result, and approval request content from streaming agent updates regardless of role. - Preserve the assistant-role gate for text and reuse the shared AG-UI content emitters without dual custom-event emission. Files changed: - packages/ag-ui/agent_framework_ag_ui/_workflow_run.py - packages/ag-ui/tests/ag_ui/test_workflow_run.py Verification: - uv run poe test -P ag-ui - uv run poe pyright -P ag-ui - uv run poe test-typing -P ag-ui - uv run poe syntax -P ag-ui -C Notes: - Existing workflow golden scenarios do not exercise participant tool calls, so no snapshot changed. - No blockers. --- .../agent_framework_ag_ui/_workflow_run.py | 11 ++- .../ag-ui/tests/ag_ui/test_workflow_run.py | 94 +++++++++++++++++++ 2 files changed, 100 insertions(+), 5 deletions(-) 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 921e9f3d11f..1b4fdc75865 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 @@ -62,6 +62,7 @@ } _INTERRUPT_CARD_EVENT_NAME = "WorkflowInterruptEvent" +_FUNCTION_CONTENT_TYPES = {"function_call", "function_result", "function_approval_request"} def _json_schema_for_response_type(response_type: Any) -> dict[str, Any] | None: @@ -675,16 +676,16 @@ def _workflow_payload_to_contents(payload: Any) -> list[Content] | None: return None return list(payload.contents or []) if isinstance(payload, AgentResponseUpdate): + contents = list(payload.contents or []) role_field = payload.role - if role_field is None: - return None if isinstance(role_field, str): role = role_field else: role = str(getattr(role_field, "value", role_field)) - if role != "assistant": - return None - return list(payload.contents or []) + if role == "assistant": + return contents + function_contents = [content for content in contents if content.type in _FUNCTION_CONTENT_TYPES] + return function_contents or None if isinstance(payload, AgentResponse): return _latest_assistant_contents(list(payload.messages or [])) if isinstance(payload, list): 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 0eb50938169..895f1e6e825 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 @@ -3,14 +3,17 @@ """Tests for native workflow AG-UI runner.""" import json +from collections.abc import AsyncIterator from enum import Enum from types import SimpleNamespace from typing import Any, cast from ag_ui.core import EventType, StateSnapshotEvent from agent_framework import ( + Agent, AgentResponse, AgentResponseUpdate, + ChatResponseUpdate, Content, Executor, Message, @@ -20,7 +23,9 @@ executor, handler, response_handler, + tool, ) +from conftest import StreamingChatClientStub # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports] from agent_framework_ag_ui._workflow_run import ( _coerce_content, @@ -395,6 +400,66 @@ async def structured(message: Any, ctx: WorkflowContext[Any, dict[str, int]]) -> assert output_custom[0].value == {"count": 3} # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] +async def test_workflow_participant_tool_call_emits_standard_tool_events() -> None: + """Participant tool calls should use the standard AG-UI tool event lifecycle.""" + + @tool + def get_weather(city: str) -> str: + return f"Sunny in {city}" + + invocation = 0 + + async def scripted_stream(messages: Any, options: Any, **kwargs: Any) -> AsyncIterator[ChatResponseUpdate]: + nonlocal invocation + del messages, options, kwargs + if invocation == 0: + yield ChatResponseUpdate( + contents=[ + Content.from_function_call( + call_id="weather-call", + name="get_weather", + arguments={"city": "Seattle"}, + ) + ], + role=None, + ) + else: + yield ChatResponseUpdate(contents=[Content.from_text("The weather is sunny.")], role="assistant") + invocation += 1 + + participant = Agent( + client=StreamingChatClientStub(scripted_stream), + name="weather-agent", + tools=[get_weather], + ) + workflow = WorkflowBuilder(start_executor=participant, output_from="all").build() + + events = [ + event + async for event in run_workflow_stream( + {"messages": [{"role": "user", "content": "What is the weather in Seattle?"}]}, + workflow, + ) + ] + + tool_events = [ + event + for event in events + if event.type in {"TOOL_CALL_START", "TOOL_CALL_ARGS", "TOOL_CALL_RESULT", "TOOL_CALL_END"} + ] + assert [event.type for event in tool_events] == [ + "TOOL_CALL_START", + "TOOL_CALL_ARGS", + "TOOL_CALL_END", + "TOOL_CALL_RESULT", + ] + assert tool_events[0].tool_call_name == "get_weather" # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + assert all(event.tool_call_id == "weather-call" for event in tool_events) # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + assert not [ + event for event in events if event.type == "CUSTOM" and getattr(event, "name", None) == "workflow_output" + ] + + async def test_workflow_run_passthroughs_ag_ui_base_events(): """Workflow outputs that are AG-UI BaseEvent instances should be emitted directly.""" @@ -1132,6 +1197,35 @@ def test_agent_response_update_none_role(self): update = AgentResponseUpdate(contents=[Content.from_text(text="hi")], role=None) assert _workflow_payload_to_contents(update) is None + def test_agent_response_update_function_call_without_role(self) -> None: + """Function call content passes through without role metadata.""" + function_call = Content.from_function_call(call_id="call-1", name="search", arguments={"query": "weather"}) + update = AgentResponseUpdate(contents=[function_call], role=None) + + assert _workflow_payload_to_contents(update) == [function_call] + + def test_agent_response_update_function_result_with_tool_role(self) -> None: + """Function result content passes through with the tool role.""" + function_result = Content.from_function_result(call_id="call-1", result={"temperature": 72}) + update = AgentResponseUpdate(contents=[function_result], role="tool") + + assert _workflow_payload_to_contents(update) == [function_result] + + def test_agent_response_update_approval_request_without_role(self) -> None: + """Approval request content passes through without role metadata.""" + function_call = Content.from_function_call(call_id="call-1", name="search", arguments={"query": "weather"}) + approval_request = Content.from_function_approval_request(id="approval-1", function_call=function_call) + update = AgentResponseUpdate(contents=[approval_request], role=None) + + assert _workflow_payload_to_contents(update) == [approval_request] + + def test_agent_response_update_assistant_text(self) -> None: + """Assistant text content continues to pass through.""" + text = Content.from_text(text="hi") + update = AgentResponseUpdate(contents=[text], role="assistant") + + assert _workflow_payload_to_contents(update) == [text] + def test_list_with_none_item(self): """List containing None causes None return.""" result = _workflow_payload_to_contents([Content.from_text(text="hi"), None]) From 309b702632a0c68ddbada246876e80548fd9c7e9 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Fri, 10 Jul 2026 15:13:09 +0900 Subject: [PATCH 2/3] Python: guard participant tool call duplication Decisions: - Assert the workflow stream emits one TOOL_CALL_START when a streamed call is also present in final conversation history. - Keep production flow unchanged because latest-assistant final-response conversion prevents duplication. Files changed: - packages/ag-ui/tests/ag_ui/test_workflow_run.py Verification: - uv run pytest packages/ag-ui/tests/ag_ui/test_workflow_run.py -k 'participant_tool_call or repeat_tool_call' -q - uv run poe test -P ag-ui - uv run poe pyright -P ag-ui - uv run poe test-typing -P ag-ui - uv run poe syntax -P ag-ui -C - git diff --check Notes: - No blockers; no call-id guard was required. --- .../ag-ui/tests/ag_ui/test_workflow_run.py | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) 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 895f1e6e825..e6976680bc8 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 @@ -460,6 +460,44 @@ async def scripted_stream(messages: Any, options: Any, **kwargs: Any) -> AsyncIt ] +async def test_workflow_stream_does_not_repeat_tool_call_from_final_response() -> None: + """A final response containing streamed history should not repeat its tool call.""" + function_call = Content.from_function_call( + call_id="weather-call", + name="get_weather", + arguments={"city": "Seattle"}, + ) + + @executor(id="participant") + async def participant(message: Any, ctx: WorkflowContext[Any, AgentResponse | AgentResponseUpdate]) -> None: + del message + await ctx.yield_output(AgentResponseUpdate(contents=[function_call], role=None)) + await ctx.yield_output( + AgentResponse( + messages=[ + Message(role="assistant", contents=[function_call]), + Message( + role="tool", + contents=[Content.from_function_result(call_id="weather-call", result="Sunny in Seattle")], + ), + Message(role="assistant", contents=[Content.from_text("The weather is sunny.")]), + ] + ) + ) + + workflow = WorkflowBuilder(start_executor=participant, output_from="all").build() + events = [ + event + async for event in run_workflow_stream( + {"messages": [{"role": "user", "content": "What is the weather in Seattle?"}]}, + workflow, + ) + ] + + tool_call_starts = [event for event in events if event.type == "TOOL_CALL_START"] + assert [event.tool_call_id for event in tool_call_starts] == ["weather-call"] # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + + async def test_workflow_run_passthroughs_ag_ui_base_events(): """Workflow outputs that are AG-UI BaseEvent instances should be emitted directly.""" From 325f5fd67d7e9f724497de20cf9bfd4c9c53662c Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Fri, 10 Jul 2026 15:31:09 +0900 Subject: [PATCH 3/3] Python: scope workflow tool content bypass to resumable tool calls - Exclude approval request content from the role bypass. Workflow approvals resume through request_info pending state, so an approval interrupt emitted from streamed content would have no pending request to resume against. - Admit mcp_server_tool_call and mcp_server_tool_result so provider-hosted MCP tool calls from workflow participants emit standard tool call events. - Add unit tests for MCP passthrough, approval exclusion, and mixed text-plus-tool content in non-assistant updates. --- .../agent_framework_ag_ui/_workflow_run.py | 9 ++++-- .../ag-ui/tests/ag_ui/test_workflow_run.py | 30 +++++++++++++++++-- 2 files changed, 34 insertions(+), 5 deletions(-) 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 1b4fdc75865..2bc10c172ea 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 @@ -62,7 +62,10 @@ } _INTERRUPT_CARD_EVENT_NAME = "WorkflowInterruptEvent" -_FUNCTION_CONTENT_TYPES = {"function_call", "function_result", "function_approval_request"} +# Tool content admitted from streaming updates regardless of role. Approval requests are +# deliberately excluded: workflow approvals resume through request_info pending state, and an +# approval interrupt emitted from streamed content would have no pending request to resume against. +_TOOL_CONTENT_TYPES = {"function_call", "function_result", "mcp_server_tool_call", "mcp_server_tool_result"} def _json_schema_for_response_type(response_type: Any) -> dict[str, Any] | None: @@ -684,8 +687,8 @@ def _workflow_payload_to_contents(payload: Any) -> list[Content] | None: role = str(getattr(role_field, "value", role_field)) if role == "assistant": return contents - function_contents = [content for content in contents if content.type in _FUNCTION_CONTENT_TYPES] - return function_contents or None + tool_contents = [content for content in contents if content.type in _TOOL_CONTENT_TYPES] + return tool_contents or None if isinstance(payload, AgentResponse): return _latest_assistant_contents(list(payload.messages or [])) if isinstance(payload, list): 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 e6976680bc8..5a0426258c5 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 @@ -1250,12 +1250,38 @@ def test_agent_response_update_function_result_with_tool_role(self) -> None: assert _workflow_payload_to_contents(update) == [function_result] def test_agent_response_update_approval_request_without_role(self) -> None: - """Approval request content passes through without role metadata.""" + """Approval request content is excluded from the role bypass. + + Workflow approvals resume through request_info pending state; an approval interrupt + emitted from streamed content would have no pending request to resume against. + """ function_call = Content.from_function_call(call_id="call-1", name="search", arguments={"query": "weather"}) approval_request = Content.from_function_approval_request(id="approval-1", function_call=function_call) update = AgentResponseUpdate(contents=[approval_request], role=None) - assert _workflow_payload_to_contents(update) == [approval_request] + assert _workflow_payload_to_contents(update) is None + + def test_agent_response_update_mcp_tool_call_without_role(self) -> None: + """MCP server tool call content passes through without role metadata.""" + mcp_call = Content.from_mcp_server_tool_call(call_id="mcp-1", tool_name="search", arguments={"q": "weather"}) + update = AgentResponseUpdate(contents=[mcp_call], role=None) + + assert _workflow_payload_to_contents(update) == [mcp_call] + + def test_agent_response_update_mcp_tool_result_without_role(self) -> None: + """MCP server tool result content passes through without role metadata.""" + mcp_result = Content.from_mcp_server_tool_result(call_id="mcp-1", output={"temperature": 72}) + update = AgentResponseUpdate(contents=[mcp_result], role=None) + + assert _workflow_payload_to_contents(update) == [mcp_result] + + def test_agent_response_update_mixed_content_without_role(self) -> None: + """Non-assistant updates keep tool content and drop text content.""" + text = Content.from_text(text="calling the tool") + function_call = Content.from_function_call(call_id="call-1", name="search", arguments={"query": "weather"}) + update = AgentResponseUpdate(contents=[text, function_call], role=None) + + assert _workflow_payload_to_contents(update) == [function_call] def test_agent_response_update_assistant_text(self) -> None: """Assistant text content continues to pass through."""