Ag UI Protocol

AG-UI Protocol (pinned)

Versions: ag-ui-protocol==1.0.0, ag-ui-langgraph==0.0.45 (Python, FastAPI side) · @ag-ui/client@1.0.0 (browser). Every identifier below was read from those installed packages. If the pins change, re-verify against ag_ui/_generated/models.py, ag_ui_langgraph/types.py, and @ag-ui/client/dist/index.d.ts before trusting this file.

Architecture: browser HttpAgent → POST → FastAPI route from add_langgraph_fastapi_endpoint → LangGraphAgent.run() → graph.astream_events(version="v2") → AG-UI frames as SSE. Two hops, no intermediary.

Field-level shapes for every event, the full subscriber callback list, and the LangGraph→AG-UI mapping table are in reference.md.

Event catalog (31 types, EventType enum)

Not in 1.0: THINKING_* (removed; use REASONING_*). Wire format is camelCase (threadId, messageId, toolCallId); Python models accept snake_case and serialize by alias.

Producer side (Python)

Emit a structured payload from a tool — the only API the adapter forwards:

from langchain_core.callbacks import adispatch_custom_event
from langchain_core.tools import tool

@tool
async def search_hotels(city: str) -> str:
    rows = await _search(city)
    await adispatch_custom_event("panel_state", {"rows": rows, "total": len(rows)})
    return f"{len(rows)} hotels found"   # what the model sees

Arrives in the browser as CUSTOM {name: "panel_state", value: {...}}, payload byte-identical. Works from any tool or node running inside the graph (config propagates via contextvars on Python ≥3.11; pass config= explicitly if you call from outside a Runnable).

Reserved custom-event names (ag_ui_langgraph.types.CustomEventNames) — the adapter special-cases these and still forwards them as CUSTOM:

RESERVED = {"manually_emit_message", "manually_emit_tool_call", "manually_emit_state", "exit"}

manually_emit_message → synthesizes a full TEXT_MESSAGE triple; manually_emit_tool_call → TOOL_CALL triple; manually_emit_state → STATE_SNAPSHOT and overrides run state; exit → forwarded only, advisory, does not end the stream. Never reuse these names for your own payloads.

Serve the graph:

from ag_ui_langgraph import LangGraphAgent, add_langgraph_fastapi_endpoint

agent = LangGraphAgent(name="ferdinand", graph=compiled_graph, emit_raw_events=False)
add_langgraph_fastapi_endpoint(app, agent, path="/api/agent/ag-ui", dependencies=[Depends(auth)])

• emit_raw_events=False — default True re-emits every LangGraph event as RAW and attaches rawEvent to nearly every frame; on graphs with real state this dominates payload size.
• The route clones the agent per request (agent.clone()); LangGraphAgent holds per-run state and is not safe to share.
• **kwargs on the endpoint go to app.post (dependencies, tags, ...) — that's where auth goes. The client's headers reach FastAPI unchanged.
• A checkpointer is mandatory. Every run calls graph.aget_state(config) with configurable.thread_id = input.threadId; no checkpointer → failure on the first request. See langgraph-patterns for which one.
• Graph state must have a messages key with the add_messages reducer. The adapter also writes tools (client-declared frontend tools) and ag-ui keys; declare them in the schema if you read them.

Consumer side (TypeScript)

import { HttpAgent, randomUUID } from "@ag-ui/client";

const agent = new HttpAgent({ url: "/api/agent/ag-ui", headers: { Authorization: `Bearer ${token}` }, threadId });

agent.subscribe({
  onCustomEvent: ({ event }) => { if (event.name === "panel_state") setPanel(event.value); },
  onTextMessageContentEvent: ({ event, textMessageBuffer }) => setDraft(event.messageId, textMessageBuffer),
  onTextMessageEndEvent: ({ event }) => commitMessage(event.messageId),
  onToolCallStartEvent: ({ event }) => setToolStatus(event.toolCallId, event.toolCallName, "running"),
  onToolCallResultEvent: ({ event }) => setToolStatus(event.toolCallId, undefined, "done"),
  onRunFinishedEvent: ({ outcome, interrupts }) => { if (outcome === "interrupt") showInterrupts(interrupts); },
  onRunErrorEvent: ({ event }) => toast(event.message),
});

agent.addMessage({ id: randomUUID(), role: "user", content: text });
await agent.runAgent();          // sends agent.messages + agent.state
agent.abortRun();                // cancel in-flight

• runAgent(params?) takes only runId | tools | context | forwardedProps | resume — not messages. Add messages first with addMessage/addMessages/setMessages.
• The client re-sends the entire message history every run; the adapter dedupes against the checkpoint by message id. Ids must be stable — generate once, never re-mint on re-render.
• textMessageBuffer / toolCallBuffer / partialToolCallArgs are accumulated for you; don't concatenate deltas yourself.
• agent.messages / agent.state are updated automatically as frames apply; onMessagesChanged / onStateChanged fire after.
• Client sends Accept: text/event-stream; the FastAPI helper reads it and picks the encoder.

Traps

Related: langgraph-patterns (checkpointers, interrupts, graph construction) · agent-tool-design (what tools return vs. emit).