Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions python/packages/ag-ui/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions python/packages/ag-ui/agent_framework_ag_ui/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -39,6 +40,7 @@
"AgentFrameworkWorkflow",
"WorkflowFactory",
"add_agent_framework_fastapi_endpoint",
"agent_framework_messages_to_agui_host_history",
"AGUIChatClient",
"AGUIChatOptions",
"AGUIEventConverter",
Expand Down
21 changes: 10 additions & 11 deletions python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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
Expand Down
143 changes: 124 additions & 19 deletions python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,26 @@
)
from agent_framework._types import ContentType # pyright: ignore[reportPrivateUsage]

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,
Expand Down Expand Up @@ -1039,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,
Expand Down Expand Up @@ -1112,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))

Expand Down Expand Up @@ -1168,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]] = []
Expand Down Expand Up @@ -1204,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:
Expand Down Expand Up @@ -1235,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)
Expand All @@ -1255,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."""
Comment thread
eavanvalkenburg marked this conversation as resolved.
if messages and isinstance(messages[0], dict):
converted = _persistable_host_payload_history(
_convert_agent_framework_messages_to_agui(messages, preserve_host_history_dicts=True)
)
Comment thread
eavanvalkenburg marked this conversation as resolved.
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.

Expand Down
17 changes: 5 additions & 12 deletions python/packages/ag-ui/agent_framework_ag_ui/_run_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
9 changes: 9 additions & 0 deletions python/packages/ag-ui/agent_framework_ag_ui/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]] = []
Expand Down
Loading
Loading