diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index bd79ecda92a..87a40da4ee5 100644 --- a/python/packages/ag-ui/AGENTS.md +++ b/python/packages/ag-ui/AGENTS.md @@ -33,7 +33,15 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard. without emitting empty text deltas. Provider `raw_representation` is omitted; `AGUIChatClient` restores the annotations on text content. These live custom events are not replayed by `MESSAGES_SNAPSHOT`. - Inbound custom event aliases are accepted: `CUSTOM`, `CUSTOM_EVENT`, and `custom_event`. -- Multimodal user inputs support both legacy (`text`, `binary`) and draft-style (`image`, `audio`, `video`, `document`) shapes. +- Multimodal user inputs support both legacy (`text`, `binary`) and AG-UI 1.0 (`image`, `audio`, `video`, `document`) shapes. +- `ag-ui-protocol` 0.1.x and 1.x are both supported. Build every `RUN_STARTED` with `_build_run_started_event` so it + declares `protocolVersion` when the installed SDK has one. Snapshot messages keep media as `binary` parts; pass them + through `snapshot_messages_for_wire` when building a `MESSAGES_SNAPSHOT`, which rewrites them to the 1.0 shape when + 1.0 is installed (1.0 rejects `binary`). +- A 1.0 `ContentPart[]` tool result becomes the function result's Content items, so media reaches chat clients with + rich function output; text-only results stay text. In the other direction, a function result holding media goes + out as content parts on `TOOL_CALL_RESULT` and in the snapshot tool message (`_function_result_to_agui_content`), + so `AgentFrameworkAgent` relays it intact; MCP results keep their Host-payload handling. - Interrupted runs complete with `RUN_FINISHED.outcome.type == "interrupt"` and canonical `outcome.interrupts`; do not document or add new flows that depend on the legacy top-level `RUN_FINISHED.interrupt` field. - `Interrupt` and `ResumeEntry` come from the `ag-ui-protocol` package (`ag_ui.core`), not from an Agent Framework-specific interrupt model. - Tool approval interrupts, including approvals surfaced through workflow `request_info`, advertise standard diff --git a/python/packages/ag-ui/README.md b/python/packages/ag-ui/README.md index 44b6728bdc3..fff4b416061 100644 --- a/python/packages/ag-ui/README.md +++ b/python/packages/ag-ui/README.md @@ -330,7 +330,8 @@ Additional compatibility and draft support: - Native `Workflow` endpoint registration via `add_agent_framework_fastapi_endpoint(...)` - Workflow-to-AG-UI event mapping (run/step/activity/tool/custom events) - Custom event compatibility for inbound `CUSTOM`, `CUSTOM_EVENT`, and `custom_event` -- Pragmatic multimodal input parsing for both legacy (`binary`) and draft media-part shapes +- Pragmatic multimodal input parsing for both legacy (`binary`) and AG-UI 1.0 media-part shapes +- Works with `ag-ui-protocol` 0.1.x and 1.x; with 1.x, `RUN_STARTED` declares `protocolVersion` - Canonical interrupt/resume handling (`availableInterrupts`, `resume`, and `RUN_FINISHED.outcome.interrupts`) ## Security: Authentication & Authorization diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py b/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py index 436607114a9..a4d7c0d9025 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py @@ -18,7 +18,6 @@ CustomEvent, MessagesSnapshotEvent, RunErrorEvent, - RunStartedEvent, StateSnapshotEvent, TextMessageContentEvent, TextMessageEndEvent, @@ -90,10 +89,17 @@ InRunPendingToolTransitionOwner, ) from ._approval_state import _APPROVAL_SCOPE_INPUT_KEY, InMemoryAGUIApprovalStateStore, approval_state_thread_id -from ._message_adapters import _APPROVAL_DECISION_IS_BOOLEAN_KEY, normalize_agui_input_messages +from ._message_adapters import ( + _function_result_to_agui_content, + _APPROVAL_DECISION_IS_BOOLEAN_KEY, + normalize_agui_input_messages, + snapshot_messages_for_wire, +) from ._predictive_state import PredictiveStateHandler from ._tooling import collect_server_tools, merge_tools from ._run_common import ( + _UNSET, + _build_run_started_event, FlowState, _approval_interrupt_for_function_call, # type: ignore _approval_steps_response_schema, # type: ignore @@ -738,18 +744,26 @@ def _make_approval_tool_result_events(resolved_approval_results: list[Content]) llm_str = _stringify_tool_result(raw) 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) + ui_content: str | list[dict[str, Any]] = _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 = _mcp_host_history_fields( host_payload, _model_items_for_agui_replay(resolved, llm_str), ) + elif display_result is _UNSET: + # Media in the result goes out as content parts, as it does in the rebuilt snapshot message + # (_resolved_tool_result_snapshot_messages). + agui_content = _function_result_to_agui_content(resolved) + if isinstance(agui_content, list): + ui_content = agui_content events.append( ToolCallResultEvent( message_id=generate_event_id(), tool_call_id=resolved.call_id, - content=ui_str, + content=ui_content, # type: ignore[arg-type] role="tool", **replay_properties, ) @@ -766,7 +780,7 @@ def _run_start_events( resolved_approval_results: list[Content], ) -> list[BaseEvent]: """Build the events that open an agent run: RUN_STARTED, PredictState, initial state, and approval results.""" - events: list[BaseEvent] = [RunStartedEvent(run_id=run_id, thread_id=thread_id)] + events: list[BaseEvent] = [_build_run_started_event(run_id, thread_id)] if predict_state_config: predict_state_value = [ { @@ -2274,11 +2288,14 @@ def _resolved_tool_result_snapshot_messages( 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) + # A result holding media keeps it as content parts (with ag-ui-protocol 1.0); see + # _function_result_to_agui_content. Host-payload results keep their own fields below. + agui_content = llm_result if has_host_payload else _function_result_to_agui_content(content) 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": llm_result, + "content": agui_content if isinstance(agui_content, list) else llm_result, } if has_host_payload: snapshot_message.update( @@ -2419,7 +2436,9 @@ def _build_messages_snapshot( if flow.snapshot_segments: _append_segmented_snapshot_messages(flow, all_messages) bounded_messages = _bound_host_payload_history(_persistable_host_payload_history(all_messages)) - return MessagesSnapshotEvent(messages=_project_host_payload_history(bounded_messages)) # type: ignore[arg-type] + return MessagesSnapshotEvent( + messages=snapshot_messages_for_wire(_project_host_payload_history(bounded_messages)) # type: ignore[arg-type] + ) # Add assistant message with tool calls only (no content) if flow.pending_tool_calls: @@ -2454,7 +2473,7 @@ def _build_messages_snapshot( all_messages.extend(flow.reasoning_messages) bounded_messages = _bound_host_payload_history(_persistable_host_payload_history(all_messages)) - return MessagesSnapshotEvent(messages=_project_host_payload_history(bounded_messages)) # type: ignore[arg-type] + return MessagesSnapshotEvent(messages=snapshot_messages_for_wire(_project_host_payload_history(bounded_messages))) # type: ignore[arg-type] def _safe_point_tool_call_ids(flow: FlowState) -> set[str]: @@ -2545,7 +2564,7 @@ def _build_safe_point_messages_snapshot( all_messages.extend(results) bounded_messages = _bound_host_payload_history(_persistable_host_payload_history(all_messages)) - return MessagesSnapshotEvent(messages=_project_host_payload_history(bounded_messages)) # type: ignore[arg-type] + return MessagesSnapshotEvent(messages=snapshot_messages_for_wire(_project_host_payload_history(bounded_messages))) # type: ignore[arg-type] def _text_events_to_snapshot_messages(events: list[BaseEvent]) -> list[dict[str, Any]]: @@ -3118,7 +3137,7 @@ async def _run_agent_stream( resume_payload, translated_message_ids, legacy_resume_error = legacy_resume raw_messages[:] = [message for message in raw_messages if id(message) not in translated_message_ids] if legacy_resume_error is not None: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent(message=legacy_resume_error, code="APPROVAL_RESUME_INVALID") return approval_resume_messages, handled_resume_ids, cancelled_resume_ids, resume_error = ( @@ -3137,7 +3156,7 @@ async def _run_agent_stream( ) ) if resume_error is not None: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) resume_error_code = getattr(resume_error, "code", None) should_clear_tool_approval_state = resume_error_code == "APPROVAL_RESUME_CANCELLED" or ( resume_error_code == "APPROVAL_RESUME_NOT_FOUND" @@ -3188,7 +3207,7 @@ async def _run_agent_stream( # Fully handled via retained results. Reconstruction may have refilled # ``raw_messages`` with the stored transcript on a client-replayed retry; # do not fall through to a fresh agent run (#8140 / eavan review). - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) for event in _make_approval_tool_result_events(retained_approval_results): yield event yield _build_run_finished_event(run_id=run_id, thread_id=thread_id) @@ -3221,7 +3240,7 @@ async def _run_agent_stream( # Handle empty messages (emit RunStarted immediately since no agent response) if not messages and not only_cancelled_resume: logger.warning("No messages provided in AG-UI input") - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield _build_run_finished_event(run_id=run_id, thread_id=thread_id) return @@ -3380,7 +3399,7 @@ async def save_thread_snapshot( ) if only_cancelled_resume: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) _clear_tool_approval_state(approval_state_store, approval_thread_id) retired_interrupt_ids = { reconciliation.interrupt_id @@ -3480,7 +3499,7 @@ async def save_thread_snapshot( for intent in authorized_executions.values(): approval_state_store.lifecycle.release_claim(intent, policy=ClaimRecoveryPolicy.SAFE_TO_RETRY) unavailable_names = ", ".join(sorted({intent.name for intent in unavailable_local_intents})) - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message=f"Approved tool(s) {unavailable_names} are temporarily unavailable; retry the approval later.", code="APPROVAL_TOOL_UNAVAILABLE", @@ -3527,7 +3546,7 @@ async def save_thread_snapshot( } if retired_interrupt_ids: await snapshot_session.clear_interrupts(interrupt_ids=retired_interrupt_ids) - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message="Function invocation is disabled; the approved tool remains pending for an explicit retry.", code="APPROVAL_INVOCATION_DISABLED", @@ -3600,7 +3619,7 @@ def merge_snapshot_interrupts( return combined or None if replacement_approval_requests: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) for request in replacement_approval_requests: for event in _emit_content( request, @@ -3659,7 +3678,7 @@ def merge_snapshot_interrupts( confirm_remaining_interrupts = remaining_stored_interrupts( {str(confirm_interrupt_id)} if confirm_interrupt_id else set() ) - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) # Emit approved state snapshot before confirmation message if approved_state_snapshot_emitted: yield StateSnapshotEvent(snapshot=flow.current_state) @@ -3919,7 +3938,7 @@ async def save_safe_point_snapshot() -> None: stream_completed = True except ApprovalInvocationDisabledError: if not run_started_emitted: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message="Function invocation is disabled; the approved tool remains pending for an explicit retry.", code="APPROVAL_INVOCATION_DISABLED", diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py b/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py index 08b3614a37b..b0f7cc536c3 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py @@ -16,6 +16,8 @@ Message, ) +from ._message_adapters import _tool_content_parts_to_result + logger = logging.getLogger(__name__) @@ -261,6 +263,11 @@ def _handle_tool_call_result(self, event: dict[str, Any]) -> ChatResponseUpdate: """Handle TOOL_CALL_RESULT event.""" tool_call_id = self._get_tool_call_id(event) or "" result = event.get("result") if event.get("result") is not None else event.get("content") + if isinstance(result, list): + # An AG-UI 1.0 ContentPart[] result: pass media on as Content items, as for tool messages. + converted = _tool_content_parts_to_result(cast(list[Any], result), tool_call_id) + if converted is not None: + result = converted return ChatResponseUpdate( role="tool", diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py b/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py index 41adb4a7d2d..bc3bb38bcf4 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py @@ -23,6 +23,7 @@ from ._utils import ( _AGUI_HOST_PAYLOAD_OMITTED_KEY, _AGUI_MCP_TOOL_RESULT_KEY, + _AGUI_PROTOCOL_VERSION, _AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY, _AGUI_TOOL_RESULT_MODEL_CONTENT_KEY, _MAX_MCP_HOST_PAYLOAD_HISTORY_SIZE_BYTES, @@ -342,14 +343,14 @@ def _deduplicate_messages(messages: list[Message]) -> list[Message]: if open_result_slots[call_id] > 0: open_result_slots[call_id] -= 1 elif call_id in declared_call_ids or kept_indexes: - new_result = msg.contents[0].result + new_has_output = _function_result_has_output(msg.contents[0]) for existing_idx in kept_indexes: existing_msg = unique_messages[existing_idx] - existing_result = None + existing_empty = True if existing_msg.contents and existing_msg.contents[0].type == "function_result": - existing_result = existing_msg.contents[0].result + existing_empty = not _function_result_has_output(existing_msg.contents[0]) - if (not existing_result or existing_result == "") and new_result: + if existing_empty and new_has_output: logger.info(f"Replacing empty tool result at index {existing_idx} with data from index {idx}") unique_messages[existing_idx] = msg break @@ -427,7 +428,9 @@ def _extract_multimodal_source_fields( elif source_type in {"base64", "data", "binary"}: data = cast(str | None, source_dict.get("value") or source_dict.get("data")) elif source_type in {"id", "file"}: - binary_id = cast(str | None, source_dict.get("id")) + # An AG-UI 1.0 file source carries its provider-issued handle in ``value``; like an ``id`` reference, + # it is unresolved and left to application-owned policy. + binary_id = cast(str | None, source_dict.get("id") or source_dict.get("value")) else: url = cast(str | None, source_dict.get("url") or source_dict.get("uri") or url) data = cast(str | None, source_dict.get("value") or source_dict.get("data") or data) @@ -496,6 +499,61 @@ def _parse_multimodal_media_part(part: dict[str, Any]) -> Content | None: return None +def _snapshot_media_part(part: dict[str, Any]) -> dict[str, Any] | None: + """Convert a legacy ``binary`` part to the AG-UI 1.0 media part shape (``image``/``audio``/``video``/``document``). + + AG-UI 1.0 removed the ``binary`` part, so a snapshot holding one fails validation. The part is mapped to the + media type its MIME type names. Returns ``None`` for a part that only references a binary ``id``, which has no + 1.0 representation. + """ + url, data, _, mime_type = _extract_multimodal_source_fields(part) + top_level_type = (mime_type or "").split("/", 1)[0].lower() + part_type = top_level_type if top_level_type in {"image", "audio", "video"} else "document" + + source: dict[str, Any] + if isinstance(url, str) and url: + source = {"type": "url", "value": url} + elif isinstance(data, str) and data: + source = {"type": "url", "value": data} if data.startswith("data:") else {"type": "data", "value": data} + if source["type"] == "data" and not mime_type: + mime_type = "application/octet-stream" + else: + logger.warning("Omitting an AG-UI %s part without a url or data source from MESSAGES_SNAPSHOT.", part_type) + return None + if mime_type: + source["mimeType"] = mime_type + return {"type": part_type, "source": source} + + +def snapshot_messages_for_wire(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Rewrite legacy ``binary`` parts in snapshot messages to the AG-UI 1.0 media part shape. + + Snapshot messages keep media as ``binary`` parts (and thread snapshots saved earlier hold them too), which a + 1.0 ``MessagesSnapshotEvent`` rejects. Every snapshot event passes its messages through here, so media + replays with ag-ui-protocol 1.0; before 1.0 the messages are returned unchanged. + """ + if _AGUI_PROTOCOL_VERSION is None: + return messages + upgraded: list[dict[str, Any]] = [] + for message in messages: + content = message.get("content") + if not isinstance(content, list) or not any( + isinstance(part, dict) and cast(dict[str, Any], part).get("type") == "binary" + for part in cast(list[Any], content) + ): + upgraded.append(message) + continue + parts: list[Any] = [] + for part in cast(list[Any], content): + if isinstance(part, dict) and cast(dict[str, Any], part).get("type") == "binary": + if (media_part := _snapshot_media_part(cast(dict[str, Any], part))) is not None: + parts.append(media_part) + else: + parts.append(part) + upgraded.append({**message, "content": parts}) + return upgraded + + def sanitize_legacy_snapshot_media( messages: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], int]: @@ -656,6 +714,43 @@ def _legacy_binary_part(part: dict[str, Any]) -> dict[str, Any]: return content +def _tool_content_parts_to_result(content: list[Any], tool_call_id: str) -> str | list[Content] | None: + """Convert an AG-UI ``ContentPart[]`` tool result to a function result. + + Text-only content becomes its concatenated text. When media parts are present, every part becomes a Content + item (media parts the same way as user media), so chat clients with rich function output receive the media. + Parts that cannot be represented are dropped with a warning. Returns ``None`` when ``content`` is not a list + of content parts, so other list payloads keep their existing handling. + """ + # "binary" is the shape stored snapshots keep tool media in, so replayed history converts the same way. + content_part_types = {"text", "image", "audio", "video", "document", "binary"} + if not content or not all( + isinstance(part, dict) and cast(dict[str, Any], part).get("type") in content_part_types for part in content + ): + return None + parts = cast(list[dict[str, Any]], content) + if all(part["type"] == "text" for part in parts): + return "".join(str(part.get("text", "")) for part in parts) + + items: list[Content] = [] + dropped: list[str] = [] + for part in parts: + if part["type"] == "text": + items.append(Content.from_text(text=str(part.get("text", "")))) + elif (media_content := _parse_multimodal_media_part(part)) is not None: + items.append(media_content) + else: + dropped.append(str(part["type"])) + if dropped: + logger.warning( + "Dropping %d media part(s) (%s) without a usable source from the result of tool call %r.", + len(dropped), + ", ".join(dropped), + tool_call_id, + ) + return items or "" + + def normalize_agui_input_messages( messages: list[dict[str, Any]], *, @@ -823,6 +918,10 @@ def _filter_modified_args( result_content = msg.get("content") if result_content is None: result_content = msg.get("result", "") + elif isinstance(result_content, list): + converted = _tool_content_parts_to_result(cast(list[Any], result_content), str(tool_call_id)) + if converted is not None: + result_content = converted # Distinguish approval payloads from actual tool results parsed: dict[str, Any] | None = None @@ -1139,6 +1238,36 @@ def _convert_framework_content_to_agui(content: Content) -> dict[str, Any] | Non return {"type": part_type, "source": source} +def _function_result_has_output(content: Content) -> bool: + """Whether a function result carries any output: text, or media items (an image-only result has no text).""" + if content.result: + return True + return any(item.type in {"uri", "data"} for item in content.items or []) + + +def _function_result_to_agui_content(content: Content) -> str | list[dict[str, Any]]: + """Encode a function result as AG-UI tool message content. + + A result holding media items becomes AG-UI 1.0 content parts, so media received from a tool result survives + into the next request's history; that needs ag-ui-protocol 1.0, since earlier tool messages only take text. + Any other result is its text, as before. MCP results keep their own handling: their model content travels in + a sidecar that host-history budgeting may deliberately omit. + """ + text = content.result if content.result is not None else "" + items = content.items or [] + if _AGUI_PROTOCOL_VERSION is None or not any(item.type in {"uri", "data"} for item in items): + return text + if _extract_mcp_tool_result_host_payload(content)[0]: + return text + parts: list[dict[str, Any]] = [] + for item in items: + if item.type == "text": + parts.append({"type": "text", "text": item.text or ""}) + elif (media_part := _convert_framework_content_to_agui(item)) is not None: + parts.append(media_part) + return parts or text + + def _encode_agui_segment(contents: list[Content], role: str) -> tuple[str | list[dict[str, Any]], list[dict[str, Any]]]: """Encode a framework content segment into AG-UI message content and tool calls. @@ -1256,7 +1385,7 @@ def emit_result(content: Content) -> None: tool_message: dict[str, Any] = { "id": next_id(), "role": "tool", - "content": content.result if content.result is not None else "", + "content": _function_result_to_agui_content(content), "toolCallId": content.call_id, } messages.append(tool_message) diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_predictive_state.py b/python/packages/ag-ui/agent_framework_ag_ui/_predictive_state.py index 99c18b9e45c..04a487bbaf6 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_predictive_state.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_predictive_state.py @@ -215,15 +215,9 @@ def _create_delta_event(self, state_key: str, value: Any) -> StateDeltaEvent: elif self.state_delta_count % 100 == 0: logger.info(f"StateDeltaEvent #{self.state_delta_count} emitted") - return StateDeltaEvent( - delta=[ - { - "op": "replace", - "path": f"/{state_key}", - "value": value, - } - ], - ) + # A JSON Patch as plain operation dicts: ag-ui-protocol 1.0 validates them into operation models. + delta: list[Any] = [{"op": "replace", "path": f"/{state_key}", "value": value}] + return StateDeltaEvent(delta=delta) def apply_pending_updates(self) -> None: """Apply pending updates to current state and clear them.""" diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py index edf6680ea2d..9e4ab409058 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py @@ -24,6 +24,7 @@ ReasoningStartEvent, RunFinishedEvent, RunFinishedInterruptOutcome, + RunStartedEvent, StateSnapshotEvent, TextMessageContentEvent, TextMessageEndEvent, @@ -35,9 +36,11 @@ ) from agent_framework import Content, ResponseStream +from ._message_adapters import _function_result_to_agui_content from ._predictive_state import PredictiveStateHandler from ._state import TOOL_RESULT_DISPLAY_KEY, TOOL_RESULT_STATE_KEY from ._utils import ( + _AGUI_PROTOCOL_VERSION, _approval_interrupt_id, _extract_mcp_tool_result_host_payload, _extract_tool_result_marker_values, @@ -505,6 +508,18 @@ def _approval_interrupt_for_function_call( } +def _build_run_started_event(run_id: str, thread_id: str) -> RunStartedEvent: + """Create a RUN_STARTED event that declares the AG-UI protocol version this producer speaks. + + AG-UI 1.0 requires a producer to declare ``protocolVersion`` on the event that opens a run, and the + ``ag-ui-protocol`` SDK leaves it unset by default. Releases before 1.0 have no such field, so nothing is + declared there. Every RUN_STARTED is built here so no emitter can forget the declaration. + """ + if _AGUI_PROTOCOL_VERSION is None: + return RunStartedEvent(run_id=run_id, thread_id=thread_id) + return RunStartedEvent(run_id=run_id, thread_id=thread_id, protocol_version=_AGUI_PROTOCOL_VERSION) + + def _build_run_finished_event( run_id: str, thread_id: str, interrupts: list[dict[str, Any]] | None = None ) -> RunFinishedEvent: @@ -648,9 +663,7 @@ def _emit_text(content: Content, flow: FlowState, skip_text: bool = False) -> li flow.text_delta_count = 0 _open_text_segment(flow, flow.message_id) events.append(TextMessageStartEvent(message_id=flow.message_id, role="assistant")) - elif flow.accumulated_text and content.text == flow.accumulated_text and ( - flow.text_delta_count > 1 - ): + elif flow.accumulated_text and content.text == flow.accumulated_text and (flow.text_delta_count > 1): # Guard against full-message replay chunks that can appear after streaming deltas. # A duplicate equal to the whole accumulated text is only a safe replay signal when # the text already streamed in more than one chunk. With a single prior chunk the @@ -847,12 +860,17 @@ def _emit_tool_result_common( display_result: Any = _UNSET, # noqa: ANN401 snapshot_result: Any = _UNSET, # noqa: ANN401 model_items: list[dict[str, Any]] | None = None, + content_parts: list[dict[str, Any]] | None = None, ) -> list[BaseEvent]: """Shared helper for emitting ToolCallResult (and ToolCallEnd when still open). Both ``_emit_tool_result`` (standard function results) and ``_emit_mcp_tool_result`` (MCP server tool results) delegate to this function. + ``content_parts`` is the result as AG-UI 1.0 content parts when it holds media; the + TOOL_CALL_RESULT event (unless a display payload overrides it) and the snapshot tool + message then carry the parts, so the media survives the relay and the next request. + ``TOOL_CALL_END`` is emitted only when ``call_id`` was opened in this run and has not already been ended (including synthetic protocol closure before an approval interrupt). On resume, a fresh ``FlowState`` never STARTs that id, @@ -873,15 +891,24 @@ def _emit_tool_result_common( flow.tool_calls_ended.add(call_id) result_content = _stringify_tool_result(raw_result) - ui_result_content = _resolve_ui_payload(result_content, display_result) + ui_result_content: str | list[dict[str, Any]] = _resolve_ui_payload(result_content, display_result) + if content_parts is not None and display_result is _UNSET: + ui_result_content = content_parts snapshot_result_content = _resolve_ui_payload(result_content, snapshot_result) message_id = generate_event_id() + event_replay_properties: dict[str, Any] = {} + if snapshot_result is not _UNSET: + event_replay_properties = _mcp_host_history_fields( + snapshot_result_content, + [{"type": "text", "text": result_content}] if model_items is None else model_items, + ) events.append( ToolCallResultEvent( message_id=message_id, tool_call_id=call_id, - content=ui_result_content, + content=ui_result_content, # type: ignore[arg-type] role="tool", + **event_replay_properties, ) ) @@ -889,22 +916,9 @@ def _emit_tool_result_common( "id": message_id, "role": "tool", "toolCallId": call_id, - "content": result_content, + "content": result_content if content_parts is None else content_parts, + **event_replay_properties, } - event_replay_properties: dict[str, Any] = {} - if snapshot_result is not _UNSET: - event_replay_properties = _mcp_host_history_fields( - snapshot_result_content, - [{"type": "text", "text": result_content}] if model_items is None else model_items, - ) - snapshot_message.update(event_replay_properties) - events[-1] = ToolCallResultEvent( - message_id=message_id, - tool_call_id=call_id, - content=ui_result_content, - role="tool", - **event_replay_properties, - ) flow.tool_results.append(snapshot_message) # A result closes the current tool-call segment: a later call opens a new # one, so `call A -> result A -> call B` snapshots as two call/result pairs @@ -954,6 +968,9 @@ def _emit_tool_result( has_host_payload, host_payload = _resolve_tool_result_host_payload(content, display_result) if has_host_payload and display_result is _UNSET: display_result = host_payload + # A result holding media is relayed as content parts (with ag-ui-protocol 1.0); MCP results keep their own + # handling, and _function_result_to_agui_content returns text for them. + agui_content = raw_result if has_host_payload else _function_result_to_agui_content(content) return _emit_tool_result_common( content.call_id, raw_result, @@ -965,6 +982,7 @@ def _emit_tool_result( model_items=( _model_items_for_agui_replay(content, _stringify_tool_result(raw_result)) if has_host_payload else None ), + content_parts=agui_content if isinstance(agui_content, list) else None, ) diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py b/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py index 85546830245..a1275ee8349 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py @@ -18,12 +18,11 @@ from ag_ui.core import ( BaseEvent, MessagesSnapshotEvent, - RunStartedEvent, StateSnapshotEvent, ) -from ._message_adapters import sanitize_legacy_snapshot_media -from ._run_common import _build_run_finished_event +from ._message_adapters import sanitize_legacy_snapshot_media, snapshot_messages_for_wire +from ._run_common import _build_run_finished_event, _build_run_started_event from ._snapshots import ( AGUIThreadSnapshot, AGUIThreadSnapshotStore, @@ -109,7 +108,7 @@ def rebind_thread_id(self, thread_id: str) -> None: async def hydrate_events(self, *, run_id: str) -> AsyncGenerator[BaseEvent]: """Replay the stored snapshot as a complete run without invoking the agent.""" - yield RunStartedEvent(run_id=run_id, thread_id=self._thread_id) + yield _build_run_started_event(run_id, self._thread_id) snapshot = self._stored if snapshot is None: yield _build_run_finished_event(run_id=run_id, thread_id=self._thread_id) @@ -118,7 +117,9 @@ async def hydrate_events(self, *, run_id: str) -> AsyncGenerator[BaseEvent]: if snapshot.state is not None: yield StateSnapshotEvent(snapshot=snapshot.state) if snapshot.messages: - yield MessagesSnapshotEvent(messages=_project_host_payload_history(snapshot.messages)) # type: ignore[arg-type] + yield MessagesSnapshotEvent( + messages=snapshot_messages_for_wire(_project_host_payload_history(snapshot.messages)) # type: ignore[arg-type] + ) yield _build_run_finished_event(run_id=run_id, thread_id=self._thread_id, interrupts=snapshot.interrupt) def effective_state( diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_utils.py b/python/packages/ag-ui/agent_framework_ag_ui/_utils.py index 78ab5cf146b..d518018d87f 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_utils.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_utils.py @@ -10,9 +10,17 @@ from collections.abc import Callable, MutableMapping, Sequence from typing import Any +import ag_ui.core as _agui_core from agent_framework import AgentResponseUpdate, ChatResponseUpdate, Content, FunctionTool from agent_framework._serialization import make_json_safe +_AGUI_PROTOCOL_VERSION: str | None = getattr(_agui_core, "PROTOCOL_VERSION", None) +"""The AG-UI protocol version the installed ``ag-ui-protocol`` speaks, or ``None`` before 1.0. + +Releases before 1.0 have no version to declare and still accept the legacy ``binary`` part, which 1.0 rejects, +so code that emits version-specific shapes gates on this value. +""" + def _mcp_tool_result_host_payload_key(core_mcp: Any) -> str: """Resolve the private core marker while supporting older core packages.""" diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py index 4b56f672098..cc1c88fad95 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py @@ -22,7 +22,6 @@ ReasoningStartEvent, RunErrorEvent, RunFinishedEvent, - RunStartedEvent, StateSnapshotEvent, TextMessageContentEvent, TextMessageEndEvent, @@ -33,6 +32,7 @@ ) from agent_framework import CheckpointID, CheckpointStorage, Content, Message, Workflow, WorkflowCheckpoint from agent_framework._telemetry import mark_feature_used +from pydantic import BaseModel from ._feature_usage import FeatureIndex from ._message_adapters import ( @@ -41,6 +41,7 @@ normalize_agui_input_messages, ) from ._run_common import ( + _build_run_started_event, _cancelled_resume_interrupt_ids, _extract_resume_payload, _is_snapshot_hydration_request, @@ -435,7 +436,8 @@ def _observe_text_start(self, event: TextMessageStartEvent) -> None: self._flush_open_reasoning_message() if self._open_text_message is not None and self._open_text_message.get("id") != event.message_id: self._flush_open_text_message() - self._open_text_message = {"id": event.message_id, "role": event.role, "content": ""} + # ag-ui-protocol 1.0 no longer defaults the role to "assistant"; a stored message needs one to replay. + self._open_text_message = {"id": event.message_id, "role": event.role or "assistant", "content": ""} def _observe_text_content(self, event: TextMessageContentEvent) -> None: if self._open_text_message is None or self._open_text_message.get("id") != event.message_id: @@ -500,7 +502,14 @@ def _observe_tool_call_result(self, event: ToolCallResultEvent) -> None: "id": event.message_id, "role": "tool", "toolCallId": event.tool_call_id, - "content": event.content, + # ag-ui-protocol 1.0 parses content parts into models; store them as wire dictionaries so the + # snapshot serializes and replays like any other stored message. + "content": [ + part.model_dump(by_alias=True, exclude_none=True) if isinstance(part, BaseModel) else part + for part in cast(list[Any], event.content) + ] + if isinstance(event.content, list) + else event.content, } if getattr(event, _AGUI_MCP_TOOL_RESULT_KEY, False) is True: message[_AGUI_MCP_TOOL_RESULT_KEY] = True @@ -799,7 +808,7 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]: ): # A factory run that omits threadId reports a new generated id, so nothing can be pending on it and a # resume entry, tool result, or approval response cannot continue anything. - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message="No pending interrupt found for this AG-UI thread.", code="WORKFLOW_RESUME_NOT_FOUND", @@ -827,7 +836,7 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]: for request_event in live_pending_events.values(): owner = getattr(request_event, _REQUEST_OWNER_ATTRIBUTE, None) if owner != live_owner and (owner is not None or request_owner != (None, None)): - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message="No pending interrupt found for this AG-UI thread.", code="WORKFLOW_RESUME_NOT_FOUND", @@ -839,14 +848,14 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]: raise LookupError(f"checkpoint '{checkpoint_id}' was not found") checkpoint = await owned_checkpoint_storage.load(checkpoint_id) except PermissionError: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message=f"No pending interrupt found for checkpointId '{checkpoint_id}'.", code="WORKFLOW_RESUME_NOT_FOUND", ) return except Exception as exc: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message=f"Could not load workflow checkpoint '{checkpoint_id}': {exc}", code="WORKFLOW_CHECKPOINT_LOAD_FAILED", @@ -862,7 +871,7 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]: request_event = live_pending_events.get(interrupt_id) owner = getattr(request_event, _REQUEST_OWNER_ATTRIBUTE, None) if owner != live_owner and (owner is not None or request_owner != (None, None)): - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent( message=f"No pending interrupt found for resume interruptId '{interrupt_id}'.", code="WORKFLOW_RESUME_NOT_FOUND", diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py b/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py index 639fc223b25..33167eee194 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py @@ -18,7 +18,6 @@ BaseEvent, CustomEvent, RunErrorEvent, - RunStartedEvent, StepFinishedEvent, StepStartedEvent, TextMessageEndEvent, @@ -50,6 +49,7 @@ _approval_interrupt_for_function_call, # pyright: ignore[reportPrivateUsage] _approval_response_schema, # pyright: ignore[reportPrivateUsage] _build_run_finished_event, + _build_run_started_event, _close_reasoning_block, _emit_content, _extract_resume_payload, @@ -1224,12 +1224,12 @@ async def run_workflow_stream( missing_code="WORKFLOW_RESUME_MISSING_INTERRUPT", ) if contract_error is not None and contract_code is not None: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield RunErrorEvent(message=contract_error, code=contract_code) return resume_error = _resume_error_for_pending_workflow_requests(resume_entries) if resume_error is not None: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield resume_error return cancelled_request_ids = { @@ -1251,7 +1251,7 @@ async def run_workflow_stream( responses, pending_before_run, on_conversational_responses=on_conversational_responses ) if response_error is not None: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield response_error return if cancelled_request_ids: @@ -1273,7 +1273,7 @@ async def run_workflow_stream( # checkpoint. ``pending_before_run`` reflects the live (pre-restore) instance, so # short-circuiting on it here would skip the restore entirely. if checkpoint_id is None and not responses and pending_before_run: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) for request_event in pending_before_run.values(): request_payload = _request_payload_from_request_event(request_event) if request_payload is None: @@ -1299,7 +1299,7 @@ async def run_workflow_stream( return if checkpoint_id is None and not responses and not messages: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) yield _build_run_finished_event( run_id=run_id, thread_id=thread_id, @@ -1413,12 +1413,12 @@ def _drain_open_blocks() -> list[BaseEvent]: if event_type == "started": if not run_started_emitted: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) run_started_emitted = True continue if not run_started_emitted: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) run_started_emitted = True if event_type == "failed": @@ -1586,7 +1586,7 @@ def _drain_open_blocks() -> list[BaseEvent]: failure_event = None logger.exception("Workflow AG-UI stream failed: %s", exc) if not run_started_emitted: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) run_started_emitted = True # Close any open reasoning block / text message so RUN_ERROR stays the final event. for end_event in _drain_open_blocks(): @@ -1615,7 +1615,7 @@ def _drain_open_blocks() -> list[BaseEvent]: yield end_event if not run_started_emitted: - yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield _build_run_started_event(run_id, thread_id) if not terminal_emitted and not run_error_emitted: if not interrupts: diff --git a/python/packages/ag-ui/agent_framework_ag_ui_examples/agents/task_steps_agent.py b/python/packages/ag-ui/agent_framework_ag_ui_examples/agents/task_steps_agent.py index b3c02e57a0b..5c14535279f 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui_examples/agents/task_steps_agent.py +++ b/python/packages/ag-ui/agent_framework_ag_ui_examples/agents/task_steps_agent.py @@ -25,6 +25,9 @@ from pydantic import BaseModel, Field from agent_framework_ag_ui import AgentFrameworkWorkflow +from agent_framework_ag_ui._message_adapters import ( + snapshot_messages_for_wire, +) class StepStatus(str, Enum): @@ -149,7 +152,9 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[Any]: case StateDeltaEvent(delta=delta): # Apply state delta to final_state if delta: - for patch in delta: + for operation in delta: + # ag-ui-protocol 1.0 parses JSON Patch operations into models; 0.1.x keeps dicts. + patch = operation.model_dump() if isinstance(operation, BaseModel) else operation if patch.get("op") == "replace" and patch.get("path") == "/steps": final_state["steps"] = patch.get("value", []) logger.info( @@ -291,9 +296,10 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[Any]: final_messages.append(summary_message) # Emit MessagesSnapshotEvent to persist in history + # Legacy binary attachments in the history must take the AG-UI 1.0 media shape first. yield MessagesSnapshotEvent( type=EventType.MESSAGES_SNAPSHOT, - messages=final_messages, + messages=snapshot_messages_for_wire(final_messages), ) except Exception as e: logger.error(f"Error generating summary: {e}") diff --git a/python/packages/ag-ui/pyproject.toml b/python/packages/ag-ui/pyproject.toml index a328a6d1c0f..9207cd46e67 100644 --- a/python/packages/ag-ui/pyproject.toml +++ b/python/packages/ag-ui/pyproject.toml @@ -25,7 +25,7 @@ classifiers = [ ] dependencies = [ "agent-framework-core>=1.22.0,<2", - "ag-ui-protocol>=0.1.19,<0.2", + "ag-ui-protocol>=0.1.19,<2", "fastapi>=0.121.0,<0.142.0", "httpx>=0.28.1,<1", "sse-starlette>=3.4.5,<4", diff --git a/python/packages/ag-ui/tests/ag_ui/test_predictive_state.py b/python/packages/ag-ui/tests/ag_ui/test_predictive_state.py index 5d9a5ec3417..c660967b5a7 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_predictive_state.py +++ b/python/packages/ag-ui/tests/ag_ui/test_predictive_state.py @@ -2,11 +2,18 @@ """Tests for predictive state handling.""" +from typing import Any + from ag_ui.core import StateDeltaEvent from agent_framework_ag_ui._predictive_state import PredictiveStateHandler +def _first_op(event: StateDeltaEvent) -> dict[str, Any]: + """Return the event's first JSON Patch operation as a dict, however ag-ui-protocol models it.""" + return event.model_dump()["delta"][0] + + class TestPredictiveStateHandlerInit: """Tests for PredictiveStateHandler initialization.""" @@ -155,9 +162,9 @@ def test_emits_delta_on_complete_json(self): events = handler.emit_streaming_deltas("write", '{"text": "hello"}') assert len(events) == 1 assert isinstance(events[0], StateDeltaEvent) - assert events[0].delta[0]["path"] == "/doc" - assert events[0].delta[0]["value"] == "hello" - assert events[0].delta[0]["op"] == "replace" + assert _first_op(events[0])["path"] == "/doc" + assert _first_op(events[0])["value"] == "hello" + assert _first_op(events[0])["op"] == "replace" def test_emits_delta_on_partial_json(self): """Emits delta from partial JSON using regex.""" @@ -165,7 +172,7 @@ def test_emits_delta_on_partial_json(self): # First chunk - partial events = handler.emit_streaming_deltas("write", '{"text": "hel') assert len(events) == 1 - assert events[0].delta[0]["value"] == "hel" + assert _first_op(events[0])["value"] == "hel" def test_does_not_emit_duplicate_deltas(self): """Does not emit delta when value unchanged.""" @@ -190,7 +197,7 @@ def test_emits_delta_on_value_change(self): handler.streaming_tool_args = "" events2 = handler.emit_streaming_deltas("write", '{"text": "world"}') assert len(events2) == 1 - assert events2[0].delta[0]["value"] == "world" + assert _first_op(events2[0])["value"] == "world" def test_tracks_pending_updates(self): """Tracks pending state updates.""" @@ -208,7 +215,7 @@ def test_unescapes_newlines(self): handler.streaming_tool_args = '{"text": "line1\\nline2' events = handler._emit_partial_deltas("write") assert len(events) == 1 - assert events[0].delta[0]["value"] == "line1\nline2" + assert _first_op(events[0])["value"] == "line1\nline2" def test_handles_escaped_quotes_partially(self): """Handles escaped quotes - regex stops at quote character.""" @@ -225,7 +232,7 @@ def test_handles_escaped_quotes_partially(self): # After .replace("\\\\", "\\") -> "say \" # After .replace('\\"', '"') -> "say " (but actually still "say \" due to order) # The actual result: backslash is preserved since it's not a valid escape sequence - assert events[0].delta[0]["value"] == "say \\" + assert _first_op(events[0])["value"] == "say \\" def test_unescapes_backslashes(self): """Unescapes \\\\ in partial values.""" @@ -233,7 +240,7 @@ def test_unescapes_backslashes(self): handler.streaming_tool_args = '{"text": "path\\\\to\\\\file' events = handler._emit_partial_deltas("write") assert len(events) == 1 - assert events[0].delta[0]["value"] == "path\\to\\file" + assert _first_op(events[0])["value"] == "path\\to\\file" class TestEmitCompleteDeltas: @@ -244,7 +251,7 @@ def test_emits_for_matching_tool(self): handler = PredictiveStateHandler(predict_state_config={"doc": {"tool": "write", "tool_argument": "text"}}) events = handler._emit_complete_deltas("write", {"text": "content"}) assert len(events) == 1 - assert events[0].delta[0]["value"] == "content" + assert _first_op(events[0])["value"] == "content" def test_skips_non_matching_tool(self): """Skips tools not matching config.""" @@ -258,7 +265,7 @@ def test_handles_wildcard_argument(self): args = {"key1": "val1", "key2": "val2"} events = handler._emit_complete_deltas("update", args) assert len(events) == 1 - assert events[0].delta[0]["value"] == args + assert _first_op(events[0])["value"] == args def test_skips_missing_argument(self): """Skips when tool_argument not in args.""" @@ -276,9 +283,9 @@ def test_creates_event(self): event = handler._create_delta_event("key", "value") assert isinstance(event, StateDeltaEvent) - assert event.delta[0]["op"] == "replace" - assert event.delta[0]["path"] == "/key" - assert event.delta[0]["value"] == "value" + assert _first_op(event)["op"] == "replace" + assert _first_op(event)["path"] == "/key" + assert _first_op(event)["value"] == "value" def test_increments_count(self): """Increments state_delta_count.""" diff --git a/python/packages/ag-ui/tests/ag_ui/test_protocol_1_0.py b/python/packages/ag-ui/tests/ag_ui/test_protocol_1_0.py new file mode 100644 index 00000000000..f2939260c0e --- /dev/null +++ b/python/packages/ag-ui/tests/ag_ui/test_protocol_1_0.py @@ -0,0 +1,424 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""AG-UI 1.0 behaviour that holds on every supported ag-ui-protocol release.""" + +import json +from typing import Any + +import httpx +import pytest +from ag_ui.core import ( + EventType, + MessagesSnapshotEvent, + TextMessageContentEvent, + TextMessageEndEvent, + TextMessageStartEvent, + ToolCallResultEvent, + ToolCallStartEvent, +) +from agent_framework import Agent, AgentResponseUpdate, Content, Message + +from agent_framework_ag_ui import AgentFrameworkAgent, AGUIChatClient +from agent_framework_ag_ui._agent_run import _resolved_tool_result_snapshot_messages +from agent_framework_ag_ui._event_converters import AGUIEventConverter +from agent_framework_ag_ui._message_adapters import ( + _deduplicate_messages, + agent_framework_messages_to_agui, + agui_messages_to_agent_framework, + snapshot_messages_for_wire, +) +from agent_framework_ag_ui._run_common import _build_run_started_event +from agent_framework_ag_ui._snapshot_session import ThreadSnapshotSession +from agent_framework_ag_ui._snapshots import AGUIThreadSnapshot +from agent_framework_ag_ui._utils import _AGUI_PROTOCOL_VERSION +from agent_framework_ag_ui._workflow import _WorkflowSnapshotBuilder + +requires_1_0 = pytest.mark.skipif(_AGUI_PROTOCOL_VERSION is None, reason="needs ag-ui-protocol >= 1.0") + + +def test_run_started_declares_the_protocol_version() -> None: + """RUN_STARTED declares the installed protocol version; releases before 1.0 have none to declare.""" + event = _build_run_started_event("run-1", "thread-1").model_dump(by_alias=True, exclude_none=True) + + assert event["type"] == EventType.RUN_STARTED + if _AGUI_PROTOCOL_VERSION is None: + assert "protocolVersion" not in event + else: + assert event["protocolVersion"] == _AGUI_PROTOCOL_VERSION + + +async def test_thread_saved_with_legacy_media_replays() -> None: + """A thread snapshot holding legacy ``binary`` media replays with whichever ag-ui-protocol is installed.""" + stored = AGUIThreadSnapshot( + messages=[ + { + "id": "u1", + "role": "user", + "content": [ + {"type": "text", "text": "look"}, + {"type": "binary", "url": "https://example.com/cat.png", "mimeType": "image/png"}, + ], + } + ] + ) + session = ThreadSnapshotSession(store=None, scope=None, thread_id="t1", stored=stored) + + events = [event async for event in session.hydrate_events(run_id="r1")] + + assert events[0].type == EventType.RUN_STARTED + snapshot = next(event for event in events if isinstance(event, MessagesSnapshotEvent)) + content = snapshot.messages[0].model_dump(by_alias=True, exclude_none=True)["content"] + if _AGUI_PROTOCOL_VERSION is None: + assert content[1]["type"] == "binary" + else: + assert content[1] == { + "type": "image", + "source": {"type": "url", "value": "https://example.com/cat.png", "mimeType": "image/png"}, + } + + +@requires_1_0 +def test_wire_snapshot_maps_legacy_media_by_mime_type() -> None: + """Legacy media becomes the 1.0 part its MIME type names; a bare id reference has no 1.0 form.""" + messages: list[dict[str, Any]] = [ + { + "id": "u1", + "role": "user", + "content": [ + {"type": "binary", "data": "YXVkaW8=", "mimeType": "audio/wav"}, + {"type": "binary", "url": "https://example.com/report.pdf", "mimeType": "application/pdf"}, + {"type": "binary", "id": "upload-1", "mimeType": "image/png"}, + {"type": "text", "text": "after"}, + ], + }, + {"id": "a1", "role": "assistant", "content": "unchanged"}, + ] + + wire = snapshot_messages_for_wire(messages) + + assert wire[0]["content"] == [ + {"type": "audio", "source": {"type": "data", "value": "YXVkaW8=", "mimeType": "audio/wav"}}, + { + "type": "document", + "source": {"type": "url", "value": "https://example.com/report.pdf", "mimeType": "application/pdf"}, + }, + {"type": "text", "text": "after"}, + ] + assert wire[1] is messages[1] + assert messages[0]["content"][0]["type"] == "binary", "the stored messages are not modified" + + +def _tool_turn(content: Any) -> list[dict[str, Any]]: + return [ + { + "id": "a1", + "role": "assistant", + "content": "", + "toolCalls": [{"id": "call-1", "type": "function", "function": {"name": "chart", "arguments": "{}"}}], + }, + {"id": "t1", "role": "tool", "toolCallId": "call-1", "content": content}, + ] + + +def _function_result(messages: list[dict[str, Any]]) -> Any: + converted = agui_messages_to_agent_framework(messages) + tool_message = next(message for message in converted if message.role == "tool") + return tool_message.contents[0] + + +def test_tool_result_media_reaches_the_function_result() -> None: + """Media in a 1.0 tool result is passed on as content, not as the text of the raw parts.""" + result = _function_result( + _tool_turn( + [ + {"type": "text", "text": "chart:"}, + { + "type": "image", + "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}, + }, + ] + ) + ) + + # Function results carry their output as Content items; ``result`` is only their text. + assert [item.type for item in result.items] == ["text", "uri"] + assert result.items[0].text == "chart:" + assert result.items[1].uri == "https://example.com/chart.png" + assert result.items[1].media_type == "image/png" + + +def test_text_only_tool_result_stays_text() -> None: + """A tool result of text parts only is passed on as its text.""" + result = _function_result(_tool_turn([{"type": "text", "text": "4"}, {"type": "text", "text": "2"}])) + + assert result.result == "42" + assert [item.type for item in result.items] == ["text"] + + +def test_file_source_is_an_unresolved_reference() -> None: + """A 1.0 file source is a provider-issued handle: it is rejected like an id reference, never read as data.""" + message = { + "role": "user", + "content": [ + {"type": "image", "source": {"type": "file", "value": "file-abc123", "provider": "openai"}}, + ], + } + + with pytest.raises(ValueError, match="Unresolved AG-UI multimodal binary references"): + agui_messages_to_agent_framework([message]) + + +def test_stored_tool_result_media_converts_on_replay() -> None: + """Tool media kept as ``binary`` parts in a stored snapshot reaches the function result when replayed.""" + result = _function_result( + _tool_turn( + [ + {"type": "text", "text": "chart:"}, + {"type": "binary", "url": "https://example.com/chart.png", "mimeType": "image/png"}, + ] + ) + ) + + assert [item.type for item in result.items] == ["text", "uri"] + assert result.items[1].uri == "https://example.com/chart.png" + + +def test_streamed_tool_result_media_reaches_the_function_result() -> None: + """A TOOL_CALL_RESULT event with content parts gives the client the same Content items as a tool message.""" + update = AGUIEventConverter().convert_event( + { + "type": "TOOL_CALL_RESULT", + "toolCallId": "call-1", + "content": [ + {"type": "text", "text": "chart:"}, + { + "type": "image", + "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}, + }, + ], + } + ) + + assert update is not None + items = update.contents[0].items + assert items is not None + assert [item.type for item in items] == ["text", "uri"] + assert items[1].uri == "https://example.com/chart.png" + + +def test_streamed_text_only_tool_result_stays_text() -> None: + """A TOOL_CALL_RESULT event of text parts gives its text, not the string form of the list.""" + update = AGUIEventConverter().convert_event( + { + "type": "TOOL_CALL_RESULT", + "toolCallId": "call-1", + "content": [{"type": "text", "text": "4"}, {"type": "text", "text": "2"}], + } + ) + + assert update is not None + assert update.contents[0].result == "42" + + +def test_workflow_text_without_a_role_snapshots_as_assistant() -> None: + """A text message started without a role (1.0's default) is stored, and replays, as the assistant's.""" + builder = _WorkflowSnapshotBuilder([]) + builder.observe(TextMessageStartEvent(message_id="text-1")) + builder.observe(TextMessageContentEvent(message_id="text-1", delta="hello")) + builder.observe(TextMessageEndEvent(message_id="text-1")) + + messages = builder.build().messages + + assert [(message["role"], message["content"]) for message in messages] == [("assistant", "hello")] + MessagesSnapshotEvent(messages=snapshot_messages_for_wire(messages)) # type: ignore[arg-type] + + +@requires_1_0 +def test_workflow_tool_media_snapshots_as_wire_parts() -> None: + """Tool media from a workflow is stored as wire dictionaries, so it serializes and replays as media.""" + image = { + "type": "image", + "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}, + } + builder = _WorkflowSnapshotBuilder([]) + builder.observe(ToolCallStartEvent(tool_call_id="call-1", tool_call_name="chart")) + builder.observe( + ToolCallResultEvent.model_validate( + { + "messageId": "result-1", + "toolCallId": "call-1", + "content": [{"type": "text", "text": "chart:"}, image], + } + ) + ) + + messages = builder.build().messages + tool_message = next(message for message in messages if message["role"] == "tool") + + assert tool_message["content"] == [{"type": "text", "text": "chart:"}, image] + json.dumps(messages) + replayed = _function_result([*_tool_turn("")[:1], {**tool_message, "toolCallId": "call-1"}]) + assert [item.type for item in replayed.items] == ["text", "uri"] + + +@requires_1_0 +def test_client_history_keeps_tool_result_media() -> None: + """Tool media received by AGUIChatClient goes back out as content parts in the next request's history.""" + update = AGUIEventConverter().convert_event( + { + "type": "TOOL_CALL_RESULT", + "toolCallId": "call-1", + "content": [ + {"type": "text", "text": "chart:"}, + { + "type": "image", + "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}, + }, + ], + } + ) + assert update is not None + history = [ + Message( + role="assistant", contents=[Content.from_function_call(call_id="call-1", name="chart", arguments="{}")] + ), + Message(role="tool", contents=update.contents), + ] + + outgoing = agent_framework_messages_to_agui(history) + + tool_message = next(message for message in outgoing if message["role"] == "tool") + assert tool_message["content"] == [ + {"type": "text", "text": "chart:"}, + {"type": "image", "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}}, + ] + + +def _image_result(call_id: str) -> Content: + return Content.from_function_result( + call_id=call_id, + result=[Content.from_uri(uri="https://example.com/chart.png", media_type="image/png")], + ) + + +def test_dedupe_keeps_an_image_only_result_over_an_empty_placeholder() -> None: + """A completed image-only result replaces an earlier empty result for the same call; it is not dropped.""" + placeholder = Message(role="tool", contents=[Content.from_function_result(call_id="call-1", result="")]) + completed = Message(role="tool", contents=[_image_result("call-1")]) + + deduplicated = _deduplicate_messages([placeholder, completed]) + + assert len(deduplicated) == 1 + items = deduplicated[0].contents[0].items + assert items is not None + assert [item.type for item in items] == ["uri"] + + +@requires_1_0 +def test_approval_snapshot_rebuild_keeps_tool_result_media() -> None: + """Rebuilding tool messages for an approval snapshot keeps an earlier result's media as content parts.""" + rebuilt = _resolved_tool_result_snapshot_messages([Message(role="tool", contents=[_image_result("call-1")])]) + + assert rebuilt[("call-1", 1)][0]["content"] == [ + {"type": "image", "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}} + ] + + +def _relayed_tool_result(events: list[Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """The TOOL_CALL_RESULT content parts of a run and the tool message of its final messages snapshot.""" + result_event = next(event for event in events if isinstance(event, ToolCallResultEvent)) + assert isinstance(result_event.content, list) + snapshot = next(event for event in reversed(events) if isinstance(event, MessagesSnapshotEvent)) + tool_message = next(message for message in snapshot.messages if message.role == "tool") + return ( + [part.model_dump(by_alias=True, exclude_none=True) for part in result_event.content], + tool_message.model_dump(by_alias=True, exclude_none=True), + ) + + +@requires_1_0 +async def test_agent_run_relays_tool_result_media() -> None: + """Media in a function result an agent yields reaches TOOL_CALL_RESULT and the messages snapshot as parts.""" + from conftest import StubAgent # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports] + + agent = AgentFrameworkAgent( + agent=StubAgent( + updates=[ + AgentResponseUpdate( + contents=[Content.from_function_call(call_id="call-1", name="chart", arguments="{}")], + role="assistant", + ), + AgentResponseUpdate( + contents=[ + Content.from_function_result( + call_id="call-1", + result=[ + Content.from_text("chart:"), + Content.from_uri(uri="https://example.com/chart.png", media_type="image/png"), + ], + ) + ], + role="tool", + ), + AgentResponseUpdate(contents=[Content.from_text("done")], role="assistant"), + ] + ) + ) + expected_parts = [ + {"type": "text", "text": "chart:"}, + {"type": "image", "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}}, + ] + + events = [ + event + async for event in agent.run( + {"threadId": "t1", "runId": "r1", "messages": [{"role": "user", "content": "chart"}]} + ) + ] + + parts, tool_message = _relayed_tool_result(events) + assert parts == expected_parts + assert tool_message["content"] == expected_parts + + +@requires_1_0 +async def test_agent_relays_tool_result_media_from_an_agui_server() -> None: + """Tool media an AGUIChatClient receives is relayed by AgentFrameworkAgent as content parts, not as text.""" + image = { + "type": "image", + "source": {"type": "url", "value": "https://example.com/chart.png", "mimeType": "image/png"}, + } + upstream_events = [ + {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1", "protocolVersion": "1.0"}, + {"type": "TOOL_CALL_START", "toolCallId": "call-1", "toolCallName": "chart"}, + {"type": "TOOL_CALL_ARGS", "toolCallId": "call-1", "delta": "{}"}, + {"type": "TOOL_CALL_END", "toolCallId": "call-1"}, + { + "type": "TOOL_CALL_RESULT", + "messageId": "result-1", + "toolCallId": "call-1", + "content": [{"type": "text", "text": "chart:"}, image], + }, + {"type": "TEXT_MESSAGE_START", "messageId": "msg_1", "role": "assistant"}, + {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg_1", "delta": "done"}, + {"type": "TEXT_MESSAGE_END", "messageId": "msg_1"}, + {"type": "RUN_FINISHED", "threadId": "thread_1", "runId": "run_1"}, + ] + + async def handler(request: httpx.Request) -> httpx.Response: + body = "".join(f"data: {json.dumps(event)}\n\n" for event in upstream_events) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode()) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: + client = AGUIChatClient(endpoint="https://agui.example.test/", http_client=http_client) + agent = AgentFrameworkAgent(agent=Agent(client=client, name="relay")) + events = [ + event + async for event in agent.run( + {"threadId": "t1", "runId": "r1", "messages": [{"role": "user", "content": "chart"}]} + ) + ] + + parts, tool_message = _relayed_tool_result(events) + assert parts == [{"type": "text", "text": "chart:"}, image] + assert tool_message["content"] == [{"type": "text", "text": "chart:"}, image] diff --git a/python/packages/ag-ui/tests/ag_ui/test_run_common.py b/python/packages/ag-ui/tests/ag_ui/test_run_common.py index 69822014420..b58cccd2843 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_run_common.py +++ b/python/packages/ag-ui/tests/ag_ui/test_run_common.py @@ -621,6 +621,7 @@ def test_mcp_host_payload_has_live_snapshot_and_approval_parity(self): approval_snapshot = _resolved_tool_result_snapshot_messages( [Message(role="tool", contents=[content], message_id="approval-result")] )[("mcp-1", 1)][0] + assert isinstance(approval_event.content, str) assert json.loads(approval_event.content) == host_payload assert approval_snapshot["content"] == "Summary" assert json.loads(approval_snapshot[_AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY]) == host_payload @@ -689,6 +690,7 @@ def test_explicit_display_payload_wins_over_mcp_host_projection(self): assert json.loads(result_event.content) == display_payload # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] assert flow.tool_results[-1]["content"] == "Summary" assert json.loads(flow.tool_results[-1][_AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY]) == display_payload + assert isinstance(approval_event.content, str) assert json.loads(approval_event.content) == display_payload assert approval_snapshot["content"] == "Summary" assert json.loads(approval_snapshot[_AGUI_TOOL_RESULT_HOST_PAYLOAD_KEY]) == display_payload diff --git a/python/packages/ag-ui/tests/ag_ui/test_task_steps_example_agent.py b/python/packages/ag-ui/tests/ag_ui/test_task_steps_example_agent.py new file mode 100644 index 00000000000..4dbf56cbace --- /dev/null +++ b/python/packages/ag-ui/tests/ag_ui/test_task_steps_example_agent.py @@ -0,0 +1,98 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Tests for the task steps example agent used by Dojo.""" + +from __future__ import annotations + +import json +from collections.abc import AsyncIterator, MutableSequence +from types import SimpleNamespace +from typing import Any + +import pytest +from ag_ui.core import MessagesSnapshotEvent, StateDeltaEvent, StateSnapshotEvent, TextMessageContentEvent +from agent_framework import ChatResponseUpdate, Content, Message + +from agent_framework_ag_ui._utils import _AGUI_PROTOCOL_VERSION +from agent_framework_ag_ui_examples.agents import task_steps_agent as task_steps_module + +_STEPS = [{"description": "Digging hole", "status": "pending"}, {"description": "Planting tree", "status": "pending"}] + + +async def _no_sleep(_: float) -> None: + return None + + +@pytest.fixture +def task_steps_agent(streaming_chat_client_stub: Any, monkeypatch: pytest.MonkeyPatch) -> Any: + """The demo agent over a scripted client: plan the steps, confirm, then summarize.""" + monkeypatch.setattr(task_steps_module, "asyncio", SimpleNamespace(sleep=_no_sleep)) + + async def stream_fn( + messages: MutableSequence[Message], options: dict[str, Any], **kwargs: Any + ) -> AsyncIterator[ChatResponseUpdate]: + del options, kwargs + last = messages[-1] + if "successfully executed" in (last.text or ""): + yield ChatResponseUpdate(contents=[Content.from_text(text="Both steps are done.")]) + elif any(content.type == "function_result" for content in last.contents): + yield ChatResponseUpdate(contents=[Content.from_text(text="I created the plan.")]) + else: + yield ChatResponseUpdate( + contents=[ + Content.from_function_call( + name="generate_task_steps", call_id="call-steps", arguments=json.dumps({"steps": _STEPS}) + ) + ] + ) + + return task_steps_module.task_steps_agent_wrapped(streaming_chat_client_stub(stream_fn)) + + +async def test_predictive_steps_delta_drives_step_execution(task_steps_agent: Any) -> None: + """The predictive /steps delta is read whichever way ag-ui-protocol models patch operations.""" + events = [ + event + async for event in task_steps_agent.run( + {"thread_id": "t1", "run_id": "r1", "messages": [{"id": "u1", "role": "user", "content": "Plant a tree"}]} + ) + ] + + steps_deltas = [ + event + for event in events + if isinstance(event, StateDeltaEvent) and event.model_dump()["delta"][0]["path"] == "/steps" + ] + assert steps_deltas + final_states = [event.snapshot for event in events if isinstance(event, StateSnapshotEvent)] + assert final_states[-1] == {"steps": [{**step, "status": "completed"} for step in _STEPS]} + + +async def test_final_snapshot_accepts_history_with_legacy_binary_attachment(task_steps_agent: Any) -> None: + """The summary MESSAGES_SNAPSHOT replays a legacy binary attachment from the request history.""" + history = [ + { + "id": "u1", + "role": "user", + "content": [ + {"type": "text", "text": "Plan around this photo"}, + {"type": "binary", "mimeType": "image/png", "url": "https://example.com/garden.png"}, + ], + } + ] + + events = [event async for event in task_steps_agent.run({"thread_id": "t1", "run_id": "r1", "messages": history})] + + summary_text = "".join(event.delta for event in events if isinstance(event, TextMessageContentEvent)) + assert "Summary generation error" not in summary_text + final_snapshot = [event for event in events if isinstance(event, MessagesSnapshotEvent)][-1] + dumped = final_snapshot.model_dump(by_alias=True, exclude_none=True)["messages"] + assert dumped[-1]["content"] == "Both steps are done." + attachment = dumped[0]["content"][1] + if _AGUI_PROTOCOL_VERSION is None: # ag-ui-protocol < 1.0 still models the binary part + assert attachment["type"] == "binary" + else: + assert attachment == { + "type": "image", + "source": {"type": "url", "value": "https://example.com/garden.png", "mimeType": "image/png"}, + } diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py index 1c46e177565..86f45d87046 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py +++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py @@ -24,6 +24,7 @@ ) from agent_framework_ag_ui import AgentFrameworkWorkflow +from agent_framework_ag_ui._message_adapters import snapshot_messages_for_wire from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY from agent_framework_ag_ui._workflow import _CHECKPOINT_REQUEST_OWNER_KEY @@ -1043,7 +1044,10 @@ def build_workflow() -> Workflow: hydrated = await _run(agent, {"thread_id": "content-thread", "messages": [], _SNAPSHOT_SCOPE_INPUT_KEY: "scope"}) replay = next(event for event in hydrated if event.type == "MESSAGES_SNAPSHOT") replay_replies = [message for message in replay.messages if message.role == "user" and message.id != "start"] - assert [message.model_dump(by_alias=True, exclude_none=True) for message in replay_replies] == user_replies + # The wire snapshot carries media in the shape the installed ag-ui-protocol accepts. + assert [message.model_dump(by_alias=True, exclude_none=True) for message in replay_replies] == ( + snapshot_messages_for_wire(user_replies) + ) assert len(handled) == 1 @@ -1167,7 +1171,7 @@ async def capture( for message in replay.messages if message.role == "user" ] - assert replay_users == [message["content"] for message in user_messages] + assert replay_users == [message["content"] for message in snapshot_messages_for_wire(user_messages)] assert handled == [resume_replies] diff --git a/python/uv.lock b/python/uv.lock index 6571372d261..eb0500deed9 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -103,14 +103,14 @@ wheels = [ [[package]] name = "ag-ui-protocol" -version = "0.1.22" +version = "1.0.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pydantic" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/e8/f7/9bf788e7d3608725d022a248a58427040f4f930ab87ddb54bd29ee4d9a51/ag_ui_protocol-0.1.22.tar.gz", hash = "sha256:d21f265284a50d9fc87ad7bcbd58f737b4b16eef7b5375f13a6e925117b52046", size = 18110, upload-time = "2026-08-31T18:20:04.334Z" } +sdist = { url = "https://files.pythonhosted.org/packages/57/92/d88fdc7f4648dc38d3d54c595066bb3e63a4c091884ffb2f98272c33eef8/ag_ui_protocol-1.0.0.tar.gz", hash = "sha256:cfebecef2e7bc942cc8a52d908a4ea86a98f631d2ffdc6ec5812bb843eac74fb", size = 30759, upload-time = "2026-09-17T18:31:19.448Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/be/6c/af3d577e68c9474c99600e65b2c6283772aae187edca6f0f2f5cbd9f565a/ag_ui_protocol-0.1.22-py3-none-any.whl", hash = "sha256:fca13ee7820f8f53e869c19e09ddd75826c1799b27c2adb6f2e567295433c704", size = 22068, upload-time = "2026-08-31T18:20:03.43Z" }, + { url = "https://files.pythonhosted.org/packages/82/b0/850be4576e6443dc575daa54188d8136ae2b2eca3f19e098e7a088fc9ebc/ag_ui_protocol-1.0.0-py3-none-any.whl", hash = "sha256:0058eaf0522201089d444e430fc35180bc8ad0222cae39e8568b313e8105d9e6", size = 36842, upload-time = "2026-09-17T18:31:18.471Z" }, ] [[package]] @@ -217,7 +217,7 @@ test = [ [package.metadata] requires-dist = [ { name = "ag-ui-a2ui-toolkit", marker = "extra == 'a2ui'", specifier = ">=0.0.4" }, - { name = "ag-ui-protocol", specifier = ">=0.1.19,<0.2" }, + { name = "ag-ui-protocol", specifier = ">=0.1.19,<2" }, { name = "agent-framework-core", editable = "packages/core" }, { name = "fastapi", specifier = ">=0.121.0,<0.142.0" }, { name = "httpx", specifier = ">=0.28.1,<1" },