# Ag UI Protocol

> Use when writing or debugging code that streams a LangGraph agent to a browser over AG-UI — emitting structured payloads from tools, mapping LangGraph events to AG-UI frames, subscribing in the browser, or diagnosing "the event never arrived" — pinned to ag-ui-protocol 1.0.0 / ag-ui-langgraph 0.0.45 / @ag-ui/client 1.0.0. Model recall of this protocol is stale (pre-1.0 THINKING_* events, copilotkit_* names); read this instead.

- Skill: `andreasbloomquist/ag-ui-protocol` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add andreasbloomquist/ag-ui-protocol`
- Raw SKILL.md: https://api.skillmd.com/api/skills/andreasbloomquist/ag-ui-protocol/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: andreasbloomquist (https://skillmd.com/u/andreasbloomquist)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/andreasbloomquist/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)

| Group | Types |
|---|---|
| Run lifecycle | `RUN_STARTED` `RUN_FINISHED` `RUN_ERROR` `STEP_STARTED` `STEP_FINISHED` |
| Text | `TEXT_MESSAGE_START` `TEXT_MESSAGE_CONTENT` `TEXT_MESSAGE_END` `TEXT_MESSAGE_CHUNK` |
| Tool calls | `TOOL_CALL_START` `TOOL_CALL_ARGS` `TOOL_CALL_END` `TOOL_CALL_CHUNK` `TOOL_CALL_RESULT` |
| State | `STATE_SNAPSHOT` `STATE_DELTA` `MESSAGES_SNAPSHOT` |
| Activity | `ACTIVITY_SNAPSHOT` `ACTIVITY_DELTA` |
| Reasoning | `REASONING_START` `REASONING_MESSAGE_START` `REASONING_MESSAGE_CONTENT` `REASONING_MESSAGE_END` `REASONING_MESSAGE_CHUNK` `REASONING_END` `REASONING_ENCRYPTED_VALUE` |
| Subagent | `SUBAGENT_STARTED` `SUBAGENT_FINISHED` `SUBAGENT_ERROR` |
| Escape hatches | `RAW` `CUSTOM` |

**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:

```python
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*:

```python
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:**

```python
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)

```ts
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

| Symptom | Cause | Fix |
|---|---|---|
| Payload emitted, never reaches browser | Used `get_stream_writer()` / `stream_mode="custom"` — the adapter consumes `astream_events` and only forwards `on_custom_event` | `adispatch_custom_event` |
| Payload arrives but also mutates chat / state | Used a reserved name | Rename; check `RESERVED` |
| Put UI data in `STATE_DELTA`/`STATE_SNAPSHOT` | Those are graph-state channels; every node exit re-snapshots, and the client echoes state back | Use `CUSTOM`; keep UI payloads out of graph state |
| First request 500s | No checkpointer, or `messages` key missing from state schema | Add both |
| Frames 10–100× larger than expected | `emit_raw_events` left at default `True` | Set `False` |
| Duplicate user messages in history | Client re-minted message ids | Stable ids |
| Tool exception → no `TOOL_CALL_RESULT` | `on_tool_error` branch resets tracking, emits nothing | Return errors as tool output (see `agent-tool-design`) so it arrives via `on_tool_end` |
| Model answered from memory: `THINKING_*`, `copilotkit_*` names | Pre-1.0 knowledge | This file |

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