AG-UI

Overview#

The Agent-User Interaction (AG-UI) protocol standardizes how frontend applications communicate with agents. In AG2, ag2.ag_ui.AGUIStream bridges an Agent to AG-UI event streams.

This solves common integration problems:

• Streaming agent output to UI clients
• Emitting tool-call lifecycle events
• Synchronizing shared state snapshots
• Supporting human-in-the-loop checkpoints through frontend actions and input-required flows

For protocol background, see AG-UI Protocol introduction.

When to use AG-UI vs direct integration#

Use AG-UI when you want a reusable UI contract across clients and frameworks.

Supported capabilities#

AG2 speaks AG-UI 1.0. Verified AG-UI features are supported in AG2:

• Streaming text events (TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_END, TEXT_MESSAGE_CHUNK)
• Backend tool lifecycle events (TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END, TOOL_CALL_RESULT)
• Frontend-tool dispatch (TOOL_CALL_CHUNK for client tools in RunAgentInput.tools)
• Shared-state snapshots (STATE_SNAPSHOT) of the client's state and the server's variables — see Shared state
• Application context (RunAgentInput.context) reaches the model — see Application context
• Human input checkpoints (input_required surfaced as user-visible message events)
• Run interrupts: an agent pauses on a question and a later run on the same thread answers it (RUN_FINISHED with an interrupt outcome, resumed through resume) — see Asking the human a question
• Tool-call approval: a gated tool puts the call itself to the client before it runs, over the same interrupt lifecycle — see Gating a tool call instead of asking a question
• Tool results in content parts: text, images, audio, video, documents and provider file handles — see Tool results
• Delegations as subagent invocations (SUBAGENT_STARTED, SUBAGENT_FINISHED, SUBAGENT_ERROR) — see Delegations
• Pending client tool calls named on the run's success outcome (pendingToolCallIds)
• Protocol version declared on every RUN_STARTED — see Protocol version
• Capabilities declared before any run starts (GET) — see Capabilities
• Input from a newer client served, with what the server does not recognise stripped — see Input from a newer client
• Every run ends with RUN_FINISHED or RUN_ERROR — see How a run ends

Installation#

Install AG2 with AG-UI support, the extra for your model provider (openai below), and the HTTP server used in the examples:

pip install "ag2[ag-ui,openai]" fastapi uvicorn

ag2[ag-ui] installs ag-ui-protocol 1.x, the Python SDK for AG-UI 1.0, so the backend sits beside a frontend on @ag-ui/* 1.0.

Basic server example#

Use the manual-dispatch pattern when you want full control over auth, logging, and middleware:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse

from ag2 import Agent
from ag2.ag_ui import AGUIStream, read_run_input
from ag2.config import OpenAIConfig

agent = Agent(
    name="support_bot",
    prompt="You help users with billing questions.",
    config=OpenAIConfig(model="gpt-4o-mini"),
)

stream = AGUIStream(agent)
app = FastAPI()

@app.get("/chat")
@app.get("/chat/capabilities")
async def capabilities() -> JSONResponse:
    return JSONResponse(stream.capabilities().model_dump(by_alias=True, exclude_none=True))

@app.post("/chat")
async def run_agent(request: Request) -> Response:
    try:
        message = read_run_input(await request.body())
    except ValueError:
        return JSONResponse({"error": "invalid AG-UI RunAgentInput body"}, status_code=400)
    return StreamingResponse(
        stream.dispatch(message, accept=request.headers.get("accept")),
        media_type="text/event-stream",
    )

The GET routes expose the agent's capabilities. The body of a POST is read with read_run_input rather than declared as a RunAgentInput parameter. A newer client may send a message role or content part this server does not know, and FastAPI would refuse the whole request with 422; read_run_input strips it and serves the run (see Input from a newer client). A body it cannot read is refused with 400 before any stream starts. The stream is Server-Sent Events whatever Accept says, so the content type is text/event-stream, never the client's Accept copied back.

Run it:

uvicorn run_ag_ui:app --reload --port 8000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from ag2 import Agent
from ag2.ag_ui import AGUIStream
from ag2.config import OpenAIConfig
from fastapi import FastAPI
from starlette.routing import Route

agent = Agent(
    name="support_bot",
    prompt="You help users with billing questions.",
    config=OpenAIConfig(model="gpt-4o-mini"),
)
app = FastAPI()
stream = AGUIStream(agent)
endpoint = stream.build_asgi()
app.routes.append(Route("/chat", endpoint))
app.routes.append(Route("/chat/capabilities", endpoint))

Test the endpoint#

Check capability discovery before starting a run:

curl http://127.0.0.1:8000/chat/capabilities

Then start a run:

curl -N -X POST http://127.0.0.1:8000/chat \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "threadId": "thread-1",
    "runId": "run-1",
    "protocolVersion": "1.0",
    "messages": [{"id": "m1", "role": "user", "content": "Hello"}],
    "state": {},
    "context": [],
    "tools": []
  }'

Example stream (truncated):

data: {"type":"RUN_STARTED","threadId":"thread-1","runId":"run-1",...}
data: {"type":"TEXT_MESSAGE_CHUNK","delta":"Hello! How can I help?",...}
data: {"type":"RUN_FINISHED","threadId":"thread-1","runId":"run-1",...}

How a run ends#

Every run ends with exactly one RUN_FINISHED or RUN_ERROR, and the response body then closes normally.

• A failure inside the run — the model call, a tool, input that cannot be mapped — ends it with RUN_ERROR, and is logged with its traceback on the ag2.ag_ui logger. dispatch does not re-raise it: the response has already started 200 OK, and the failure has been reported on it. Watch the logger, not your endpoint's exception handler.
• A run the server stops — await stream.aclose(), or A2UIServer's shutdown — first closes whatever it had open (a text message, a tool call, a subagent invocation), then finishes with RUN_FINISHED and a cancelled outcome.

Token usage#

Both run-terminating events carry the run's token usage, broken down per provider and model configuration:

data: {"type":"RUN_FINISHED","threadId":"thread-1","runId":"run-1",
       "usage":[{"provider":"openai","model":"gpt-5","inputTokens":1284,
                 "outputTokens":96,"totalTokens":1380,"cachedInputTokens":1024}]}

RUN_ERROR carries the same field, reporting what the run had spent before it failed, so a crashed run is not accounted for as free. (RUN_ERROR has no threadId / runId — the protocol defines neither for it; a client correlates the run from RUN_STARTED on the same event stream.)

The figures cover every model call in the run: the whole tool loop, delegated sub-agents, history compaction and memory aggregation. They come from the same accounting events as AgentReply.usage() for the same run.

An entry carries inputTokens, outputTokens, totalTokens, reasoningTokens, cachedInputTokens and cacheWriteInputTokens, in AG-UI 1.0's accounting, so entries from different providers add up:

• inputTokens and outputTokens are totals. Input counts every prompt token the call was charged for, including tokens read from and written to the provider's cache. Output counts every generated token, reasoning included. Where a provider reports these counts beside a smaller figure, AG2 adds them in before the entry leaves: Anthropic's cache counts go into the input, and Gemini's thinking tokens into the output.
• cachedInputTokens, cacheWriteInputTokens and reasoningTokens are parts of those totals, never additions to them. Cache reads and writes are reported apart, because providers price them differently.
• totalTokens is always inputTokens + outputTokens. It is computed, not copied from the provider.
• A count the provider did not report is absent, not 0. cachedInputTokens absent means "never measured"; 0 means the provider measured no cache hit. An input or output the provider did not report leaves totalTokens absent too.

The correction applies only to what reaches the wire. AgentReply.usage() keeps each provider's own figures, so a server-side total can differ from the AG-UI one for the same run: for Anthropic, prompt_tokens excludes the cache counts that inputTokens includes.

One entry is emitted per distinct provider/model pair, in order of first appearance in the run. A delegated sub-agent's calls are attributed to the pairs that made them, however many it used. A run that spent nothing omits usage entirely.

Usage is reported per run, not per turn. A turn that pauses on a question is carried by several runs, and each reports only its own spend: the run that ends on the interrupt reports what was spent up to the pause, and the run that resumes the turn — or cancels it — reports what was spent after.

The A2UI AG-UI transport (A2UIServer(..., transport=AgUiTransport())) reports the same field on the same two events, so what a client can show does not depend on which endpoint it connected to.

Protocol version#

Every RUN_STARTED declares the AG-UI version AG2 speaks, as "protocolVersion": "1.0". AG2 reads the version a client declares on RunAgentInput:

Capabilities#

A GET on the endpoint returns what the agent can do, in the protocol's AgentCapabilities shape, before any run starts. A GET on {path}/capabilities returns the same document: that is where other AG-UI integrations serve it. Both routes are registered in the basic server example. A2UIServer serves both paths automatically; with build_asgi(), register its endpoint on both routes as shown in the simpler example above.

The document is read off the agent, because it is served before any run: a hitl_hook or tools passed to one dispatch call are not known to it. A group AG2 cannot vouch for is omitted, which the protocol reads as "says nothing", not as "unsupported". A delegation tool from agent.as_tool(...) records the agent behind it, including when placed inside a Toolkit.

multimodal.input follows the configured provider mapper. It cannot know the capabilities of an individual model. An inbound part the provider cannot take is skipped with a warning naming its kind and media type; the run continues. An empty user message adds nothing, and an empty tool result answers with the empty string. Outside AG-UI, provider mappers still reject unsupported input. Invalid options on an otherwise supported part remain errors.

Gemini function calls carry their thought_signature as REASONING_ENCRYPTED_VALUE, bound to the tool call id. Clients must restate its encryptedValue on the call in later run history so Gemini can continue it.

Input from a newer client#

AG-UI 1.0 lets a newer client send what an older server does not describe, and requires the server to serve the run anyway. read_run_input, build_asgi() and A2UIServer strip it before validation:

• a property the protocol does not define, on the input or on anything in it;
• a message with a role this server does not know;
• a content part of a type this server does not know;
• a content part whose source is of a kind this server does not know. The whole part goes, since one left without its source would be malformed.
• a resume entry with a status this server does not know. The whole entry goes, leaving any held interrupt unanswered.

A message left with no parts is treated like any other empty message: a user message adds nothing to the turn, and a tool message still answers its call, with the empty string. Each removal logs one warning on the ag2.ag_ui logger, naming its JSON Pointer, such as /messages/0/content/1. Open objects — state, forwardedProps, metadata, a tool's parameters — are values rather than protocol members and are passed on whole. A known field carrying a value the schema rejects is malformed, not unrecognised, and the body is refused with 400. dispatch strips unknown properties from a RunAgentInput you validated yourself, but a role or part type your own validation rejected never reaches it.

Shared state#

The run's state seeds the turn's variables: tool code reads the client's keys from context.variables. They are merged with the agent's own variables and the variables= passed to dispatch, and the framework's reserved keys are removed from what the client sent.

A STATE_SNAPSHOT replaces the client's state wholesale, so each one carries all of it: the client's keys and the server's variables together. One is sent as the run starts, when that differs from what the client sent, and one as it ends, when the variables changed since the last one sent. A value with no JSON form, such as a client or a connection object, is left out; None is kept.

A state that is not a JSON object, such as a list or a string, is served but seeds no variables, and no snapshot is sent for it.

The A2UI transport seeds variables from state the same way, and sends no snapshots.

Application context#

RunAgentInput.context carries entries the application shares with the agent, such as the page the user is on. They reach the model as one block, added to the run's system prompt after the agent's own prompt:

## Context from the application

- Current page: /billing/invoices
- Plan: pro

Each line is an entry's description, then its value exactly as the client sent it. The entries are prose for the model, not variables: tool code does not see them. To render them differently, pass dispatch a context_prompt= function that takes the entries and returns the block, or None for no block. context_prompt=None leaves the entries out. The A2UI transport adds the same block.

Tool results#

A server tool's result reaches a 1.0 client as content parts. A result that is a single piece of text with no metadata is sent as a plain string. Anything else is sent as a list, and a part's metadata travels on it:

A client that declares no protocol version predates 1.0 and reads a tool result as a string only. It gets the result's text parts, joined in order. Media parts are dropped, with a warning naming the dropped kinds. Nothing is invented in their place, and a result that was all media arrives as "".

The other direction works the same way. A client may answer a client tool call with a list of parts, such as a screenshot next to its caption, and the model sees each part. A part with a provider file handle (a file source) reaches the model when it names no provider, or names the provider the run is configured with. A handle issued by another provider is skipped, and a warning names both providers, never the handle. A tool message left with no parts after that still answers its call: the model sees the empty string.

Delegations#

Each delegation to a sub-agent (agent.as_tool(...), run_subtask) reaches the client as a subagent invocation with its own subagentRunId, so two parallel delegations to the same agent can be told apart:

SUBAGENT_STARTED  {"subagentRunId":"4f1c...","name":"researcher","description":"<the objective>"}
SUBAGENT_FINISHED {"subagentRunId":"4f1c...","result":"<what it returned>"}

A failed delegation ends with SUBAGENT_ERROR carrying the error message, and the run carries on: the parent's model is told the delegation failed and may recover. A delegation that is cancelled or runs out of time ends with SUBAGENT_ERROR too, with the message cancelled (cancelled: <reason> when there is one) or expired: the protocol has no outcome for a stopped invocation. Every invocation is closed before its run ends. One still open when the run finishes is closed with SUBAGENT_ERROR. The invocation events carry no usage. The run's own usage already includes what the delegation spent. Delegations are no longer reported as STEP_STARTED / STEP_FINISHED.

When a delegated agent asks the human a question, the question reaches the client like any other interrupt, tagged with the invocation's subagentRunId. Before the run ends on it, the invocation is closed as suspended, naming that interrupt. The resuming run announces the invocation again under the same id before anything else about it, and then finishes it:

run-1:  RUN_STARTED → SUBAGENT_STARTED → … → SUBAGENT_FINISHED (suspended) → RUN_FINISHED (interrupt)
run-2:  RUN_STARTED → SUBAGENT_STARTED (same id) → … → SUBAGENT_FINISHED → RUN_FINISHED

Asking the human a question#

An agent can stop mid-run and ask. When a tool calls context.input(...), the question leaves as the outcome of the run that raised it, and the turn is held in the server — suspended exactly where it stopped — until a later run on the same thread answers it.

@agent.tool
async def book_flight(context: Context, flight: str) -> str:
    """Book a flight, once the traveller confirms it."""
    answer = await context.input(f"Book {flight}? (yes/no)")
    return "Booked." if answer.strip().lower() == "yes" else "Cancelled."

The exchange ends normally — not with an error — carrying the question:

data: {"type":"RUN_FINISHED","threadId":"thread-1","runId":"run-1",
       "outcome":{"type":"interrupt","interrupts":[
         {"id":"e2c1...","reason":"input_required","message":"Book LH441? (yes/no)",
          "responseSchema":{"type":"string","title":"Answer"},
          "expiresAt":"2026-01-01T12:15:00+00:00",
          "metadata":{"ag2":{"proof":"g7Qk..."}}}]}}

The client answers with a new run id on the same thread, naming the interrupt in resume:

{"threadId": "thread-1", "runId": "run-2", "messages": [],
 "resume": [{"interruptId": "e2c1...", "status": "resolved", "payload": "yes",
             "metadata": {"ag2": {"proof": "g7Qk..."}}}]}

context.input returns "yes" inside the tool, and the turn runs on to completion in that second exchange — RUN_STARTED for the resuming run, then whatever the agent does next. A client that has given up sends "status": "cancelled" instead. The turn is stopped, and the run finishes with a cancelled outcome and no result. Before it ends, the run sends whatever the turn did while it waited, and closes anything the turn still had open. The thread then holds nothing, so the next run on it is an ordinary new run.

run-1:  RUN_STARTED → TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END → RUN_FINISHED (interrupt)
run-2:  RUN_STARTED → TOOL_CALL_RESULT → TEXT_MESSAGE_CHUNK → RUN_FINISHED

A paused turn keeps working while it waits: another tool call from the same model response can finish in the meantime. Nothing it emits is lost — it is sent at the start of the run that resumes the turn. Questions asked at once — two parallel tool calls each calling context.input — go out one per run: the run that answers the first ends on the second question, and a queued question's expiresAt already counts the time it spent waiting for its turn.

Two things must travel back untouched: the interrupt id, and the metadata envelope. The envelope carries proof that the answer comes from whoever the question was put to — an answer attributed to a human is the input an agent trusts most, and an interrupt id alone is a bare string in a request body. A resume that carries the envelope must carry the one issued; a resume that carries none is accepted unless the server is built with require_resume_proof=True (on AGUIStream and on the A2UI AgUiTransport). The protocol does not make clients echo it, and @ag-ui/client does not do it for you: to get the check, copy the interrupt's metadata into the resume entry yourself.

import { buildResumeArray } from "@ag-ui/client";

const [interrupt] = agent.pendingInterrupts;
await agent.runAgent({
  resume: buildResumeArray(agent.pendingInterrupts, {
    [interrupt.id]: { status: "resolved", payload: "yes", metadata: interrupt.metadata },
  }),
});

A client that cannot set a resume entry's metadata cannot answer these interrupts on a server running with require_resume_proof=True. CopilotKit's useInterrupt is one: its resolve() and cancel() send no metadata.

Every refusal arrives as RUN_ERROR with a machine-readable code, never as a stream that simply stops:

A refusal that is not the turn's own fault — a bad payload, an id it is not waiting on — leaves the held turn resumable by a legitimate answer, without extending its deadline.

Every refusal is a lone RUN_ERROR before RUN_STARTED. A fresh run on a thread with a held interrupt is refused; omitting an answer does not release the turn. An unrecognised entry beside a valid one is dropped with a warning, and the valid answer is applied. When no turn is held for the thread — unknown id, already answered, expired, or lost to a restart — the resume is dropped with a warning and the request runs as an ordinary run, as the protocol recommends.

A runnable version of this exchange — server and client in one script — is in examples/ag_ui/human_input.py.

Gating a tool call instead of asking a question#

A tool wrapped in approval_required() pauses the same way, over the same resume. But the client is being asked to approve a call rather than to type an answer, and it should not have to read the prose to tell the two apart — so the interrupt says which it is:

data: {"type":"RUN_FINISHED","threadId":"thread-1","runId":"run-1",
       "outcome":{"type":"interrupt","interrupts":[
         {"id":"9236...","reason":"tool_call",
          "message":"Agent wants to call the tool:\n`delete_account`, {\"user_id\": \"abc-123\"}\n...",
          "toolCallId":"7880...",
          "responseSchema":{"type":["string","boolean"],"title":"Approval"},
          "expiresAt":"2026-01-01T12:15:00+00:00",
          "metadata":{"ag2":{"proof":"g7Qk..."}}}]}}

Answer it with true or false — a client that drew a pair of buttons should not have to know which words this server accepts:

{"threadId": "thread-1", "runId": "run-2", "messages": [],
 "resume": [{"interruptId": "9236...", "status": "resolved", "payload": true,
             "metadata": {"ag2": {"proof": "g7Qk..."}}}]}

false refuses the call without failing the turn: the model is handed the middleware's refusal as the tool's result and carries on, and the run still finishes with a success outcome. Because a gated call is a paused tool call, it splits across the two exchanges exactly as above — toolCallId names the TOOL_CALL_START the first exchange already emitted, so a client has something to attach the prompt to.

With a hook, nothing changes#

Supplying a human-input hook (Agent(..., hitl_hook=...), or hitl_hook= on dispatch) keeps today's behaviour exactly: the question is answered in process, no interrupt is emitted, and the run finishes in one exchange. The interrupt path is what a run with no hook now does — where it used to fail with "nobody could be asked", it now asks the client.

Retention, and the deadline clients are shown#

How long a held turn survives, and how many may be held at once, is Retention — set where the transport is constructed, on both of them:

from ag2.ag_ui import AGUIStream, Retention
from ag2.a2ui import A2UIServer
from ag2.a2ui.transports import AgUiTransport

retention = Retention(ttl=900.0, max_held=128)  # the defaults

stream = AGUIStream(agent, retention=retention)
server = A2UIServer(agent, transport=AgUiTransport(retention=retention))

ttl is not an internal detail: it is what a client is shown as the interrupt's expiresAt, so changing it changes what clients are told. The advertised deadline is the earlier of ttl and the deadline implied by any timeout= passed to context.input — the protocol treats an absent deadline as a promise that the interrupt never expires, and clients reject late answers locally on the strength of it, so the figure on the wire is always the one that will in fact apply. Past it the turn is gone and its task cancelled; a max_held overflow cancels the oldest.

That deadline is kept by the server's own clock, not by the next request to arrive: a turn nobody ever answers is released when it expires, on a process that has gone quiet as much as on a busy one.

Memory is proportional to retention. A held turn is a suspended coroutine with its whole conversation history, not a record — ttl × arrival rate, capped by max_held, is roughly what a pausing server holds.

Deployment: a held turn lives in one process#

This is invisible on the wire — the two exchanges look the same either way — and it is the constraint that matters most when you deploy:

• Sticky routing is required. A resume must reach the process holding the turn. Route by threadId; behind a round-robin load balancer, resumes will land on a process that knows nothing; it drops the resume with a warning and starts a fresh run instead of continuing the paused one.
• A restart drops every held turn. After a deploy or a crash, a client answering a question asked by the old process has its resume dropped with a warning and gets an ordinary run — the same as for an expired one. Treat it as "ask again", not as a bug.
• Close the stream on the way down. A2UIServer does it for you (it runs AgUiTransport.aclose() on ASGI shutdown). AGUIStream is not an app, so nothing can hang a shutdown hook on it for you — call await stream.aclose() from your own, or use it as an async context manager, so held turns are cancelled rather than left running into the void. A run still streaming when that happens ends with a cancelled outcome.

Why a resume does not replay the run#

Because the call is suspended inside your Python function. context.input stopped mid-way through book_flight, with its local variables, its loop position and whatever it had already done still live. A message history cannot rebuild that: replaying the run would re-enter the tool from the top and redo whatever it did before it asked.

That is also why held turns cannot be moved to Redis or a database and shared across processes: a suspended coroutine is not serialisable. A durable, cross-process version of this is a different feature — one where the agent's continuation is expressed as resumable state rather than as a paused function — not a setting on this one.

UI clients#

Any AG-UI client can consume ordinary runs from this endpoint. To resume an interrupt, the client must copy its metadata into the resume entry, as shown in Asking the human a question.

For React/Next.js UIs, CopilotKit is the recommended client path in AG2 docs because it provides:

• Streaming chat components
• Tool UI rendering hooks/components
• Shared state patterns for interactive workflows

Start from the CopilotKit UI quickstart.

The same endpoint can also power a bot in Slack and other messaging platforms — see Channels.

AG-UI Dojo#

For protocol-level testing and event inspection, use the AG2 Dojo profile:

• AG2 Dojo - agentic_chat

Next steps#

1. Build the AG-UI endpoint from the minimal example above.
2. Follow the CopilotKit UI quickstart to connect a React/Next.js client.
3. Validate runtime behavior with the AG2 Dojo - agentic_chat.
4. Run the same endpoint as a bot in Slack and other messaging platforms with Channels.