UI Event Streams

If you’re building a chat app or other interactive frontend for an AI agent, your backend will need to receive agent run input (like a chat message or complete message history) from the frontend, and will need to stream the agent’s events (like text, thinking, and tool calls) to the frontend so that the user knows what’s happening in real time.

While your frontend could use Pydantic AI’s ModelRequest and AgentStreamEvent directly, you’ll typically want to use a UI event stream protocol that’s natively supported by your frontend framework.

To push data of your own to the frontend mid-run, like a progress update from a long-running tool, emit a custom event: both protocols below forward it as their own custom-data event, unless its class is declared ui=False for events you only want server-side.

Pydantic AI natively supports two UI event stream protocols:

• Agent-User Interaction (AG-UI) Protocol
• Vercel AI Data Stream Protocol

These integrations are implemented as subclasses of the abstract UIAdapter class, so they also serve as a reference for integrating with other UI event stream protocols.

Usage

The protocol-specific UIAdapter subclass (i.e. AGUIAdapter or VercelAIAdapter) is responsible for transforming agent run input received from the frontend into arguments for Agent.run_stream_events(), running the agent, and then transforming Pydantic AI events into protocol-specific events. The event stream transformation is handled by a protocol-specific UIEventStream subclass, which you typically won’t use directly unless the agent’s events reach you outside the request that serves the frontend, as covered in “Encoding events without a request”.

If you’re using a Starlette-based web framework like FastAPI, you can use the UIAdapter.dispatch_request() class method from an endpoint function to directly handle a request and return a streaming response of protocol-specific events. This is demonstrated in the next section.

If you’re using a web framework not based on Starlette (e.g. Django or Flask) or need fine-grained control over the input or output, you can create a UIAdapter instance and directly use its methods. This is demonstrated in the “Advanced Usage” section below.

Usage with Starlette/FastAPI

Besides the request, UIAdapter.dispatch_request() takes the agent, the same optional arguments as Agent.run_stream_events(), an optional on_complete callback for successful runs, and an optional on_cancel callback that receives RunCancelled for first-party cancelled runs (a client disconnect is an external cancellation and does not trigger it). Both callbacks can optionally yield additional protocol-specific events.

on_complete runs once the agent run has finished and the response is already streaming, so an exception it raises, such as a failure to save the conversation, doesn’t reach your request handler. It goes to the event stream’s on_error() like any other error during streaming, which sends the client the protocol’s error event. It’s also recorded, as its exception type only, on the OpenTelemetry span that’s current when it’s raised, which is your HTTP request’s span if your server is instrumented. To log the exception or raise it yourself, override on_error in a subclass of the adapter’s event stream, and return that subclass from your adapter’s build_event_stream().

Note

These examples use the VercelAIAdapter, but the same patterns apply to all UIAdapter subclasses.

dispatch_request.py

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent
from pydantic_ai.ui.vercel_ai import VercelAIAdapter

agent = Agent('openai:gpt-6.1-sol')

app = FastAPI()

@app.post('/chat')
async def chat(request: Request) -> Response:
    return await VercelAIAdapter.dispatch_request(request, agent=agent)

Advanced Usage

If you’re using a web framework not based on Starlette (e.g. Django or Flask) or need fine-grained control over the input or output, you can create a UIAdapter instance and directly use its methods, which can be chained to accomplish the same thing as the UIAdapter.dispatch_request() class method shown above:

1. The UIAdapter.build_run_input() class method takes the request body as bytes and returns a protocol-specific run input object, which you can then pass to the UIAdapter() constructor along with the agent.
   • You can also use the UIAdapter.from_request() class method to build an adapter directly from a Starlette/FastAPI request.
2. The UIAdapter.run_stream() method runs the agent and returns a stream of protocol-specific events. It supports the same optional arguments as Agent.run_stream_events(), including the on_complete and on_cancel callbacks.
   • You can also use UIAdapter.run_stream_native() to run the agent and return a stream of Pydantic AI events instead, which can then be transformed into protocol-specific events using UIAdapter.transform_stream().
   • A request the adapter rejects as malformed — an AG-UI resume payload that doesn’t match the schema it advertised, for instance — raises on the first iteration of the stream rather than when the run method is called, so that pairing it with transform_stream() turns it into a protocol error event. Consuming the native stream without that pairing means catching it around your async for.
3. The UIAdapter.encode_stream() method encodes the stream of protocol-specific events as SSE (HTTP Server-Sent Events) strings, which you can then return as a streaming response.
   • You can also use UIAdapter.streaming_response() to generate a Starlette/FastAPI streaming response directly from the protocol-specific event stream returned by run_stream().

Note

This example uses FastAPI, but can be modified to work with any web framework.

run_stream.py

import json
from http import HTTPStatus

from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.responses import Response, StreamingResponse
from pydantic import ValidationError

from pydantic_ai import Agent
from pydantic_ai.ui import SSE_CONTENT_TYPE
from pydantic_ai.ui.vercel_ai import VercelAIAdapter

agent = Agent('openai:gpt-6.1-sol')

app = FastAPI()


@app.post('/chat')
async def chat(request: Request) -> Response:
    media_type = request.headers.get('content-type', '').split(';')[0].strip().lower()
    if media_type != 'application/json':  # (1)
        return Response(status_code=HTTPStatus.UNSUPPORTED_MEDIA_TYPE)

    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        run_input = VercelAIAdapter.build_run_input(await request.body())
    except ValidationError as e:
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    adapter = VercelAIAdapter(agent=agent, run_input=run_input, accept=accept)
    event_stream = adapter.run_stream()

    sse_event_stream = adapter.encode_stream(event_stream)
    return StreamingResponse(sse_event_stream, media_type=accept)

build_run_input() takes bytes, so it can't apply the media-type check that from_request() does — see the trust model for what it's for. Do it in your own framework's idiom, as here, whenever you read the body yourself.

Running the agent elsewhere

The adapter assumes the agent runs inside the request that serves the frontend. It doesn’t have to. If the agent runs on a durable execution worker, a background job, or another service, the adapter’s two jobs simply split across that boundary: the request side builds the run input and transforms the events it receives back into protocol events, and whatever runs the agent builds the run arguments from the same request body.

The request side never calls run_stream(). It hands UIAdapter.transform_stream() the agent’s events as they arrive over your transport (the media-type and validation handling from the example above still applies, and is elided here):

remote_run_handler.py

async def chat(request: Request) -> Response:
    body = await request.body()
    events = start_the_run_somewhere_else(body)  # an async iterator of Pydantic AI events

    adapter = VercelAIAdapter(agent=agent, run_input=VercelAIAdapter.build_run_input(body))
    return adapter.streaming_response(adapter.transform_stream(events))

The transport has to deliver the run’s AgentRunResultEvent as well as its AgentStreamEvents — that’s what closes the protocol out and what on_complete receives — so it needs to carry what Agent.run_stream_events() yields, not just the model’s events.

Temporal’s Workflow Streams are one such transport, and are worth reading as a worked example: the workflow is the queue, the events are durable and offset-addressed, and a frontend that reconnects can be reattached to a run it didn’t start.

Encoding events without a request

If the agent doesn’t run inside the request that serves the frontend — its events reach your API edge over a transport of their own, like a durable execution workflow, a queue, or a websocket fan-out — there’s no request body to build a run input from, and no UIAdapter to run the agent. Use the protocol-specific UIEventStream subclass on its own instead: it transforms and encodes the agent’s events and takes no run input.

encode_events.py

from collections.abc import AsyncIterator

from pydantic_ai.ui import NativeEvent
from pydantic_ai.ui.vercel_ai import VercelAIEventStream


async def encode_events(events: AsyncIterator[NativeEvent]) -> AsyncIterator[str]:
    event_stream = VercelAIEventStream()
    async for sse_event in event_stream.encode_stream(event_stream.transform_stream(events)):
        yield sse_event

An event stream instance carries the state of one run as it goes (the current message ID, the part it’s streaming, the tool calls it’s waiting on), so build a new one per run.

The AG-UI protocol identifies every run to the frontend: its RUN_STARTED and RUN_FINISHED events carry a thread ID and a run ID, which AGUIEventStream reads off the run input when it has one, warning you if you pass IDs it then overrides. Without a run input, pass the IDs your own transport already assigns to the conversation and the run:

encode_ag_ui_events.py

from pydantic_ai.ui.ag_ui import AGUIEventStream

event_stream = AGUIEventStream(thread_id='conversation-123', run_id='workflow-456')

Each defaults to a new UUID, minted every time the stream is constructed: a conversation that spans more than one run — a run resumed after a tool approval above all — has to pass its own thread_id for the frontend to correlate the runs. Constructing the stream inside replay-able durable execution workflow code makes that a determinism hazard too, as the defaults are re-minted on every replay, so pass an explicit thread_id and run_id there. Passing the conversation_id and run_id of the agent run itself lines the protocol’s identity up with the run’s traces. On the request path, AGUIAdapter lines up the thread ID only: it maps the run input’s thread ID onto the agent’s conversation_id, while the protocol’s run ID stays the one the client sent and is never wired to the agent run’s run_id, so the two differ.

Trust model for client-submitted messages

UI adapter endpoints aren’t authentication boundaries. Both the AG-UI and Vercel AI protocols are designed around the client transmitting the full conversation history on each request, so anything in message_history from the protocol — assistant messages, tool calls, file URLs, tool results — is under the caller’s control. An AG-UI client also declares its own frontend tools, and their names, descriptions and argument schemas reach the model as the client sent them; see the AG-UI trust model for how to limit them. Treat the adapter endpoint as an internal backend service, running it inside your own authenticated route handler. See the AG-UI security considerations page for more on the deployment model both protocols assume.

Authenticating the endpoint settles who may call it, not what caused the call. Where that authentication is a cookie or any other credential the browser attaches on its own, a page your user has open elsewhere can post to your route on their behalf: the run starts and its tools execute, and the attacker never needs to read the response. To keep a request a browser can forge without a preflight from reaching the agent, the adapters accept only application/json request bodies, answering anything else with a 415 before the body is read. Both protocols’ SDK transports send that content type, so a frontend on another origin is preflighted and admitted by your own CORS policy exactly as before. Widen the set, or skip the check on a route that carries CSRF protection of its own, with the allowed_content_types argument to UIAdapter.from_request() and UIAdapter.dispatch_request(). It’s one control rather than a CSRF strategy: on a route that carries ambient credentials, pair it with whatever your framework offers.

The adapters apply a few defaults so that the authoritative state stays on your side:

• System prompts — client-submitted SystemPromptParts are stripped by default and replaced with the agent’s configured prompt. InstructionDeltaParts and InstructionBaselineParts are stripped too, and request-level ModelRequest.instructions are reset to None. RetryFeedbackParts are kept, so the model still sees why it corrected itself, but the metadata marker that reloads one is echoed by the client, so a client can forge one and pick its cause. Pydantic AI’s own retry feedback reaches the model in the user voice, fenced, for a validation error or a ModelRetry, and in the system voice only for 'no_output', so a returned 'no_output' part is demoted to the same fenced user text: the client can’t earn the system voice by forging one. Control with UIAdapter.manage_system_prompt; 'client' hands over the system prompt and 'no_output' feedback both — see each adapter’s docs for details. Request-scoped ones are left out of dump_messages altogether, with either setting, since they’re only for the model’s current request, so a history that round-trips through the frontend loses them; keep it server-side, as described below, where that matters. Persist append-on-change instruction history server-side as well; UI message dumps omit its operator records.
• Dangling tool calls — if the client-submitted history ends in a ModelResponse with unresolved ToolCallParts and no matching deferred_tool_results, the tool calls are dropped with a warning, as a best-effort default so the agent doesn’t execute an unresolved tool call the model never emitted. For human-in-the-loop resumption, pass explicit deferred_tool_results to the run method — tool calls resolved by those results are kept (see the warning below).
• File URL schemes — only http and https are accepted by default for FileUrl parts in client-submitted messages. Non-HTTP schemes like s3:// or gs:// are dropped, since they cause the provider to fetch the object using your server’s IAM role or service account. See UIAdapter.allowed_file_url_schemes.
• File URL download mode — FileUrl.force_download values other than False are reset to False by default on client-submitted messages. This prevents clients from forcing the server to fetch a URL, or using 'allow-local' to opt out of the SSRF private-IP block. After auditing your frontend, opt into additional values with UIAdapter.allowed_file_url_force_download.
• Uploaded files — client-submitted UploadedFile parts are dropped by default, just like non-HTTP FileUrls, since the server resolves them against the provider’s file storage API using its own credentials. After auditing your frontend, honor them by setting UIAdapter.allow_uploaded_files to True. This is a purely inbound security setting: file content the agent produces is always serialized on the way back out to the client.
• Workspace references — ModelResponse.workspace_ref is reset to None on client-submitted messages by default, so a client can’t point your agent at another conversation’s workspace through your server’s provider credentials. As a result, a sandbox capability starts a new sandbox on every request. To keep one per conversation, save result.workspace.ref in on_complete and pass it back as workspace= on the next request; see Continuing in the same workspace. The Vercel AI and AG-UI protocols do not carry workspace references, so strip_workspace_refs=False only affects sanitize_messages and custom adapters; it cannot round-trip a ref through those protocols. Keep the saved ref server-side and authorize it before passing workspace=.
• Framework-reserved metadata — the __pydantic_ai__ key in ModelRequest.metadata and ModelResponse.metadata holds state Pydantic AI and its capabilities act on in later runs, such as the file descriptions FileUnderstanding records, so it’s dropped from client-submitted messages with a warning, and that state is computed again where it’s needed. Your own metadata keys are kept. The UI message id the adapter keeps there is the client’s own, so it survives, and dump_messages gives each message back under the id it was sent with.

Tool approvals and results are submitted by the client

On the human-in-the-loop resumption path, the DeferredToolResults passed to the run method — approvals, denials, and externally-executed tool results — are submitted by the client along with the history, and the adapter does not verify that an approved tool call is one the server actually issued. A client that can reach the endpoint can therefore approve a tool call of its own making, including one for an approval-gated tool. Approval guards against the model acting without human sign-off; it is not an authorization boundary against the client.

Authenticate the endpoint (as above) and enforce authorization for sensitive actions inside the tool function itself — against the authenticated user carried in your dependencies, not the client-supplied approval — for any call that reaches the tool function. For approval-gated tools, also keep the history on the server: with manage_message_history='server', an approval or denial has to answer a tool call that the stored history is waiting on, so a request with an approval for a call that isn’t pending, including a tool call the model didn’t make, is rejected before any tool runs. Alternatively, persist paused runs server-side and pass your own deferred_tool_results rather than honoring the client’s.

For stricter conversation integrity (e.g. ensuring prior assistant turns and tool returns match what the server actually produced), keep the history on the server, as described below — caller-supplied message_history is trusted as coming from server-side persistence and isn’t subject to this sanitization.

These defaults narrow what a fabricated history can reach, but a client that can submit history can always fabricate it; see Trust boundary for client-supplied history for what that means for your application’s authorization model.

Keeping the history on the server

The AI SDK’s useChat and @ag-ui/client both send the whole conversation on every request. To make your own stored history the conversation instead, persist the agent’s messages when a run completes, pass them back as message_history on the next request, and set manage_message_history='server', which UIAdapter.from_request() and UIAdapter.dispatch_request() also accept:

server_side_history.py

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent, AgentRunResult
from pydantic_ai.messages import ModelMessage
from pydantic_ai.ui.vercel_ai import VercelAIAdapter

agent = Agent('openai:gpt-6.1-sol')

app = FastAPI()

conversations: dict[str, list[ModelMessage]] = {}  # (1)


@app.post('/chat/{chat_id}')
async def chat(chat_id: str, request: Request) -> Response:
    def save(result: AgentRunResult[str]) -> None:
        conversations[chat_id] = result.all_messages()

    return await VercelAIAdapter.dispatch_request(
        request,
        agent=agent,
        message_history=conversations.get(chat_id),
        manage_message_history='server',
        on_complete=save,
    )

Use a database in production, and check that the user may access the conversation before reading or writing it.

From the client’s messages, the adapter then uses only the new turn: the messages after the last assistant response, such as the user’s new message or the output of a tool the frontend ran. Everything before it comes from message_history, so the model sees each turn once, and the client can’t change what the model saw on earlier turns. This works whether the client sends its whole history or only its new messages.

• Regenerating and editing. When the new turn starts with a user message whose UI message id is already in message_history, the stored history is cut just before that message, so the run replaces that turn and everything after it. Tool results the server recorded alongside that message stay, so the client can’t answer those calls again. That’s what the AI SDK’s regenerate() and sendMessage({ messageId }) send, and what a retried submit looks like. A client can roll its own conversation back to any of its earlier user messages this way, as those features require, but not to any other message. When the id isn’t in message_history, the new turn is appended, except for a regenerate, which is rejected with a protocol error event (Vercel AI’s error chunk, AG-UI’s RUN_ERROR) instead of answering a turn the server doesn’t have: either the server never stored that message, as after a run that didn’t complete (see below), or the client holds a different conversation than the server, as a client in a stale tab does.
• Saving the stored record. Saving result.all_messages() over the stored history drops the turns a cut replaced, the client’s own messages included, so a client can remove them before moderation or review reads them. If you need an append-only record, keep the earlier version too, for example by appending each run’s result.new_messages() to a separate log, or, if your app doesn’t offer edit or regenerate, reject a request whose first message resubmits a stored user message id before calling the adapter.
• Histories without UI message ids. Matching a resubmitted message relies on the UI message ids the adapter keeps on the requests it loads, so message_history must be the history an adapter run produced, such as result.all_messages() saved in on_complete. A history persisted another way, such as from agent.run() or rebuilt from your own tables, carries no UI message ids, so every Vercel AI regenerate is rejected as stale and an edit is appended instead of replacing the turn.
• Tool results. A tool result from the client is kept only if it answers a tool call that is still unanswered at the end of message_history. Others are dropped with a warning.
• Capability loads. A CapabilityLoadDeltaPart in the new turn is dropped, like a compaction part, since message_history holds the loads the server recorded.
• Tool approvals. A run that resumes after an approval has no new turn: message_history ends with the calls awaiting approval, and the approvals and denials come from the request. Each one has to answer one of those calls. A request with an approval for a call that isn’t pending, such as a tool call the model didn’t make or one that was already answered, is rejected with a protocol error event, and no tool runs. Who may approve a call is still up to your application, as described in the warning above.
• Sanitization. The new turn is still client input, so the defaults above apply to it.
• Runs that don’t complete. on_complete only runs for a run that completes, so a turn whose run fails or is stopped isn’t in the stored history. Once the client holds an assistant response after it, such as the partial one of a stopped stream, the client’s copy of the turn isn’t used either, since only the messages after the client’s last assistant response are: the model doesn’t see that turn again, though the frontend may still show it. A regenerate() of that turn, which is how the AI SDK’s usual retry after an error or a stop works, is rejected as described above, since the server has no copy of the message to regenerate: have the frontend send the user’s message again as a new one (sendMessage) instead. To keep a cancelled run, also save RunCancelled.all_messages() in on_cancel. A client disconnect, as when the user stops useChat, is an external cancellation that triggers neither callback. A saved history that ends with a request no response answered, as a run cancelled before its first response leaves, gets that request again from the client, so the model sees it twice: until that’s handled (#11031), drop such a trailing request before saving, or don’t save a cancelled run that produced no response.

The default, manage_message_history='client', treats the client’s messages as the conversation and puts any message_history you pass in front of them, which suits server-only context such as a pinned preamble. If a client message’s UI message id already appears in message_history, the adapter warns, because the model sees that message twice. Only a history that went through an adapter carries those ids, so for one built any other way, such as by hand or from agent.run(), the adapter can’t tell: passing it as message_history while the client sends its whole conversation sends that conversation to the model twice, without a warning.

The client needs no changes, but it can save bandwidth by sending only its new messages; see the Vercel AI and AG-UI pages.

Saving to a conversation store

Instead of loading and saving the history yourself, you can give the adapter a ConversationStore (see Persistence) as conversation_store. Passing a store means the history is held on the server: manage_message_history is ignored, and the client’s new turn follows the 'server' rules above. A store also keeps the conversation’s usage, so it can’t be combined with the run methods’ message_history or usage arguments. The adapter then loads the conversation the request names (the Vercel AI chat id or the AG-UI threadId, unless you pass conversation_id), applies the rules above to the client’s new turn, runs the agent and saves what the run added. Always pass the authenticated user or tenant as conversation_scope: a conversation saved with another scope is treated as not found, as is one saved with any scope when you pass none, so dispatch_request answers 404 before running anything, and the other run methods reject the request with a protocol error event.

A store keys conversations by ID alone, and the ID comes from the client, so in a multi-tenant app conversation IDs must be unguessable or prefixed per scope, which you can do by passing your own conversation_id. Otherwise, a tenant who creates an ID first holds it, and its owner gets that 404. A client can also tell from the 404 that an ID it holds exists under another scope, but no tenant reaches another’s messages. A delete that names the owner’s scope leaves the conversation as it is, so freeing the ID takes one from your app without a scope.

conversation_store_ui.py

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent
from pydantic_ai.conversation_store import InMemoryConversationStore
from pydantic_ai.ui.vercel_ai import VercelAIAdapter

agent = Agent('openai:gpt-6.1-sol')

app = FastAPI()

store = InMemoryConversationStore()  # (1)


@app.post('/chat')
async def chat(request: Request) -> Response:
    user_id = 'user-123'  # (2)
    return await VercelAIAdapter.dispatch_request(
        request,
        agent=agent,
        conversation_store=store,
        conversation_scope=user_id,
    )

Use a database-backed store in production; see Persistence.

The user your own authentication identified.

With a store, the adapter also uses what the store knows that the messages don’t:

• Tool approvals and results. The stored conversation’s deferred_tool_requests say what each pending call is waiting for, so an approval or denial has to answer a call waiting on one, and a tool result a call waiting on a result. A result for a call that’s waiting on an approval is dropped with a warning.
• Regenerating and editing. The turn a resubmitted message replaces stays stored as a branch of the conversation, rather than being dropped from the record. The stored conversation only changes once the new run completes, so a regenerate whose run fails leaves it as it was.
• Usage limits. A run counts on from the stored conversation’s usage, so usage_limits, including the default request_limit of 50, cover the whole conversation rather than each request. A conversation that reaches a limit fails every later turn, since a failed run isn’t saved, so pass limits sized for a whole chat, or UsageLimits(request_limit=None).
• Concurrent requests. If another request saved the conversation while this one ran, this run’s turn isn’t saved, and the stream ends with a protocol error event asking the client to reload the conversation.
• Store failures. If the store itself raises while loading or saving, the client’s protocol error event, or the 500 that dispatch_request answers if the store fails before the run starts, says only that the conversation could not be loaded or saved, since a database error can hold queries or hostnames. The original exception’s type and text go to a UserWarning on the server, and its type is recorded on the current OpenTelemetry span, if one is recording. A UserError the store raises itself is the exception: it’s sent to the client as is, so keep queries, hostnames and credentials out of it.

After saving, the stream sends the ids of the stored ConversationEntrys to the client, so it can record them next to its own messages, for a rating, say: a transient Vercel AI data-conversation-saved part, or an AG-UI pydantic_ai_conversation_saved custom event, each with a conversation_id and a list of entry_ids. An adapter instance drives one run at a time, as from_request builds one per request.

A run that fails or is stopped isn’t saved, so the rules under “Runs that don’t complete” above still apply.