Skip to content
Pydantic Docs

pydantic_ai.ui.ag_ui

AG-UI protocol integration for Pydantic AI agents.

AGUIEventStream

Bases: UIEventStream[RunAgentInput, BaseEvent, AgentDepsT, OutputDataT]

UI event stream transformer for the Agent-User Interaction (AG-UI) protocol.

Methods

handle_event

@async

def handle_event(event: NativeEvent) -> AsyncIterator[BaseEvent]

Override to set timestamps on all AG-UI events.

Returns

AsyncIterator[BaseEvent]

AGUIAdapter

Bases: UIAdapter[RunAgentInput, Message, BaseEvent, AgentDepsT, OutputDataT]

UI adapter for the Agent-User Interaction (AG-UI) protocol.

Attributes

ag_ui_version

AG-UI protocol version controlling behavior thresholds.

Accepts any version string (e.g. '0.1.13'). Defaults to the version detected from the installed ag-ui-protocol package.

Known thresholds:

  • < 0.1.13: emits THINKING_* events during streaming, drops ThinkingPart from dump_messages output.
  • >= 0.1.13: emits REASONING_* events with encrypted metadata during streaming, and includes ThinkingPart as ReasoningMessage in dump_messages output for full round-trip fidelity of thinking signatures and provider metadata.
  • >= 0.1.15: emits typed multimodal input content (ImageInputContent, AudioInputContent, VideoInputContent, DocumentInputContent) instead of generic BinaryInputContent.

load_messages always accepts ReasoningMessage and multimodal content types regardless of this setting.

Type: str Default: DEFAULT_AG_UI_VERSION

preserve_file_data

Whether to round-trip FilePart and UploadedFile through reserved pydantic_ai_* activity messages.

Defaults to False. AG-UI has no native representation for agent-generated files (FilePart) or uploaded-file references (UploadedFile), so when this is True they are serialized as sidecar activity messages on dump_messages and reconstructed on load_messages. A frontend only completes the round-trip if it echoes these activity messages back on the next request.

This is a representation setting, not a security one: honoring a reconstructed inbound UploadedFile still requires allow_uploaded_files, which the shared sanitize_messages step enforces regardless of this flag. Multimodal tool-return files are unaffected — they ride inline in ToolMessage.content.

Type: bool Default: False

messages

Pydantic AI messages from the AG-UI run input.

Type: list[ModelMessage]

toolset

Toolset representing frontend tools from the AG-UI run input.

Type: AbstractToolset[AgentDepsT] | None

state

Frontend state from the AG-UI run input.

Type: dict[str, Any] | None

conversation_id

Conversation ID from the AG-UI RunAgentInput.threadId.

Type: str | None

deferred_tool_results

Translate AG-UI RunAgentInput.resume[] into Pydantic AI DeferredToolResults.

See docs.ag-ui.com/concepts/interrupts.

Each ResumeEntry is mapped to an approval keyed by the original tool_call_id. The mapping is deny-by-default: approval requires an explicit payload.approved == True. Any other shape is treated as a denial so a malformed or hostile client cannot accidentally execute a tool that requires human approval.

  • status == 'cancelled' → ToolDenied('Cancelled by user.')
  • payload.approved is True with payload.editedArgs → ToolApproved(override_args=...)
  • payload.approved is True without edits → ToolApproved()
  • Anything else (False, missing, null, non-bool, non-dict payload) → ToolDenied(payload.get('reason')) if reason is a non-empty string, else ToolDenied() (which carries the default "The tool call was denied." message).

Returns None when resume is missing or empty, or when the installed ag-ui-protocol predates the interrupt lifecycle.

Type: DeferredToolResults | None

Methods

build_run_input

@classmethod

def build_run_input(cls, body: bytes) -> RunAgentInput

Build an AG-UI run input object from the request body.

Returns

RunAgentInput

build_event_stream 
def build_event_stream(

) -> UIEventStream[RunAgentInput, BaseEvent, AgentDepsT, OutputDataT]

Build an AG-UI event stream transformer.

Returns

UIEventStream[RunAgentInput, BaseEvent, AgentDepsT, OutputDataT]

from_request

@async

@classmethod

def from_request(
    cls,
    request: Request,
    *,
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    ag_ui_version: str = DEFAULT_AG_UI_VERSION,
    preserve_file_data: bool = False,
    manage_system_prompt: Literal['server', 'client'] = 'server',
    allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'}),
    allowed_file_url_force_download: frozenset[ForceDownloadMode] = frozenset(),
    allow_uploaded_files: bool = False,
    **kwargs: Any,
) -> AGUIAdapter[AgentDepsT, OutputDataT]

Extends from_request with AG-UI-specific parameters.

Returns

AGUIAdapter[AgentDepsT, OutputDataT]

load_messages

@classmethod

def load_messages(
    cls,
    messages: Sequence[Message],
    *,
    preserve_file_data: bool = False,
) -> list[ModelMessage]

Transform AG-UI messages into Pydantic AI messages.

Returns

list[ModelMessage]

dump_messages

@classmethod

def dump_messages(
    cls,
    messages: Sequence[ModelMessage],
    *,
    ag_ui_version: str = DEFAULT_AG_UI_VERSION,
    preserve_file_data: bool = False,
) -> list[Message]

Transform Pydantic AI messages into AG-UI messages.

Note: The round-trip dump_messages -> load_messages is not fully lossless:

  • TextPart.id, .provider_name, .provider_details are lost.
  • ToolCallPart.id, .provider_name, .provider_details are lost.
  • NativeToolCallPart.id, .provider_details are lost (only .provider_name survives via the prefixed tool call ID).
  • NativeToolReturnPart.provider_details is lost.
  • tool_kind is lost when ag_ui_version < '0.1.11' (before its encrypted_value carrier existed), so typed tool parts reload as their base classes.
  • tool_kind is not restored on error/denied tool returns (a typed return implies success to its readers), so those reload as plain ToolReturnPart.
  • RetryPromptPart becomes ToolReturnPart (or UserPromptPart) on reload.
  • CachePoint and UploadedFile content items are dropped (unless preserve_file_data=True).
  • FileUrl.force_download is dropped when ag_ui_version < '0.1.15' (before typed multimodal content gained a metadata carrier).
  • ThinkingPart is dropped when ag_ui_version='0.1.10'.
  • FilePart is silently dropped unless preserve_file_data=True.
  • UploadedFile in a multi-item UserPromptPart is split into a separate activity message when preserve_file_data=True, which reloads as a separate UserPromptPart.
  • MultiModalContent items in ToolReturnPart/NativeToolReturnPart.content always round-trip, regardless of preserve_file_data: the full content (files as base64/URL dicts) is serialized inline into the JSON ToolMessage.content and rehydrated on reload via the ToolReturnContent discriminator. The same serialization is used for both history (dump_messages) and the live event stream (ToolCallResultEvent.content), so files survive either round-trip.
  • Part ordering within a ModelResponse may change when text follows tool calls.
Returns

list[Message] — A list of AG-UI Message objects.

Parameters

messages : Sequence[ModelMessage]

A sequence of ModelMessage objects to convert.

ag_ui_version : str Default: DEFAULT_AG_UI_VERSION

AG-UI protocol version controlling ThinkingPart emission.

preserve_file_data : bool Default: False

Whether to include FilePart and UploadedFile items as ActivityMessages. (Multimodal tool-return files always ride inline in ToolMessage.content and are unaffected.)

DEFAULT_AG_UI_VERSION

The default AG-UI version, auto-detected from the installed ag-ui-protocol package.

Type: str Default: detect_ag_ui_version()