Skip to content
Docs Pydantic Docs

Agent-User Interaction (AG-UI) Protocol

The Agent-User Interaction (AG-UI) Protocol is an open standard introduced by the CopilotKit team that standardises how frontend applications communicate with AI agents, with support for streaming, frontend tools, shared state, and custom events.

Installation

The only dependencies are:

You can install Pydantic AI with the ag-ui extra to ensure you have all the required AG-UI dependencies:

Terminal
pip install 'pydantic-ai-slim[ag-ui]'

To run the examples you’ll also need:

  • uvicorn or another ASGI compatible server
Terminal
pip install uvicorn

Usage

There are three ways to run a Pydantic AI agent based on AG-UI run input with streamed AG-UI events as output, from most to least flexible. If you’re using a Starlette-based web framework like FastAPI, you’ll typically want to use the second method.

  1. The AGUIAdapter.run_stream() method, when called on an AGUIAdapter instantiated with an agent and an AG-UI RunAgentInput object, will run the agent and return a stream of AG-UI events. It also takes optional Agent.iter() arguments including deps. Use this if you’re using a web framework not based on Starlette (e.g. Django or Flask) or want to modify the input or output some way.
  2. The AGUIAdapter.dispatch_request() class method takes an agent and a Starlette request (e.g. from FastAPI) coming from an AG-UI frontend, and returns a streaming Starlette response of AG-UI events that you can return directly from your endpoint. It also takes optional Agent.iter() arguments including deps, that you can vary for each request (e.g. based on the authenticated user). This is a convenience method that combines AGUIAdapter.from_request(), AGUIAdapter.run_stream(), and AGUIAdapter.streaming_response().
  3. Build a stand-alone Starlette app with a single / route that calls AGUIAdapter.dispatch_request(). The same Starlette app can be mounted at a path in an existing FastAPI app.

When a run ends in first-party cancellation — ctx.cancel(), AgentRun.cancel(), a CancellationToken your server wires to a cancel endpoint, or a timeout= running out — the adapter closes any open text or tool events and emits a bare RUN_FINISHED. AG-UI currently has no cancelled outcome, so cancellation is not reported as RUN_ERROR. Pass an on_cancel callback (see the run_stream() example below) to persist the resumable message history from RunCancelled.all_messages().

Handle run input and output directly

This example uses AGUIAdapter.run_stream() and performs its own request parsing and response generation. This can be modified to work with any web framework.

run_ag_ui.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, RunCancelled
from pydantic_ai.ui import SSE_CONTENT_TYPE
from pydantic_ai.ui.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-6.1-sol', instructions='Be fun!')

app = FastAPI()


async def on_cancel(cancelled: RunCancelled) -> None:
    messages = cancelled.all_messages()  # the resumable history to persist
    print(f'cancelled after {len(messages)} messages')


@app.post('/')
async def run_agent(request: Request) -> Response:
    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        run_input = AGUIAdapter.build_run_input(await request.body())  # (1)
    except ValidationError as e:
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    adapter = AGUIAdapter(agent=agent, run_input=run_input, accept=accept)
    event_stream = adapter.run_stream(on_cancel=on_cancel)  # (2)

    sse_event_stream = adapter.encode_stream(event_stream)
    return StreamingResponse(sse_event_stream, media_type=accept) # (3)
  1. AGUIAdapter.build_run_input() takes the request body as bytes and returns an AG-UI RunAgentInput object. You can also use the AGUIAdapter.from_request() class method to build an adapter directly from a request.
  2. AGUIAdapter.run_stream() runs the agent and returns a stream of AG-UI events. It supports the same optional arguments as Agent.run_stream_events(), including deps. You can also use AGUIAdapter.run_stream_native() to run the agent and return a stream of Pydantic AI events instead, which can then be transformed into AG-UI events using AGUIAdapter.transform_stream().
  3. AGUIAdapter.encode_stream() encodes the stream of AG-UI events as strings according to the accept header value. You can also use AGUIAdapter.streaming_response() to generate a streaming response directly from the AG-UI event stream returned by run_stream().

Since app is an ASGI application, it can be used with any ASGI server:

Terminal
uvicorn run_ag_ui:app

This will expose the agent as an AG-UI server, and your frontend can start sending requests to it.

Handle a Starlette request

This example uses AGUIAdapter.dispatch_request() to directly handle a FastAPI request and return a response. Something analogous to this will work with any Starlette-based web framework.

handle_ag_ui_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.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-6.1-sol', instructions='Be fun!')

app = FastAPI()


@app.post('/')
async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent) # (1)
  1. This method essentially does the same as the previous example, but it’s more convenient to use when you’re already using a Starlette/FastAPI app.

Since app is an ASGI application, it can be used with any ASGI server:

Terminal
uvicorn handle_ag_ui_request:app

This will expose the agent as an AG-UI server, and your frontend can start sending requests to it.

Stand-alone ASGI app

When you don’t already have a Starlette/FastAPI app to mount the route on, build a minimal Starlette app whose single / route calls AGUIAdapter.dispatch_request():

ag_ui_app.py
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent
from pydantic_ai.ui.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-6.1-sol', instructions='Be fun!')


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent)


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

Since app is an ASGI application, it can be used with any ASGI server:

Terminal
uvicorn ag_ui_app:app

This will expose the agent as an AG-UI server, and your frontend can start sending requests to it.

Design

The Pydantic AI AG-UI integration supports all features of the spec:

The integration receives messages in the form of a RunAgentInput object that describes the details of the requested agent run including message history, state, and available tools.

These are converted to Pydantic AI types and passed to the agent’s run method. Events from the agent, including tool calls, are converted to AG-UI events and streamed back to the caller as Server-Sent Events (SSE).

A user request may require multiple round trips between client UI and Pydantic AI server, depending on the tools and events needed.

Features

State management

The integration provides full support for AG-UI state management, which enables real-time synchronization between agents and frontend applications.

In the example below we have document state which is shared between the UI and server using the StateDeps dependencies type that can be used to automatically validate state contained in RunAgentInput.state using a Pydantic BaseModel specified as a generic parameter.

ag_ui_state.py
from dataclasses import replace

from pydantic import BaseModel
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent
from pydantic_ai.ui import StateDeps
from pydantic_ai.ui.ag_ui import AGUIAdapter


class DocumentState(BaseModel):
    """State for the document being written."""

    document: str = ''


agent = Agent(
    'openai:gpt-6.1-sol',
    instructions='Be fun!',
    deps_type=StateDeps[DocumentState],
)
deps = StateDeps(DocumentState())


async def run_agent(request: Request) -> Response:
    # `dispatch_request` mutates `deps.state` from the request, so give each request its own copy.
    return await AGUIAdapter.dispatch_request(request, agent=agent, deps=replace(deps))


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

Since app is an ASGI application, it can be used with any ASGI server:

Terminal
uvicorn ag_ui_state:app --host 0.0.0.0 --port 9000

Tools

AG-UI frontend tools are seamlessly provided to the Pydantic AI agent, enabling rich user experiences with frontend user interfaces.

Context

Alongside messages, an AG-UI client can send a context array of description/value pairs describing things it considers relevant to the run: the originating platform, the requesting user, or a channel’s standing instructions. Every entry is a claim the client made — it can send any description/value it likes — so they describe a request, they never establish who is making it.

These entries are not passed to the model automatically, and they don’t belong in instructions. Instructions carry operator authority — they’re treated as your instruction to the model — so building them out of text a client sent lets a prompt injection inherit that authority. Delivering them as data denies them that authority but doesn’t make them safe: they’re still indirect prompt-injection input, so scope and re-authorize side-effecting tools from deps your server established, never from an entry’s description or value. See Mid-conversation system prompts and the trust model.

Read the entries off adapter.run_input.context and deliver them to the model as data. Facts your server established — the authenticated user, the workspace — are what go in instructions:

ag_ui_context.py
from dataclasses import dataclass

from ag_ui.core import Context
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent, RunContext
from pydantic_ai.ui.ag_ui import AGUIAdapter


@dataclass
class ChannelDeps:
    workspace: str  # (1)
    context: list[Context]  # (2)


agent = Agent('openai:gpt-6.1-sol', deps_type=ChannelDeps)
app = FastAPI()


@agent.instructions
def workspace(ctx: RunContext[ChannelDeps]) -> str:
    return f'You are answering in the {ctx.deps.workspace} workspace.'


@agent.tool
def frontend_context(ctx: RunContext[ChannelDeps]) -> list[str]:
    """Context the frontend says is relevant to this conversation."""
    return [f'{entry.description}: {entry.value}' for entry in ctx.deps.context]


def authenticated_workspace(request: Request) -> str:
    """Whatever your auth layer already established — a session, a signed token, an API key."""
    ...


@app.post('/')
async def run_agent(request: Request) -> Response:
    adapter = await AGUIAdapter[ChannelDeps, str].from_request(request, agent=agent)
    deps = ChannelDeps(workspace=authenticated_workspace(request), context=adapter.run_input.context)
    return adapter.streaming_response(adapter.run_stream(deps=deps))

Established by your server, so it can shape how the agent behaves.

Sent by the client, so it reaches the model as tool output the agent can read — never as an instruction.

To let a client-supplied fact change how the agent behaves, authenticate it first: verify the caller or channel, look up the policy your server holds for it, and write the instruction from that. The entry itself stays data.

Anything that isn’t meant for the model at all — a Slack channel ID, a locale — is better carried in forwardedProps, which the adapter passes through untouched as adapter.run_input.forwarded_props. Validating it proves shape, not identity: who the user is, what tenant they’re in, and what they’re allowed to do come from authenticated server state.

When the agent’s events reach you outside the request that serves the frontend, there’s no run input to read them off at all — see “Encoding events without a request”, where AGUIEventStream.thread_id and run_id take over as the source of the identity the protocol requires.

context, forwardedProps and parentRunId are read straight off run_input rather than through adapter properties of their own. The adapter’s properties — messages, toolset, state, conversation_id, deferred_tool_results — are the concepts every UI protocol shares and that the adapter itself feeds into the agent run. These three are AG-UI-specific and consumed only by your code, so they stay on the protocol object where their types are the protocol’s own.

Tool approval (interrupts)

Tools declared with requires_approval=True map onto AG-UI’s interrupt-aware run lifecycle. When the model proposes such a call, the run pauses and the adapter ends the SSE stream with a RUN_FINISHED event whose outcome.type is "interrupt" and whose outcome.interrupts[] describes each pending approval. The client renders an approval UI from that list and POSTs the next RunAgentInput with a resume[] array of ResumeEntry items addressing each interrupt.

The mapping the adapter applies (matching the AG-UI Python SDK field names):

AG-UI directionPydantic AI source / sink
Interrupt.reasonAlways "tool_call" for requires_approval=True tools
Interrupt.tool_call_idThe ToolCallPart.tool_call_id of the proposed call
Interrupt.idf"int-{tool_call_id}" (round-trips back to tool_call_id on resume)
Interrupt.metadataDeferredToolRequests.metadata.get(tool_call_id)
ResumeEntry.payload{ "approved": bool, "editedArgs"?: object, "reason"?: string }, validated against Interrupt.response_schema; omitting either optional field or sending it as null is accepted
payload.approved=TrueToolApproved
payload.editedArgsToolApproved.override_args (fully replaces the proposed args)
payload.approved=FalseToolDenied with message=payload.reason
status="cancelled"ToolDenied with message="Cancelled by user." regardless of payload
A payload that fails validation, or an interruptId that maps to no call the agent proposedA RUN_ERROR event that ends the run — see below

A payload that doesn’t validate against the advertised Interrupt.response_schema is a protocol error, not an outcome the user chose, so the run ends with a RUN_ERROR event naming the offending interrupt. This is the spec’s prescribed outcome once an agent validates against responseSchema (“Agents should handle missing or invalid resume payloads via RunError, not silent failures”), and it means a client-side serialization bug — a double-encoded editedArgs, a stringly-typed approved — is reported rather than looking to the user like the tool was denied. approved is required, so a payload that omits it is invalid too: send approved: false explicitly to deny, and to have your reason reach the model. One invalid entry fails the whole run, so no sibling approval in the same resume[] is applied. An interruptId without the int- prefix the adapter assigned can’t be mapped back to a tool call id at all, so it ends the run the same way. A prefixed id that matches no tool call the agent proposed ends the run the same way, since the agent rejects a result for a call it never made.

A validation failure stops the run before the agent resumes, so it never results in the tool running. Note that this is a guard against a malformed response, not an authorization boundary against the client — see Trust model for client-submitted messages.

This whole mapping applies to the resume[] the client sends. If you follow the trust-model advice and pass your own deferred_tool_results to the run method instead, the client’s resume[] is ignored entirely — including a malformed one, which is then neither validated nor reported.

The agent must include DeferredToolRequests in its output_type so the run can pause cleanly instead of erroring on the proposed call:

ag_ui_tool_approval.py
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent
from pydantic_ai.tools import DeferredToolRequests
from pydantic_ai.ui.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-6.1-sol', output_type=[str, DeferredToolRequests])


@agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    """Delete a file. Pauses on a `RUN_FINISHED` interrupt outcome until the user approves."""
    return f'deleted {path}'


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent)


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

On the resumed turn the agent re-executes the tool against the original tool_call_id, so only a TOOL_CALL_RESULT event is emitted for that id — no fresh TOOL_CALL_START. This preserves the audit trail the AG-UI spec requires.

See Deferred tools and human-in-the-loop tool approval for the underlying Pydantic AI primitive that also works outside AG-UI.

Events

To send events to the client while a run is in progress — for example progress updates from a long-running tool — emit a CustomEvent via ctx.emit():

ag_ui_custom_events.py
from dataclasses import dataclass

from pydantic_ai import Agent, CustomEvent, RunContext

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


@dataclass(kw_only=True)
class SearchIndexProgressEvent(CustomEvent):
    done: int
    total: int


@agent.tool
async def reindex(ctx: RunContext, total: int) -> str:
    for done in range(1, total + 1):
        # Do a unit of work, then tell the frontend how far along we are.
        await ctx.emit(SearchIndexProgressEvent(done=done, total=total))
    return f'Reindexed {total} documents'

Each event reaches the client as an AG-UI CustomEvent with its name and the result of to_payload() as its value — here, name='search_index_progress' and value={'done': 1, 'total': 3}. Events arrive as they are emitted, while the tool is still running.

The value shape is the same whether or not the event was emitted from inside a tool call, so a frontend written against one shape doesn’t break when the same event class is later emitted from somewhere else. Override to_payload() to control the shape — to name the fields the way the frontend expects, or to put the tool attribution on the wire:

ag_ui_custom_event_payload.py
from dataclasses import dataclass
from typing import Any

from pydantic_ai import CustomEvent


@dataclass(kw_only=True)
class SearchIndexPhaseEvent(CustomEvent):
    done: int
    total: int

    def to_payload(self) -> dict[str, Any]:
        return {
            'completed': self.done,
            'total': self.total,
            'toolCallId': self.tool_call_id,
        }

Returning an AG-UI BaseEvent from to_payload() sends that event verbatim instead, so an emitted event can also carry a protocol event such as a state snapshot. An event class declared ui=False is never forwarded, so events meant only for server-side consumers stay off the wire; nor is an event whose class this process never imported, since its opt-out travels on the class rather than the wire.

Pydantic AI tools can also attach AG-UI events to a tool result, by returning a ToolReturn object with a BaseEvent (or a list of events) as metadata. Unlike emitted events, these are part of the message and survive a message-history round-trip, which is what you want for state updates the frontend must be able to rebuild; the trade-off is that they are sent when the tool returns rather than while it runs.

ag_ui_tool_events.py
from dataclasses import replace

from ag_ui.core import CustomEvent, EventType, StateSnapshotEvent
from pydantic import BaseModel
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent, RunContext, ToolReturn
from pydantic_ai.ui import StateDeps
from pydantic_ai.ui.ag_ui import AGUIAdapter


class DocumentState(BaseModel):
    """State for the document being written."""

    document: str = ''


agent = Agent(
    'openai:gpt-6.1-sol',
    instructions='Be fun!',
    deps_type=StateDeps[DocumentState],
)
deps = StateDeps(DocumentState())


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent, deps=replace(deps))


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])


@agent.tool
async def update_state(ctx: RunContext[StateDeps[DocumentState]]) -> ToolReturn:
    return ToolReturn(
        return_value='State updated',
        metadata=[
            StateSnapshotEvent(
                type=EventType.STATE_SNAPSHOT,
                snapshot=ctx.deps.state,
            ),
        ],
    )


@agent.tool_plain
async def custom_events() -> ToolReturn:
    return ToolReturn(
        return_value='Count events sent',
        metadata=[
            CustomEvent(
                type=EventType.CUSTOM,
                name='count',
                value=1,
            ),
            CustomEvent(
                type=EventType.CUSTOM,
                name='count',
                value=2,
            ),
        ]
    )

Since app is an ASGI application, it can be used with any ASGI server:

Terminal
uvicorn ag_ui_tool_events:app --host 0.0.0.0 --port 9000

Protocol version compatibility

Pydantic AI supports every ag-ui-protocol release from 0.1.10 on, and features added after that floor are gated on the version you have installed rather than requiring an upgrade.

That gate runs in both directions. On the way out, content an older protocol version can’t express is downgraded or omitted — see AGUIAdapter.ag_ui_version for the negotiated thresholds. On the way in, a message role or input content type your installed ag-ui-protocol has no class for is skipped with a UserWarning naming the tag, and the rest of the request runs — so a frontend on a newer protocol version than your server keeps working, minus the content your install has no type for. For instance, a gateway that forwards image attachments as typed multimodal content (ag-ui-protocol >= 0.1.15) still delivers the accompanying text to an agent running on an older install.

What gets skipped is decided by the tag alone: any role or type string the installed models don’t declare qualifies, so a client that misspells "txet" is skipped with the same warning as one sending genuinely newer content — the server has no way to tell those apart. The skip is scoped to well-formed items: a message must still carry a string id, the field every AG-UI message type requires.

Everything else is still rejected with 422 Unprocessable Entity — a payload that is malformed under a role or type the install does know, a role or type that isn’t a string at all, and a body that isn’t valid JSON. If you see the warning and the content was real, upgrading ag-ui-protocol is what makes it reach your agent.

Trust model

AG-UI’s RunAgentInput.messages is fully client-controlled. The AGUIAdapter applies defaults to strip untrusted parts before the agent runs — see Trust model for client-submitted messages in the UI adapter overview, which covers system prompts, file URL schemes, uploaded files (allow_uploaded_files), workspace references (strip_workspace_refs), and unresolved tool calls. Those defaults don’t make client-submitted history authentic — see Trust boundary for client-supplied history.

Records carried as activity messages

Some of what Pydantic AI keeps in message history isn’t a chat message, so the adapter sends it to the frontend as an activity message with a reserved activityType, streamed as an ACTIVITY_SNAPSHOT (ag-ui-protocol 0.1.19 or later) or included in AGUIAdapter.dump_messages output:

AGUIAdapter.load_messages turns these activity messages back into the history they came from, but only if the frontend sends them. The JavaScript client, @ag-ui/client, whose HttpAgent most AG-UI frontends use, doesn’t: it keeps activity messages in agent.messages for the UI to render, and leaves them out of every request it sends. AG-UI treats activities as client-side state, which a backend that needs them keeps itself. It has done so since it added activity messages in 0.0.42; see the filter in prepareRunAgentInput.

So when the frontend holds the message history through @ag-ui/client, these records are missing from the history the next run loads:

  • Deferred tools a ToolAvailabilityDeltaPart made available are hidden from the model again, and tools that appeared mid-run are sent in tools[] instead.
  • The model’s request includes everything before a compaction again, without the compaction item.
  • Enqueued messages, such as a background task’s result, are no longer in the history.
  • Files kept with preserve_file_data are dropped.

The model then sees a different conversation than it did on the earlier run, and the request no longer starts with the same bytes, so the prompt cache misses as well.

To keep these records, keep the history on the server: persist the agent’s messages keyed by the thread ID and pass them to the adapter as message_history, with the client’s messages supplying only the latest turn, as described under Trust model for client-submitted messages. The records then never depend on the client.

If the frontend must hold the history, it can send the records back itself. This middleware puts the pydantic_ai_* activity messages from agent.messages back into each request, in their original positions:

import { HttpAgent } from '@ag-ui/client';

const agent = new HttpAgent({ url: '/agent' });
agent.use((input, next) => {
  const sent = new Map(input.messages.map((message) => [message.id, message]));
  const messages = agent.messages.flatMap((message) =>
    sent.has(message.id)
      ? [sent.get(message.id)!]
      : message.role === 'activity' && message.activityType.startsWith('pydantic_ai_')
        ? [message]
        : [],
  );
  return next.run({ ...input, messages });
});

The records the client sends back this way are under its control, like the rest of a client-held history; see Trust model.

Compaction

CompactionParts are carried as pydantic_ai_compaction activity messages, so compacted conversations keep working when the frontend holds the message history and sends those activity messages back. @ag-ui/client doesn’t send them back; see Records carried as activity messages. A compaction item submitted by the frontend is honored — the conversation stays compacted — with two caveats. First, it is never trusted to stand in for the system prompt: whichever prompt applies per System prompts and instructions still reaches the model on every request. Second, if the run also receives server-side message_history (the server-side persistence pattern), frontend compaction items are ignored — everything before a compaction item is hidden from the model, so honoring one from the frontend would let it hide the server’s stored history. See Client-held history for the trade-offs and the recommended server-side pattern.

Enqueued messages

Messages injected mid-run with ctx.enqueue(), such as a background task’s result, are streamed as an ACTIVITY_SNAPSHOT with activityType: "pydantic_ai_enqueued_messages" and messageId set to the enqueue ID (ag-ui-protocol 0.1.19 or later). Its content.messages holds the delivered messages in the shape AGUIAdapter.dump_messages gives them. The client keeps the activity in its message list, and a UI only shows activity types it has a renderer for. If the frontend sends the activity back on the next run, AGUIAdapter.load_messages restores the messages at the position they were delivered in. @ag-ui/client doesn’t send activity messages back, so unless the history is kept on the server or the frontend sends them back as described under Records carried as activity messages, the enqueued messages are missing from the next run’s history.

An enqueued system prompt, such as a capability’s announcement, is only sent when the frontend owns the system prompt. Otherwise it would be stripped on the way back in, like any other system prompt the client sends, so it stays on the server. The other messages get the same handling as the rest of the submitted history. They round-trip as dump_messages output does, so TextContent.metadata is dropped and the text comes back as plain user text.

MCP Apps

After the result of an MCP Apps tool call, the stream carries an ACTIVITY_SNAPSHOT with activityType: "mcp-apps" (ag-ui-protocol 0.1.19 or later). Its content: resourceUri, toolInput, result (the full MCP CallToolResult), and serverId and serverHash (both the toolset id, when set). This is the shape the AG-UI MCP Apps middleware emits, so renderers built for it can show it; they load the page through your route, as below. Dumped history keeps it as an ActivityMessage; loading ignores it.

CopilotKit’s MCP Apps renderer loads the page and relays its requests by posting a run with forwardedProps.__proxiedMCPRequest (serverId, method, params) to your AG-UI endpoint, and reads the answer from RUN_FINISHED.result. AGUIAdapter runs the agent for these, so answer them in your route before calling dispatch_request: get the result from the MCP Apps relay, and send it as RUN_STARTED followed by RUN_FINISHED with that result.

Assistant message identity

Every streamed tool call carries a parentMessageId naming the assistant message that owns it, and that message is always announced by a TEXT_MESSAGE_START first — including when the model’s response is nothing but tool calls, in which case the message carries no content and is closed immediately. A frontend that rebuilds the conversation from the event stream alone therefore never has to infer that a message exists.

Tool calls attach to whichever assistant message is open when they stream: text appearing before them in the same response shares their message, and text appearing after starts a new one that any later tool calls attach to instead. AGUIAdapter.dump_messages splits text and tool calls the same way. It splits on more than that, though: a compaction part always starts a new assistant message when history is loaded, a reasoning part does so from ag-ui-protocol 0.1.11, and a file part does so only under AGUIAdapter.preserve_file_data. The stream splits on none of them, so a response interleaving one of those with tool calls yields more messages loaded than streamed.

Message ids across round-trips

AGUIAdapter.load_messages keeps each inbound message’s id in the reserved __pydantic_ai__ key of the message it produces, and AGUIAdapter.dump_messages uses it as the id again, so a history the client sent comes back with the ids the client assigned. Each ModelRequest or ModelResponse holds one id, so when consecutive AG-UI messages merge into one message, such as the tool results of parallel tool calls or a system message followed by a user message, only the last id is kept and it goes on the last message dumped from it; the others get fresh ids. Messages produced by an agent run have no kept id and get a fresh UUID on every dump.

Preserving failed tool outcomes

AG-UI’s ToolCallResultEvent has no error or outcome field. Although encrypted reasoning continuity is the intended use of ReasoningEncryptedValueEvent, it is also AG-UI’s standard event for attaching encrypted_value to a message or tool call. Pydantic AI uses that attachment mechanism with a namespaced payload to preserve a non-'success' outcome on a ToolReturnPart when using ag-ui-protocol >= 0.1.11. That covers 'retried', which is how a tool retry answers its call, as well as 'failed', 'denied', and 'interrupted'.

If the client sends those messages back on a later run, the adapter restores the outcome. This is a history-continuity mechanism: a 'failed', 'denied', or 'retried' result also carries its content in ToolMessage.error, but neither carrier guarantees that a frontend visually renders the result as an error. Event streams produced with earlier protocol versions have no metadata carrier for the outcome, so reloading them reconstructs a 'retried' or 'denied' result as 'failed' (from that error field) and an 'interrupted' one as 'success'.

Preserving retry feedback

AGUIAdapter.dump_messages writes each RetryFeedbackPart as a message in the voice the model was shown it in — a UserMessage for validation errors and ModelRetry feedback, a SystemMessage for 'no_output' feedback, the one cause that speaks for the harness — holding the same rendered text. A dumped history therefore carries one of those wherever the agent retried its own output, including for an agent with no configured system_prompt at all. The part itself rides in that message’s encrypted_value under the same namespaced payload as failed tool outcomes, because neither message type has another metadata slot, and the rendered text alone can’t give back the part’s retry or its raw content.

On the way back in, only that marker rebuilds a RetryFeedbackPart: a message the frontend wrote loads as a plain SystemPromptPart or UserPromptPart. The marker separates provenance rather than proving it — the frontend echoes it, so it can also forge it, and can name any cause — which is why, under manage_system_prompt='server', a rebuilt 'no_output' part is demoted to fenced user text rather than sent in the system voice, while the frontend’s system messages are stripped.

The carrier here is the encrypted_value field, which ag-ui-protocol added to BaseMessage (and so to both SystemMessage and UserMessage) in 0.1.11 — the same floor as tool outcomes, which ride the same payload on the live event stream via ReasoningEncryptedValueEvent. Below 0.1.11 the field doesn’t exist, so the feedback reloads as a plain SystemPromptPart or UserPromptPart — the message’s own voice — holding the rendered text, and dump_messages warns whenever the history holds a RetryFeedbackPart. Upgrading to 0.1.11 or later preserves the part and silences the warning.

Preserving files across round-trips

AG-UI has no native representation for agent-generated files (FilePart) or UploadedFile references, so they are omitted from dump_messages output by default. Set AGUIAdapter.preserve_file_data to True to round-trip them through reserved pydantic_ai_* activity messages, which a frontend completes by sending those activity messages back on the next request. @ag-ui/client doesn’t do that; see Records carried as activity messages. This is a representation opt-in, not a security one: an UploadedFile reconstructed from a round-tripped activity message is still subject to the inbound allow_uploaded_files gate before it reaches the agent.

System prompts and instructions

Pydantic AI supports two ways to provide guidance to the model: system_prompt (stored in the message history as SystemPromptParts) and instructions (injected fresh on every request, never persisted). When you control the server side, instructions is the recommended default.

The rest of this section only matters if you use system_prompt. If you only use instructions, there’s nothing to configure — they’re always applied regardless of the AG-UI message history.

For system_prompt, you choose who owns it with the manage_system_prompt parameter on AGUIAdapter:

  • 'server' (default): the agent’s configured system_prompt is authoritative. Any SystemMessage sent by the frontend is stripped with a warning (a malicious client could otherwise inject arbitrary instructions via crafted API requests), and the agent’s own system prompt is reinjected at the head of the first request via the ReinjectSystemPrompt capability.
  • 'client': the frontend owns the system prompt. Frontend SystemMessages are preserved as-is, and the agent’s configured system_prompt is not injected — the caller is fully responsible for sending it on every turn if desired. To opt into fallback-to-configured behavior, add the ReinjectSystemPrompt capability to your agent. Retry feedback is preserved along with them, so a frontend that attaches the encrypted_value marker to a SystemMessage of its own writing also reaches the model in the system voice. Reach for 'client' only when the frontend is as trusted as the server.
ag_ui_client_managed_system_prompt.py
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent
from pydantic_ai.ui.ag_ui import AGUIAdapter

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

app = FastAPI()


@app.post('/')
async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(
        request, agent=agent, manage_system_prompt='client'
    )

Channels

The agent you exposed over AG-UI can also power a bot in Slack or another messaging platform. The CopilotKit Channels SDK receives platform events, runs your agent over AG-UI, and renders its response as native platform content.

How it fits together

For managed Slack, CopilotKit Intelligence holds the Slack credentials and delivers each turn to a long-running Node process built with @copilotkit/channels. That process sends the conversation to your Pydantic AI server over AG-UI and returns the streamed response to Slack.

Slack  ──►  CopilotKit Intelligence  ──►  channel process (Node)  ──►  Pydantic AI server (AG-UI)

Follow the CopilotKit guide to create the channel process and point its AG-UI client at your Pydantic AI server. When the process handles a platform event, it passes the triggering message to the agent and can attach platform and user details as AG-UI context entries.

AG-UI context is client-provided data, so Pydantic AI deliberately does not put it in the model prompt automatically. Run the channel process alongside the ag_ui_context.py server above: it reads adapter.run_input.context, keeps authenticated workspace data separate, and exposes the channel entries through the frontend_context tool rather than treating them as instructions.

Terminal
uvicorn ag_ui_context:app

Start the channel process as described in the CopilotKit guide. It needs a long-running host because a serverless request handler cannot own its persistent gateway connection.

Slack

The managed Slack connection is configured in CopilotKit Intelligence, which walks you through creating the Slack app and holds its credentials. Mention the bot in a real workspace and test a direct message to verify the platform connection, gateway listener, AG-UI server, and reply path together.

Other platforms

Managed and developer-operated connections have different setup and support. See the Channels SDK reference for the current managed platforms, direct adapters, and provider-specific guides.

Examples

For more examples see pydantic_ai_examples.ag_ui, which includes a server for use with the AG-UI Dojo.

The examples above serve AG-UI over SSE, where each run is its own request. pydantic-ai-ws-agent is a community example that carries the same protocol over a WebSocket instead, using chanx. Because the connection is bidirectional and long-lived, one run streams to several browser tabs at once and any of them can approve a tool call; message history is kept on the server. The adapter it builds on is a copy-in chanx-kit component.