pydantic_ai.ag_ui
Provides an AG-UI protocol adapter for the Pydantic AI agent.
This package provides seamless integration between pydantic-ai agents and ag-ui for building interactive AI applications with streaming event-based communication.
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.
Bases: Protocol
Protocol for state handlers in agent runs. Requires the class to be a dataclass with a state field.
Get the current state of the agent run.
Type: Any
Bases: Generic[StateT]
Dependency type that holds state.
This class is used to manage the state of an agent run. It allows setting
the state of the agent run with a specific type of state model, which must
be a subclass of BaseModel.
The state is set using the state setter by the Adapter when the run starts.
Implements the StateHandler protocol.
@async
def handle_ag_ui_request(
agent: AbstractAgent[AgentDepsT, Any],
request: Request,
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,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
on_complete: OnCompleteFunc[BaseEvent] | None = None,
) -> Response
Handle an AG-UI request by running the agent and returning a streaming response.
Response — A streaming Starlette response with AG-UI protocol events.
agent : AbstractAgent[AgentDepsT, Any]
The agent to run.
The Starlette request (e.g. from FastAPI) containing the AG-UI run input.
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.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
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.
on_complete : OnCompleteFunc[BaseEvent] | None Default: None
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.
def run_ag_ui(
agent: AbstractAgent[AgentDepsT, Any],
run_input: RunAgentInput,
accept: str = SSE_CONTENT_TYPE,
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,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
on_complete: OnCompleteFunc[BaseEvent] | None = None,
) -> AsyncIterator[str]
Run the agent with the AG-UI run input and stream AG-UI protocol events.
agent : AbstractAgent[AgentDepsT, Any]
The agent to run.
The AG-UI run input containing thread_id, run_id, messages, etc.
accept : str Default: SSE_CONTENT_TYPE
The accept header value for the 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.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
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.
on_complete : OnCompleteFunc[BaseEvent] | None Default: None
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.
Content type header value for Server-Sent Events (SSE).
Default: 'text/event-stream'
Callback function type that receives the AgentRunResult of the completed run. Can be sync, async, or an async generator of protocol-specific events.
Type: TypeAlias Default: Callable[[AgentRunResult[Any]], None] | Callable[[AgentRunResult[Any]], Awaitable[None]] | Callable[[AgentRunResult[Any]], AsyncIterator[EventT]]