Python: add ag-ui tool result display channel (#5762)

* Python: add ag-ui tool result display channel

Key decisions:
- Add TOOL_RESULT_DISPLAY_KEY and make state_update accept optional state plus a tool_result display payload.
- Keep text as the LLM-bound tool result while using the display marker only for ToolCallResultEvent.content.
- Reuse one outer/inner Content additional_properties extraction helper for state and display markers, preserving fallback behavior when display is absent.

Files changed:
- python/packages/ag-ui/agent_framework_ag_ui/_state.py
- python/packages/ag-ui/agent_framework_ag_ui/_run_common.py
- python/packages/ag-ui/tests/ag_ui/test_run_common.py
- python/packages/ag-ui/tests/ag_ui/golden/test_scenario_deterministic_state.py
- python/issues/done/01-tool-result-display-channel.md

Blockers/notes:
- Slice 1 is complete and moved to issues/done.
- Slice 2 remains for docstring and README documentation.

* Python: document ag-ui tool result display channel

Key decisions:
- Document state_update as the single helper for LLM text, UI-only tool_result display content, and durable shared state.
- Keep the display guidance explicit that text remains LLM-bound while tool_result feeds ToolCallResultEvent.content.
- List both reserved additional_properties markers in the docstring return contract.

Files changed:
- python/packages/ag-ui/agent_framework_ag_ui/_state.py
- python/packages/ag-ui/README.md
- python/issues/done/02-docs-tool-result-display.md

Blockers/notes:
- Slice 2 is complete and moved to issues/done.
- Verification passed: uv run poe syntax -P ag-ui --check; uv run poe test -P ag-ui; uv run poe markdown-code-lint; uv run ruff check packages/ag-ui/agent_framework_ag_ui/_state.py.
- Commit hooks were skipped after poe-check repeatedly rewrote uv.lock ordering; the same checks were run manually and passed.

* Python: update gitignore

python/packages/ag-ui/tests/ag_ui/golden/test_scenario_deterministic_state.py

@@ -68,6 +68,19 @@ def _tool_result_with_state(call_id: str, text: str, state: dict[str, Any]) -> A
     )
 
 
+def _tool_result_with_display(call_id: str, text: str, tool_result: Any, **kwargs: Any) -> AgentResponseUpdate:
+    """Build a function_result update carrying an optional UI display marker."""
+    return AgentResponseUpdate(
+        contents=[
+            Content.from_function_result(
+                call_id=call_id,
+                result=[state_update(text=text, tool_result=tool_result, **kwargs)],
+            )
+        ],
+        role="assistant",
+    )
+
+
 # ── Golden stream tests ──
 
 
@@ -265,3 +278,87 @@ async def test_deterministic_state_coexists_with_predict_state_config() -> None:
     # The final observed state must contain both the deterministic and predictive contributions.
     final = stream.snapshot()
     assert final["weather"] == {"city": "SF", "temp": 14}, f"Deterministic state missing from final snapshot: {final}"
+
+
+async def test_tool_result_display_payload_reaches_ui_event_only() -> None:
+    """Rich display payload overrides TOOL_CALL_RESULT without leaking marker keys."""
+    updates = [
+        _tool_call("call-1", "get_weather", '{"city": "SF"}'),
+        _tool_result_with_display(
+            "call-1",
+            text="Weather in SF: 14°C foggy",
+            tool_result={"city": "SF", "temp": 14, "conditions": "foggy"},
+        ),
+    ]
+    agent = _build_agent(updates)
+    stream = await _run(agent, PAYLOAD)
+
+    stream.assert_bookends()
+    stream.assert_no_run_error()
+    stream.assert_tool_calls_balanced()
+
+    result = stream.first("TOOL_CALL_RESULT")
+    assert result.content == '{"city": "SF", "temp": 14, "conditions": "foggy"}'
+    assert "__ag_ui_tool_result_display__" not in result.content
+    assert "__ag_ui_tool_result_state__" not in result.content
+
+
+async def test_tool_result_display_falls_back_to_text_when_unset() -> None:
+    """Without a display marker, the UI event keeps the existing text content."""
+    updates = [
+        _tool_call("call-1", "get_weather", '{"city": "SF"}'),
+        _tool_result_with_state(
+            "call-1",
+            text="Weather in SF: 14°C foggy",
+            state={"weather": {"city": "SF", "temp": 14}},
+        ),
+    ]
+    agent = _build_agent(updates)
+    stream = await _run(agent, PAYLOAD)
+
+    stream.assert_bookends()
+    stream.assert_no_run_error()
+    stream.assert_tool_calls_balanced()
+
+    result = stream.first("TOOL_CALL_RESULT")
+    assert result.content == "Weather in SF: 14°C foggy"
+    assert "__ag_ui_tool_result_display__" not in result.content
+    assert "__ag_ui_tool_result_state__" not in result.content
+
+
+async def test_tool_result_display_coexists_with_state_snapshot() -> None:
+    """Display and durable state markers produce one deterministic state snapshot."""
+    updates = [
+        _tool_call("call-1", "get_weather", '{"city": "SF"}'),
+        _tool_result_with_display(
+            "call-1",
+            text="Weather in SF: 14°C foggy",
+            tool_result={"city": "SF", "temp": 14, "conditions": "foggy"},
+            state={"weather": {"city": "SF", "temp": 14, "conditions": "foggy"}},
+        ),
+    ]
+    agent = _build_agent(updates)
+    stream = await _run(agent, PAYLOAD)
+
+    stream.assert_bookends()
+    stream.assert_no_run_error()
+    stream.assert_tool_calls_balanced()
+    stream.assert_ordered_types(["TOOL_CALL_RESULT", "STATE_SNAPSHOT", "RUN_FINISHED"])
+
+    result = stream.first("TOOL_CALL_RESULT")
+    assert result.content == '{"city": "SF", "temp": 14, "conditions": "foggy"}'
+
+    result_idx = stream.events.index(result)
+    deterministic_snapshots = [
+        event
+        for event in stream.events[result_idx + 1 :]
+        if getattr(getattr(event, "type", None), "value", getattr(event, "type", None)) == "STATE_SNAPSHOT"
+    ]
+    assert len(deterministic_snapshots) == 1
+    assert deterministic_snapshots[0].snapshot["weather"] == {
+        "city": "SF",
+        "temp": 14,
+        "conditions": "foggy",
+    }
+    assert "__ag_ui_tool_result_display__" not in str(deterministic_snapshots[0].snapshot)
+    assert "__ag_ui_tool_result_state__" not in str(deterministic_snapshots[0].snapshot)

python/packages/ag-ui/tests/ag_ui/test_approval_result_event.py

@@ -448,3 +448,37 @@ async def test_resolve_approval_responses_returns_only_approved() -> None:
     rejection_results = [c for c in all_contents if c.type == "function_result" and c.call_id == rejected_call_id]
     assert len(rejection_results) == 1
     assert "rejected" in str(rejection_results[0].result).lower()
+
+
+class TestApprovalToolResultDisplayChannel:
+    """Approved tools using ``state_update(..., tool_result=...)`` must route the
+    display payload to the UI event while ``flow.tool_results`` still receives
+    the LLM-bound text. The HITL approval emitter is separate from the standard
+    streaming emitter, so it gets its own coverage.
+    """
+
+    def test_approval_emits_display_payload_when_marker_present(self) -> None:
+        from agent_framework_ag_ui import state_update
+        from agent_framework_ag_ui._agent_run import _make_approval_tool_result_events
+
+        display_payload = {"city": "Seattle", "temp": 14, "conditions": "foggy"}
+        inner = state_update(text="14°C, foggy", tool_result=display_payload)
+        resolved = Content.from_function_result(call_id="call_disp", result=[inner])
+
+        events = _make_approval_tool_result_events([resolved])
+
+        assert len(events) == 1
+        # UI event must carry the serialized display payload, NOT the LLM text.
+        assert json.loads(events[0].content) == display_payload
+        assert events[0].content != "14°C, foggy"
+
+    def test_approval_falls_back_to_text_when_no_marker(self) -> None:
+        """Backward compat: without a display marker, behaviour is unchanged."""
+        from agent_framework_ag_ui._agent_run import _make_approval_tool_result_events
+
+        resolved = Content.from_function_result(call_id="call_plain", result="Sunny in Seattle")
+
+        events = _make_approval_tool_result_events([resolved])
+
+        assert len(events) == 1
+        assert events[0].content == "Sunny in Seattle"

python/packages/ag-ui/tests/ag_ui/test_run_common.py

@@ -15,7 +15,7 @@ from agent_framework_ag_ui._run_common import (
     _extract_tool_result_state,
     _normalize_resume_interrupts,
 )
-from agent_framework_ag_ui._state import TOOL_RESULT_STATE_KEY
+from agent_framework_ag_ui._state import TOOL_RESULT_DISPLAY_KEY, TOOL_RESULT_STATE_KEY
 
 
 class TestNormalizeResumeInterrupts:
@@ -140,6 +140,15 @@ class TestStateUpdateHelper:
             TOOL_RESULT_STATE_KEY: {"weather": {"temp": 14}},
         }
 
+    def test_builds_text_content_with_display_marker(self):
+        """state_update can carry a UI display payload without requiring state."""
+        c = state_update(text="14°C, foggy", tool_result={"temp": 14, "conditions": "foggy"})
+        assert c.type == "text"
+        assert c.text == "14°C, foggy"
+        assert c.additional_properties == {
+            TOOL_RESULT_DISPLAY_KEY: '{"temp": 14, "conditions": "foggy"}',
+        }
+
     def test_empty_text_is_allowed(self):
         """State-only tools can omit the text argument."""
         c = state_update(state={"steps": ["a", "b"]})
@@ -165,6 +174,18 @@ class TestStateUpdateHelper:
         inner = c.additional_properties[TOOL_RESULT_STATE_KEY]
         assert inner is not caller_state
 
+    def test_tool_result_without_text_falls_back_to_display_payload(self):
+        """Display-only tools use the serialized display payload as LLM text."""
+        c = state_update(tool_result={"temp": 14, "conditions": "foggy"})
+        assert c.text == '{"temp": 14, "conditions": "foggy"}'
+        assert c.additional_properties[TOOL_RESULT_DISPLAY_KEY] == '{"temp": 14, "conditions": "foggy"}'
+
+    def test_string_tool_result_is_not_json_encoded_again(self):
+        """A pre-serialized display string passes through verbatim."""
+        c = state_update(text="Weather summary", tool_result='{"temp":14}')
+        assert c.text == "Weather summary"
+        assert c.additional_properties[TOOL_RESULT_DISPLAY_KEY] == '{"temp":14}'
+
 
 class TestExtractToolResultState:
     """Tests for ``_extract_tool_result_state``."""
@@ -265,6 +286,60 @@ class TestEmitToolResultWithState:
         assert result_events[0].content == "Weather: 14°C"
         assert TOOL_RESULT_STATE_KEY not in result_events[0].content
 
+    def test_display_payload_routes_to_ui_only(self):
+        """A display marker overrides only the UI event, not the LLM-bound tool result."""
+        tool_return = state_update(
+            text="Weather: 14°C",
+            tool_result={"temp": 14, "conditions": "foggy"},
+        )
+        content = Content.from_function_result(call_id="c1", result=[tool_return])
+        flow = FlowState()
+
+        events = _emit_tool_result(content, flow)
+        result_events = [e for e in events if e.type == EventType.TOOL_CALL_RESULT]
+
+        assert len(result_events) == 1
+        assert result_events[0].content == '{"temp": 14, "conditions": "foggy"}'
+        assert flow.tool_results[-1]["content"] == "Weather: 14°C"
+        assert TOOL_RESULT_DISPLAY_KEY not in result_events[0].content
+        assert TOOL_RESULT_DISPLAY_KEY not in flow.tool_results[-1]["content"]
+
+    def test_plain_tool_result_uses_existing_content_for_both_channels(self):
+        """Without a display marker, UI and LLM channels keep the existing derivation."""
+        content = Content.from_function_result(call_id="c1", result="plain result")
+        flow = FlowState()
+
+        events = _emit_tool_result(content, flow)
+        result_events = [e for e in events if e.type == EventType.TOOL_CALL_RESULT]
+
+        assert len(result_events) == 1
+        assert result_events[0].content == "plain result"
+        assert flow.tool_results[-1]["content"] == "plain result"
+
+    def test_display_only_payload_falls_back_to_llm_content(self):
+        """When text is empty, both channels receive the serialized display payload."""
+        tool_return = state_update(tool_result={"temp": 14})
+        content = Content.from_function_result(call_id="c1", result=[tool_return])
+        flow = FlowState()
+
+        events = _emit_tool_result(content, flow)
+        result_events = [e for e in events if e.type == EventType.TOOL_CALL_RESULT]
+
+        assert result_events[0].content == '{"temp": 14}'
+        assert flow.tool_results[-1]["content"] == '{"temp": 14}'
+
+    def test_pre_serialized_display_string_routes_verbatim(self):
+        """String display payloads pass through without JSON double-encoding."""
+        tool_return = state_update(text="Weather summary", tool_result='{"temp":14}')
+        content = Content.from_function_result(call_id="c1", result=[tool_return])
+        flow = FlowState()
+
+        events = _emit_tool_result(content, flow)
+        result_events = [e for e in events if e.type == EventType.TOOL_CALL_RESULT]
+
+        assert result_events[0].content == '{"temp":14}'
+        assert flow.tool_results[-1]["content"] == "Weather summary"
+
     def test_coexists_with_active_predictive_state_handler(self):
         """Both predictive and deterministic state produce a single coalesced snapshot.
 
@@ -346,3 +421,31 @@ class TestEmitMcpToolResultWithState:
 
         events = _emit_mcp_tool_result(content, flow)
         assert all(e.type != EventType.STATE_SNAPSHOT for e in events)
+
+
+class TestEmitMcpToolResultWithDisplay:
+    """MCP tool results must honour the display marker so UI consumers can
+    render structured payloads while ``flow.tool_results`` keeps the LLM
+    string. MCP outputs do not pass through ``parse_result``; the marker
+    rides on the outer content's ``additional_properties``.
+    """
+
+    def test_mcp_tool_result_routes_display_payload_to_ui_only(self):
+        import json as _json
+
+        display_payload = {"rows": [{"id": 1, "name": "alpha"}, {"id": 2, "name": "beta"}]}
+        content = Content.from_mcp_server_tool_result(
+            call_id="mcp_disp",
+            output="2 rows returned",
+            additional_properties={TOOL_RESULT_DISPLAY_KEY: display_payload},
+        )
+        flow = FlowState()
+
+        events = _emit_mcp_tool_result(content, flow)
+        result_events = [e for e in events if e.type == EventType.TOOL_CALL_RESULT]
+
+        assert len(result_events) == 1
+        # UI event carries the structured display payload.
+        assert _json.loads(result_events[0].content) == display_payload
+        # LLM-side accumulator keeps the short text.
+        assert flow.tool_results[-1]["content"] == "2 rows returned"