microsoft · Eduard van Valkenburg (eavanvalkenburg) · Aug 31, 2026
@@ -604,6 +604,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | AG-UI approval-time follow-up | The full grouped user-input pause remains in message history and emits no synthetic `TOOL_CALL_RESULT`. | `packages/ag-ui/tests/ag_ui/test_approval_result_event.py::test_approval_follow_up_group_remains_in_history_without_live_tool_result` |
 | AG-UI approval execution failure | A grouped executor failure becomes one deterministic terminal error result for the approved call. | `packages/ag-ui/tests/ag_ui/test_approval_result_event.py::test_approval_execution_failure_emits_one_terminal_error_result` |
 | AG-UI no-approval path | Ordinary tool results do not gain an extra approval result event. | `packages/ag-ui/tests/ag_ui/test_approval_result_event.py::test_no_approval_path_emits_no_approval_specific_duplicate_result` |
+| AG-UI MCP Host payload | Core preserves bounded successful and error MCP `CallToolResult` payloads separately from built-in or custom model-facing results; oversized Host payloads are rejected by a bounded preflight without changing the model projection. AG-UI projects retained Host payloads consistently through ordinary and approval-resolved live events, aggregate-bounded messages snapshots, and the supported Host-history converter without changing generic outbound requests or replaying UI-only data to the model. | `packages/core/tests/core/test_mcp.py::test_parse_tool_result_from_mcp_preserves_complete_host_payload_once`, `test_custom_mcp_result_parser_preserves_host_payload_and_model_projection`, `test_oversized_mcp_host_payload_is_omitted_without_changing_model_result`, `test_mcp_host_payload_size_preflight_matches_json_and_aborts_before_dump`, `test_mcp_error_preserves_complete_host_payload_on_function_result`, `packages/ag-ui/tests/ag_ui/test_run_common.py::TestEmitToolResultWithState::test_plain_tool_result_does_not_serialize_replay_items`, `test_mcp_host_payload_routes_to_live_event_and_snapshot`, `test_mcp_snapshot_replays_rich_model_items_without_host_payload_duplication`, `test_messages_snapshot_bounds_cumulative_mcp_host_payloads`, `packages/ag-ui/tests/ag_ui/test_message_adapters.py::test_agent_framework_to_agui_preserves_mcp_host_payload_after_reload`, `test_host_history_conversion_is_public_and_bounds_cumulative_payloads`, `test_agui_mcp_fallback_requires_provenance_and_hides_error_details` |
 | AG-UI client-tool request isolation | Client tool declarations are validated before use and remain request-scoped; a rejected collision or earlier successful request cannot change a later request's server-tool execution. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_failed_client_tool_collision_does_not_affect_next_request`, `test_endpoint_client_tools_do_not_persist_into_next_request` |
 | AG-UI `confirm_changes` snapshot | An accepted synthetic confirmation is replaced only when its original function call has a real result; rejection is cleaned explicitly, and missing accepted results remain inert. | `packages/ag-ui/tests/ag_ui/test_confirm_changes_snapshot.py` |
 | AG-UI malformed `confirm_changes` metadata | Non-list tool-call metadata and malformed argument JSON are ignored without guessing a target call. | `test_confirm_changes_target_ignores_non_list_tool_calls`, `test_confirm_changes_target_rejects_malformed_arguments_json` |

@@ -8,6 +8,8 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
 - **`AgentFrameworkWorkflow`** - Wraps native `Workflow` objects, or accepts `workflow_factory(thread_id)` for thread-scoped workflow instances without subclassing
 - **`AGUIChatClient`** - Chat client that speaks AG-UI protocol
 - **`AGUIHttpService`** - HTTP service for AG-UI endpoints
+- **`agent_framework_messages_to_agui_host_history()`** - Converts persisted Agent Framework messages to bounded
+  AG-UI Host history while retaining MCP widget payloads and model replay metadata
 - **`AGUIEventConverter`** - Converts between Agent Framework and AG-UI events
 - **`add_agent_framework_fastapi_endpoint()`** - Add AG-UI endpoint to FastAPI app (`SupportsAgentRun` or `Workflow`)
 - **`InMemoryAGUIThreadSnapshotStore`** - Memory-only latest AG-UI Thread Snapshot store for local development, demos, and tests

@@ -10,6 +10,7 @@
 from ._endpoint import add_agent_framework_fastapi_endpoint
 from ._event_converters import AGUIEventConverter
 from ._http_service import AGUIHttpService
+from ._message_adapters import agent_framework_messages_to_agui_host_history
 from ._snapshots import (
     DEFAULT_MAX_THREAD_SNAPSHOTS,
     AGUIThreadID,
@@ -39,6 +40,7 @@
     "AgentFrameworkWorkflow",
     "WorkflowFactory",
     "add_agent_framework_fastapi_endpoint",
+    "agent_framework_messages_to_agui_host_history",
     "AGUIChatClient",
     "AGUIChatOptions",
     "AGUIEventConverter",

@@ -91,8 +91,8 @@
     _new_tool_call_segment_id,  # type: ignore
     _reconstruct_messages_from_thread_snapshot,  # type: ignore
     _resume_contract_error,  # type: ignore
+    _resolve_tool_result_host_payload,  # type: ignore
     _resolve_ui_payload,  # type: ignore
-    _stringify_tool_result,  # type: ignore
     _track_tool_call_segment,  # type: ignore
 )
 from ._snapshots import (
@@ -102,8 +102,14 @@
 )
 from ._snapshot_session import ThreadSnapshotSession, _event_messages_to_snapshot_dicts
 from ._utils import (
+    _AGUI_MCP_TOOL_RESULT_KEY,
+    _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY,
     _approval_interrupt_id,
+    _bound_host_payload_history,
     _function_call_server_label,
+    _model_items_for_agui_replay,
+    _stringify_tool_result,
+    DEFAULT_MAX_HOST_PAYLOAD_HISTORY_SIZE_BYTES,
     canonical_function_arguments,
     convert_agui_tools_to_agent_framework,
     generate_event_id,
@@ -687,7 +693,9 @@ def _make_approval_tool_result_events(resolved_approval_results: list[Content])
         if resolved.call_id:
             raw = resolved.result if resolved.result is not None else ""
             llm_str = _stringify_tool_result(raw)
-            ui_str = _resolve_ui_payload(llm_str, _extract_tool_result_display(resolved))
+            display_result = _extract_tool_result_display(resolved)
+            has_host_payload, host_payload = _resolve_tool_result_host_payload(resolved, display_result)
+            ui_str = _resolve_ui_payload(llm_str, host_payload if has_host_payload else display_result)
             events.append(
                 ToolCallResultEvent(
                     message_id=generate_event_id(),
@@ -1968,12 +1976,21 @@ def _resolved_tool_result_snapshot_messages(resolved_messages: list[Message]) ->
         ]
         for content in function_results:
             call_id = str(content.call_id)
-            result_by_call_id[call_id] = {
+            llm_result = _stringify_tool_result(content.result if content.result is not None else "")
+            display_result = _extract_tool_result_display(content)
+            has_host_payload, host_payload = _resolve_tool_result_host_payload(content, display_result)
+            snapshot_message: dict[str, Any] = {
                 "id": msg.message_id if msg.message_id and len(function_results) == 1 else generate_event_id(),
                 "role": "tool",
                 "toolCallId": call_id,
-                "content": _stringify_tool_result(content.result if content.result is not None else ""),
+                "content": _stringify_tool_result(host_payload) if has_host_payload else llm_result,
             }
+            if has_host_payload:
+                snapshot_message[_AGUI_MCP_TOOL_RESULT_KEY] = True
+                snapshot_message[_AGUI_TOOL_RESULT_MODEL_CONTENT_KEY] = _model_items_for_agui_replay(
+                    content, llm_result
+                )
+            result_by_call_id[call_id] = snapshot_message
     return result_by_call_id
 
 
@@ -2067,7 +2084,12 @@ def _build_messages_snapshot(
 
     if flow.snapshot_segments:
         _append_segmented_snapshot_messages(flow, all_messages)
-        return MessagesSnapshotEvent(messages=all_messages)  # type: ignore[arg-type]
+        return MessagesSnapshotEvent(
+            messages=_bound_host_payload_history(  # type: ignore[arg-type]
+                all_messages,
+                max_size_bytes=DEFAULT_MAX_HOST_PAYLOAD_HISTORY_SIZE_BYTES,
+            )
+        )  # type: ignore[arg-type]
 
     # Add assistant message with tool calls only (no content)
     if flow.pending_tool_calls:
@@ -2101,7 +2123,12 @@ def _build_messages_snapshot(
     # MESSAGES_SNAPSHOT retain reasoning content after streaming ends.
     all_messages.extend(flow.reasoning_messages)
 
-    return MessagesSnapshotEvent(messages=all_messages)  # type: ignore[arg-type]
+    return MessagesSnapshotEvent(
+        messages=_bound_host_payload_history(  # type: ignore[arg-type]
+            all_messages,
+            max_size_bytes=DEFAULT_MAX_HOST_PAYLOAD_HISTORY_SIZE_BYTES,
+        )
+    )  # type: ignore[arg-type]
 
 
 def _text_events_to_snapshot_messages(events: list[BaseEvent]) -> list[dict[str, Any]]:

@@ -15,9 +15,19 @@
     Message,
 )
 
+from ._state import TOOL_RESULT_DISPLAY_KEY
 from ._utils import (
+    _AGUI_MCP_TOOL_RESULT_KEY,
+    _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY,
     AGUI_TO_FRAMEWORK_ROLE,
+    DEFAULT_MAX_HOST_PAYLOAD_HISTORY_SIZE_BYTES,
     FRAMEWORK_TO_AGUI_ROLE,
+    _bound_host_payload_history,
+    _extract_mcp_tool_result_host_payload,
+    _extract_tool_result_marker_values,
+    _model_content_from_mcp_host_payload,
+    _model_items_for_agui_replay,
+    _stringify_tool_result,
     get_role_value,
     normalize_agui_role,
     safe_json_parse,
@@ -845,6 +855,29 @@ def _filter_modified_args(
                 result.append(chat_msg)
                 continue
 
+            if msg.get(_AGUI_MCP_TOOL_RESULT_KEY) is True:
+                serialized_items = msg.get(_AGUI_TOOL_RESULT_MODEL_CONTENT_KEY)
+                model_items: list[Content] | None = None
+                if isinstance(serialized_items, list):
+                    try:
+                        model_items = [
+                            Content.from_dict(item)
+                            for item in serialized_items
+                            if isinstance(item, dict) and item.get("type")
+                        ]
+                    except (TypeError, ValueError):
+                        model_items = None
+                if not model_items:
+                    model_items = [Content.from_text(_model_content_from_mcp_host_payload(parsed))]
+                chat_msg = Message(
+                    role="tool",
+                    contents=[Content.from_function_result(call_id=str(tool_call_id), result=model_items)],
+                )
+                if "id" in msg:
+                    chat_msg.message_id = msg["id"]
+                result.append(chat_msg)
+                continue
+
             # Cast result_content to acceptable type for function_result content
             func_result: str | dict[str, Any] | list[Any]
             if isinstance(result_content, str):
@@ -958,15 +991,13 @@ def _filter_modified_args(
     return result
 
 
-def agent_framework_messages_to_agui(messages: list[Message] | list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Convert Agent Framework messages to AG-UI format.
-
-    Args:
-        messages: List of Agent Framework Message objects or AG-UI dicts (already converted)
+def _convert_agent_framework_messages_to_agui(
+    messages: list[Message] | list[dict[str, Any]],
+    *,
+    include_host_payload: bool,
+) -> list[dict[str, Any]]:
+    """Convert Agent Framework messages to AG-UI format."""
 
-    Returns:
-        List of AG-UI message dictionaries
-    """
     from ._utils import generate_event_id
 
     result: list[dict[str, Any]] = []
@@ -1025,14 +1056,24 @@ def agent_framework_messages_to_agui(messages: list[Message] | list[dict[str, An
             # id (e.g. f"{base_id}-1") risks colliding with a legitimate id elsewhere
             # in the history, which would let id-keyed clients re-collapse results.
             for idx, fr in enumerate(function_results):
-                result.append(
-                    {
-                        "id": msg.message_id if (idx == 0 and msg.message_id) else generate_event_id(),
-                        "role": "tool",
-                        "content": fr.result if fr.result is not None else "",
-                        "toolCallId": fr.call_id,
-                    }
-                )
+                model_result = fr.result if fr.result is not None else ""
+                tool_message: dict[str, Any] = {
+                    "id": msg.message_id if (idx == 0 and msg.message_id) else generate_event_id(),
+                    "role": "tool",
+                    "content": model_result,
+                    "toolCallId": fr.call_id,
+                }
+                has_host_payload, host_payload = _extract_mcp_tool_result_host_payload(fr)
+                if include_host_payload and has_host_payload:
+                    display_values = _extract_tool_result_marker_values(fr, TOOL_RESULT_DISPLAY_KEY)
+                    tool_message["content"] = _stringify_tool_result(
+                        display_values[-1] if display_values else host_payload
+                    )
+                    tool_message[_AGUI_MCP_TOOL_RESULT_KEY] = True
+                    tool_message[_AGUI_TOOL_RESULT_MODEL_CONTENT_KEY] = _model_items_for_agui_replay(
+                        fr, _stringify_tool_result(model_result)
+                    )
+                result.append(tool_message)
             # A mixed message may also carry text / function_call contents alongside
             # the tool results (e.g. a finalized assistant turn). Emit those as a
             # separate, distinctly-identified message so they are not lost.
@@ -1061,6 +1102,24 @@ def agent_framework_messages_to_agui(messages: list[Message] | list[dict[str, An
     return result
 
 
+def agent_framework_messages_to_agui(messages: list[Message] | list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Convert Agent Framework messages to model-safe AG-UI request format."""
+    return _convert_agent_framework_messages_to_agui(messages, include_host_payload=False)
+
+
+def agent_framework_messages_to_agui_host_history(
+    messages: list[Message] | list[dict[str, Any]],
+    *,
+    max_host_payload_history_size_bytes: int = DEFAULT_MAX_HOST_PAYLOAD_HISTORY_SIZE_BYTES,
+) -> list[dict[str, Any]]:
+    """Convert Agent Framework messages to bounded AG-UI Host history with replay metadata."""
+    converted = _convert_agent_framework_messages_to_agui(messages, include_host_payload=True)
+    return _bound_host_payload_history(
+        converted,
+        max_size_bytes=max_host_payload_history_size_bytes,
+    )
+
+
 def extract_text_from_contents(contents: list[Any]) -> str:
     """Extract text from Agent Framework contents.