pydantic_ai.ui.ag_ui
AG-UI protocol integration for Pydantic AI agents.
Bases: UIEventStream[RunAgentInput, BaseEvent, AgentDepsT, OutputDataT]
UI event stream transformer for the Agent-User Interaction (AG-UI) protocol.
@async
def handle_event(event: NativeEvent) -> AsyncIterator[BaseEvent]
Override to set timestamps on all AG-UI events.
AsyncIterator[BaseEvent]
Bases: UIAdapter[RunAgentInput, Message, BaseEvent, AgentDepsT, OutputDataT]
UI adapter for the Agent-User Interaction (AG-UI) protocol.
Pydantic AI messages from the AG-UI run input.
Type: list[ModelMessage]
Toolset representing frontend tools from the AG-UI run input.
Type: AbstractToolset[AgentDepsT] | None
Frontend state from the AG-UI run input.
@classmethod
def build_run_input(cls, body: bytes) -> RunAgentInput
Build an AG-UI run input object from the request body.
RunAgentInput
def build_event_stream(
) -> UIEventStream[RunAgentInput, BaseEvent, AgentDepsT, OutputDataT]
Build an AG-UI event stream transformer.
UIEventStream[RunAgentInput, BaseEvent, AgentDepsT, OutputDataT]
@classmethod
def load_messages(cls, messages: Sequence[Message]) -> list[ModelMessage]
Transform AG-UI messages into Pydantic AI messages.
AG-UI protocol integration for Pydantic AI agents.
Bases: Generic[AgentDepsT, OutputDataT], Starlette
ASGI application for running Pydantic AI agents with AG-UI protocol support.
def __init__(
agent: AbstractAgent[AgentDepsT, OutputDataT],
output_type: OutputSpec[Any] | None = None,
message_history: Sequence[ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: Model | KnownModelName | str | None = None,
deps: AgentDepsT = None,
model_settings: ModelSettings | None = None,
usage_limits: UsageLimits | None = None,
usage: RunUsage | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
on_complete: OnCompleteFunc[Any] | None = None,
debug: bool = False,
routes: Sequence[BaseRoute] | None = None,
middleware: Sequence[Middleware] | None = None,
exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
on_startup: Sequence[Callable[[], Any]] | None = None,
on_shutdown: Sequence[Callable[[], Any]] | None = None,
lifespan: Lifespan[Self] | None = None,
) -> None
An ASGI application that handles every request by running the agent and streaming the response.
Note that the deps will be the same for each request, with the exception of the frontend state that’s
injected into the state field of a deps object that implements the StateHandler protocol.
To provide different deps for each request (e.g. based on the authenticated user),
use AGUIAdapter.run_stream() or
AGUIAdapter.dispatch_request() instead.
The agent to run.
Custom output type to use for this run, output_type may only be used if the agent has
no output validators since output validators would expect an argument that matches the agent’s
output type.
message_history : Sequence[ModelMessage] | None Default: None
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
Optional model to use for this run, required if model was not set when creating the agent.
Optional dependencies to use for this run.
model_settings : ModelSettings | None Default: None
Optional settings to use for this model’s request.
usage_limits : UsageLimits | None Default: None
Optional limits on model request count or token usage.
Optional usage to start with, useful for resuming a conversation or agents used in tools.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
Optional additional builtin tools for this run.
Optional callback function called when the agent run completes successfully.
The callback receives the completed AgentRunResult and can access all_messages() and other result data.
debug : bool Default: False
Boolean indicating if debug tracebacks should be returned on errors.
A list of routes to serve incoming HTTP and WebSocket requests.
A list of middleware to run for every request. A starlette application will always
automatically include two middleware classes. ServerErrorMiddleware is added as the very
outermost middleware, to handle any uncaught errors occurring anywhere in the entire stack.
ExceptionMiddleware is added as the very innermost middleware, to deal with handled
exception cases occurring in the routing or endpoints.
A mapping of either integer status codes, or exception class types onto
callables which handle the exceptions. Exception handler callables should be of the form
handler(request, exc) -> response and may be either standard functions, or async functions.
A list of callables to run on application startup. Startup handler callables do not take any arguments, and may be either standard functions, or async functions.
A list of callables to run on application shutdown. Shutdown handler callables do not take any arguments, and may be either standard functions, or async functions.
A lifespan context function, which can be used to perform startup and shutdown tasks.
This is a newer style that replaces the on_startup and on_shutdown handlers. Use one or
the other, not both.