Python: [BREAKING]: Emit TOOL_CALL events for workflow participant tool calls in AG-UI

python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py

@@ -62,6 +62,10 @@
 }
 
 _INTERRUPT_CARD_EVENT_NAME = "WorkflowInterruptEvent"
+# 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:
@@ -675,16 +679,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
+        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):

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,104 @@ 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_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."""
 
@@ -1132,6 +1235,61 @@ 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 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) 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."""
+        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])