@@ -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",
 
@@ -103,11 +103,10 @@
 from ._snapshot_session import ThreadSnapshotSession, _event_messages_to_snapshot_dicts
 from ._utils import (
     _AGUI_MCP_TOOL_RESULT_KEY,
-    _AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY,
-    _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY,
     _approval_interrupt_id,
     _bound_host_payload_history,
     _function_call_server_label,
+    _mcp_host_history_fields,
     _model_items_for_agui_replay,
     _persistable_host_payload_history,
     _project_host_payload_history,
@@ -700,11 +699,10 @@ def _make_approval_tool_result_events(resolved_approval_results: list[Content])
             ui_str = _resolve_ui_payload(llm_str, host_payload if has_host_payload else display_result)
             replay_properties: dict[str, Any] = {}
             if has_host_payload:
-                replay_properties = {
-                    _AGUI_MCP_TOOL_RESULT_KEY: True,
-                    _AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY: _stringify_tool_result(host_payload),
-                    _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY: _model_items_for_agui_replay(resolved, llm_str),
-                }
+                replay_properties = _mcp_host_history_fields(
+                    host_payload,
+                    _model_items_for_agui_replay(resolved, llm_str),
+                )
             events.append(
                 ToolCallResultEvent(
                     message_id=generate_event_id(),
@@ -1996,10 +1994,11 @@ def _resolved_tool_result_snapshot_messages(resolved_messages: list[Message]) ->
                 "content": llm_result,
             }
             if has_host_payload:
-                snapshot_message[_AGUI_MCP_TOOL_RESULT_KEY] = True
-                snapshot_message[_AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY] = _stringify_tool_result(host_payload)
-                snapshot_message[_AGUI_TOOL_RESULT_MODEL_CONTENT_KEY] = _model_items_for_agui_replay(
-                    content, llm_result
+                snapshot_message.update(
+                    _mcp_host_history_fields(
+                        host_payload,
+                        _model_items_for_agui_replay(content, llm_result),
+                    )
                 )
             result_by_call_id[call_id] = snapshot_message
     return result_by_call_id
 
@@ -8,24 +8,34 @@
 import binascii
 import json
 import logging
-from typing import Any, cast
+from typing import Any, cast, get_args
 
 from agent_framework import (
     Content,
     Message,
 )
 from agent_framework._types import ContentType  # pyright: ignore[reportPrivateUsage]
-from typing_extensions import get_args
 
+from ._state import TOOL_RESULT_DISPLAY_KEY
 from ._utils import (
     _AGUI_HOST_PAYLOAD_OMITTED_KEY,
     _AGUI_MCP_TOOL_RESULT_KEY,
     _AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY,
     _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY,
+    _MAX_MCP_HOST_PAYLOAD_HISTORY_SIZE_BYTES,
     AGUI_TO_FRAMEWORK_ROLE,
     FRAMEWORK_TO_AGUI_ROLE,
+    _bound_host_payload_history,
+    _extract_mcp_tool_result_host_payload,
+    _extract_tool_result_marker_values,
+    _host_payload_history_size,
+    _mcp_host_history_fields,
     _model_content_from_mcp_host_payload,
+    _model_items_for_agui_replay,
+    _persistable_host_payload_history,
+    _project_host_payload_history,
     _sanitize_model_replay_item,
+    _stringify_tool_result,
     get_role_value,
     normalize_agui_role,
     safe_json_parse,
@@ -1040,7 +1050,12 @@ def _encode_agui_segment(contents: list[Content]) -> tuple[str, list[dict[str, A
     return text, tool_calls
 
 
-def _split_mixed_message_to_agui(msg: Message, role: str, unresolved_call_ids: set[str]) -> list[dict[str, Any]]:
+def _split_mixed_message_to_agui(
+    msg: Message,
+    role: str,
+    unresolved_call_ids: set[str],
+    emitted_results: list[tuple[Content, dict[str, Any]]] | None = None,
+) -> list[dict[str, Any]]:
     """Convert a Message that carries function_result content into ordered AG-UI messages.
 
     A single Agent Framework message can interleave assistant content (text,
@@ -1113,14 +1128,15 @@ def flush_segment() -> None:
         messages.append(assistant_msg)
 
     def emit_result(content: Content) -> None:
-        messages.append(
-            {
-                "id": next_id(),
-                "role": "tool",
-                "content": content.result if content.result is not None else "",
-                "toolCallId": content.call_id,
-            }
-        )
+        tool_message: dict[str, Any] = {
+            "id": next_id(),
+            "role": "tool",
+            "content": content.result if content.result is not None else "",
+            "toolCallId": content.call_id,
+        }
+        messages.append(tool_message)
+        if emitted_results is not None:
+            emitted_results.append((content, tool_message))
         if content.call_id is not None:
             unresolved_call_ids.discard(str(content.call_id))
 
@@ -1169,15 +1185,13 @@ def drain_queued() -> None:
     return messages
 
 
-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)
-
-    Returns:
-        List of AG-UI message dictionaries
-    """
+def _convert_agent_framework_messages_to_agui(
+    messages: list[Message] | list[dict[str, Any]],
+    *,
+    emitted_results: list[tuple[Content, dict[str, Any]]] | None = None,
+    preserve_host_history_dicts: bool = False,
+) -> list[dict[str, Any]]:
+    """Convert Agent Framework messages to AG-UI format."""
     from ._utils import generate_event_id
 
     result: list[dict[str, Any]] = []
@@ -1205,6 +1219,12 @@ def track_emitted(
         if isinstance(msg, dict):
             # Always work on a copy to avoid mutating input
             normalized_msg = msg.copy()
+            if not preserve_host_history_dicts and normalized_msg.get(_AGUI_MCP_TOOL_RESULT_KEY) is True:
+                normalized_msg = _persistable_host_payload_history([normalized_msg])[0].copy()
+                normalized_msg.pop(_AGUI_MCP_TOOL_RESULT_KEY, None)
+                normalized_msg.pop(_AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY, None)
+                normalized_msg.pop(_AGUI_TOOL_RESULT_MODEL_CONTENT_KEY, None)
+                normalized_msg.pop(_AGUI_HOST_PAYLOAD_OMITTED_KEY, None)
             normalized_msg["role"] = normalize_agui_role(normalized_msg.get("role"))
             # Ensure ID exists
             if "id" not in normalized_msg:
@@ -1236,7 +1256,14 @@ def track_emitted(
         # result is dropped and each result stays after its matching call. Messages
         # with no result use the simple single-message form below.
         if any(content.type == "function_result" for content in msg.contents):
-            result.extend(_split_mixed_message_to_agui(msg, role, unresolved_call_ids))
+            result.extend(
+                _split_mixed_message_to_agui(
+                    msg,
+                    role,
+                    unresolved_call_ids,
+                    emitted_results,
+                )
+            )
             continue
 
         content_text, tool_calls = _encode_agui_segment(msg.contents)
@@ -1256,6 +1283,83 @@ def track_emitted(
     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)
+
+
+def _prepare_host_history_fields(
+    emitted_results: list[tuple[Content, dict[str, Any]]],
+    *,
+    max_size_bytes: int,
+) -> tuple[dict[int, dict[str, Any]], set[int]]:
+    """Materialize only the newest MCP Host projections that fit the aggregate budget."""
+    retained_fields: dict[int, dict[str, Any]] = {}
+    omitted_ids: set[int] = set()
+    retained_size = 0
+    budget_exhausted = False
+
+    for content, _ in reversed(emitted_results):
+        has_host_payload, host_payload = _extract_mcp_tool_result_host_payload(content)
+        if not has_host_payload:
+            continue
+        content_id = id(content)
+        if budget_exhausted:
+            omitted_ids.add(content_id)
+            continue
+
+        display_values = _extract_tool_result_marker_values(content, TOOL_RESULT_DISPLAY_KEY)
+        if display_values:
+            host_payload = display_values[-1]
+        model_result = _stringify_tool_result(content.result if content.result is not None else "")
+        fields = _mcp_host_history_fields(
+            host_payload,
+            _model_items_for_agui_replay(content, model_result),
+        )
+        message_size = _host_payload_history_size({"content": model_result, **fields})
+        if retained_size + message_size > max_size_bytes:
+            omitted_ids.add(content_id)
+            budget_exhausted = True
+            continue
+        retained_size += message_size
+        retained_fields[content_id] = fields
+
+    return retained_fields, omitted_ids
+
+
+def agent_framework_messages_to_agui_host_history(
+    messages: list[Message] | list[dict[str, Any]],
+    *,
+    max_host_payload_history_size_bytes: int = _MAX_MCP_HOST_PAYLOAD_HISTORY_SIZE_BYTES,
+) -> list[dict[str, Any]]:
+    """Convert Agent Framework messages to bounded AG-UI Host history with replay metadata."""
+    if messages and isinstance(messages[0], dict):
+        converted = _persistable_host_payload_history(
+            _convert_agent_framework_messages_to_agui(messages, preserve_host_history_dicts=True)
+        )
+    else:
+        message_objects = cast(list[Message], messages)
+        emitted_results: list[tuple[Content, dict[str, Any]]] = []
+        converted = _convert_agent_framework_messages_to_agui(
+            message_objects,
+            emitted_results=emitted_results,
+        )
+        host_history_fields, omitted_ids = _prepare_host_history_fields(
+            emitted_results,
+            max_size_bytes=max_host_payload_history_size_bytes,
+        )
+        for content, tool_message in emitted_results:
+            if fields := host_history_fields.get(id(content)):
+                tool_message.update(fields)
+            elif id(content) in omitted_ids:
+                tool_message[_AGUI_HOST_PAYLOAD_OMITTED_KEY] = True
+    bounded = _bound_host_payload_history(
+        converted,
+        max_size_bytes=max_host_payload_history_size_bytes,
+    )
+    return _project_host_payload_history(bounded)
+
+
 def extract_text_from_contents(contents: list[Any]) -> str:
     """Extract text from Agent Framework contents.
 
 
@@ -38,12 +38,10 @@
 from ._predictive_state import PredictiveStateHandler
 from ._state import TOOL_RESULT_DISPLAY_KEY, TOOL_RESULT_STATE_KEY
 from ._utils import (
-    _AGUI_MCP_TOOL_RESULT_KEY,
-    _AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY,
-    _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY,
     _approval_interrupt_id,
     _extract_mcp_tool_result_host_payload,
     _extract_tool_result_marker_values,
+    _mcp_host_history_fields,
     _model_items_for_agui_replay,
     _stringify_tool_result,
     generate_event_id,
@@ -810,16 +808,11 @@ def _emit_tool_result_common(
     }
     event_replay_properties: dict[str, Any] = {}
     if snapshot_result is not _UNSET:
-        snapshot_message[_AGUI_MCP_TOOL_RESULT_KEY] = True
-        snapshot_message[_AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY] = snapshot_result_content
-        snapshot_message[_AGUI_TOOL_RESULT_MODEL_CONTENT_KEY] = (
-            [{"type": "text", "text": result_content}] if model_items is None else model_items
+        event_replay_properties = _mcp_host_history_fields(
+            snapshot_result_content,
+            [{"type": "text", "text": result_content}] if model_items is None else model_items,
         )
-        event_replay_properties = {
-            _AGUI_MCP_TOOL_RESULT_KEY: True,
-            _AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY: snapshot_result_content,
-            _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY: snapshot_message[_AGUI_TOOL_RESULT_MODEL_CONTENT_KEY],
-        }
+        snapshot_message.update(event_replay_properties)
         events[-1] = ToolCallResultEvent(
             message_id=message_id,
             tool_call_id=call_id,
 
@@ -204,6 +204,15 @@ def _host_payload_history_size(message: dict[str, Any]) -> int:
     return content_size + sidecar_size
 
 
+def _mcp_host_history_fields(host_payload: Any, model_items: list[dict[str, Any]]) -> dict[str, Any]:
+    """Build the private fields that preserve one MCP Host result for safe replay."""
+    return {
+        _AGUI_MCP_TOOL_RESULT_KEY: True,
+        _AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY: _stringify_tool_result(host_payload),
+        _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY: model_items,
+    }
+
+
 def _persistable_host_payload_history(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
     """Keep canonical persisted content safe for readers that ignore private replay fields."""
     persisted: list[dict[str, Any]] = []