Ai Core/Ag Ui Protocol

AG-UI Protocol

This skill builds on ai-core. Read it first for critical rules.

Setup — Server Endpoint Producing AG-UI Events via SSE

typescriptCopy code
import { chat, toServerSentEventsResponse } from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'

export async function POST(request: Request) {
  const { messages } = await request.json()
  const stream = chat({
    adapter: openaiText('gpt-5.6'),
    messages,
  })
  return toServerSentEventsResponse(stream)
}

chat() returns an AsyncIterable<StreamChunk>. Each StreamChunk is a typed AG-UI event (discriminated union on type). The toServerSentEventsResponse() helper encodes that iterable into an SSE-formatted Response with correct headers.

Setup — Receiving AG-UI RunAgentInput on the Server

typescriptCopy code
import {
  chat,
  chatParamsFromRequestBody,
  mergeAgentTools,
  toServerSentEventsResponse,
} from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'
import { serverTools } from './tools'

export async function POST(req: Request) {
  let params
  try {
    params = await chatParamsFromRequestBody(await req.json())
  } catch (error) {
    return new Response(
      error instanceof Error ? error.message : 'Bad request',
      { status: 400 },
    )
  }

  const stream = chat({
    adapter: openaiText('gpt-5.6'),
    messages: params.messages,
    tools: mergeAgentTools(serverTools, params.tools),
  })

  return toServerSentEventsResponse(stream)
}

chatParamsFromRequestBody validates the body against RunAgentInputSchema from @ag-ui/core. mergeAgentTools merges the server's tool registry with client-declared tools (server wins on collision; client-only tools become no-execute stubs that flow through the runtime's ClientToolRequest path).

params.messages is a mixed array of TanStack UIMessage anchors (with parts) and AG-UI fan-out duplicates ({role:'tool',...}, {role:'reasoning',...}). The existing convertMessagesToModelMessages (called inside chat()) handles dedup automatically.

Wire shape (POST body): AG-UI RunAgentInput — {threadId, runId, parentRunId?, state, messages, tools, context, forwardedProps}. The messages array carries TanStack UIMessage anchors with their canonical parts plus AG-UI mirror fields (content, toolCalls) inline; tool results and thinking parts are additionally emitted as fan-out {role:'tool',...} and {role:'reasoning',...} entries.

forwardedProps security: Don't spread it directly into chat() — clients could override adapter, model, tools, etc. Always allowlist specific fields.

Core Patterns

1. SSE Format — toServerSentEventsStream / toServerSentEventsResponse

Wire format: Each event is data: <JSON>\n\n. Stream ends with data: [DONE]\n\n.

typescriptCopy code
import {
  chat,
  toServerSentEventsStream,
  toServerSentEventsResponse,
} from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'

const messages = [{ role: 'user' as const, content: 'Hello' }]

// Option A: Get a ReadableStream (manual Response construction)
const abortController = new AbortController()
const stream = chat({
  adapter: openaiText('gpt-5.6'),
  messages,
  abortController,
})
const sseStream = toServerSentEventsStream(stream, abortController)

const response = new Response(sseStream, {
  headers: {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
  },
})

// Option B: Use the helper (sets headers automatically)
const response2 = toServerSentEventsResponse(stream, { abortController })
// Default headers: Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive

Default response headers set by toServerSentEventsResponse():

Custom headers merge on top (user headers override defaults):

typescriptCopy code
toServerSentEventsResponse(stream, {
  headers: {
    'X-Accel-Buffering': 'no', // Disable nginx buffering
    'Cache-Control': 'no-store', // Override default
  },
  abortController,
})

Error handling: If the stream throws, a RUN_ERROR event is emitted automatically before the stream closes. If the abortController is already aborted, the error event is suppressed and the stream closes silently.

2. HTTP Stream (NDJSON) — toHttpStream / toHttpResponse

Wire format: Each event is <JSON>\n (newline-delimited JSON, no SSE prefix, no [DONE] marker).

typescriptCopy code
import { chat, toHttpStream, toHttpResponse } from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'

const messages = [{ role: 'user' as const, content: 'Hello' }]

// Option A: Get a ReadableStream
const abortController = new AbortController()
const stream = chat({
  adapter: openaiText('gpt-5.6'),
  messages,
  abortController,
})
const ndjsonStream = toHttpStream(stream, abortController)

const response = new Response(ndjsonStream, {
  headers: {
    'Content-Type': 'application/x-ndjson',
  },
})

// Option B: Use the helper (does NOT set headers automatically)
const response2 = toHttpResponse(stream, { abortController })
// Note: toHttpResponse does NOT set Content-Type automatically.
// You should pass headers explicitly:
const response3 = toHttpResponse(stream, {
  headers: { 'Content-Type': 'application/x-ndjson' },
  abortController,
})

Client-side pairing: SSE endpoints are consumed by fetchServerSentEvents(). HTTP stream endpoints are consumed by fetchHttpStream(). Both are connection adapters from @tanstack/ai-react (or the framework-specific package).

3. AG-UI Event Types Reference

All events extend BaseAGUIEvent which carries type, timestamp, optional model, and optional rawEvent.

Typical event sequence for a text-only response:

Copy code
RUN_STARTED -> TEXT_MESSAGE_START -> TEXT_MESSAGE_CONTENT (repeated) -> TEXT_MESSAGE_END -> RUN_FINISHED

Typical event sequence with tool calls:

Copy code
RUN_STARTED -> TEXT_MESSAGE_START -> TEXT_MESSAGE_CONTENT* -> TEXT_MESSAGE_END
            -> TOOL_CALL_START -> TOOL_CALL_ARGS* -> TOOL_CALL_END
            -> RUN_FINISHED (finishReason: 'tool_calls')

Type aliases: StreamChunk is an alias for AGUIEvent (the discriminated union of all event interfaces). StreamChunkType is an alias for AGUIEventType (the string union of all event type literals).

4. Typed CUSTOM Events — ChatStream and KnownCustomEvent

The CUSTOM row above describes the raw StreamChunk union, where the single generic CustomEvent member types value as any -- once merged into a union, that any poisons every other member too, so narrowing on name still leaves value: any. chat() doesn't return raw StreamChunk; by default (no outputSchema, stream not explicitly false) it returns ChatStream, which swaps that generic member for KnownCustomEvent -- a discriminated union of every CUSTOM event TanStack AI itself emits, each with a literal name and a concrete value. Narrow with a plain if -- no helper, no cast:

typescriptCopy code
import { chat } from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'

const messages = [{ role: 'user' as const, content: 'Hello' }]

const stream = chat({
  adapter: openaiText('gpt-5.6'),
  messages,
})

for await (const chunk of stream) {
  if (chunk.type === 'CUSTOM' && chunk.name === 'sandbox.file.diff') {
    console.log(chunk.value.path, chunk.value.diff) // typed, no helper, no cast
  } else if (
    chunk.type === 'CUSTOM' &&
    chunk.name === 'structured-output.complete'
  ) {
    console.log(chunk.value.object) // typed, no helper, no cast
  }
}

Caveat -- .endsWith() (or any non-literal check) does not narrow. SessionIdEvent['name'] is the template-literal type `${string}.session-id`. TypeScript's control-flow narrowing only understands exact comparisons (===) and in/type-predicate checks against a discriminant -- a runtime chunk.name.endsWith('.session-id') check doesn't inform the type system, so chunk.value stays the union of every KnownCustomEvent's value, not { sessionId: string }. Compare against the exact literal you expect, or write a user-defined type predicate ((c): c is SessionIdEvent => c.name.endsWith('.session-id')) and call that in the if instead.

User-emitted emitCustomEvent names are typed out of ChatStream. Tools that call context.emitCustomEvent('my-app:progress', ...) still stream a CUSTOM chunk at runtime, but 'my-app:progress' isn't one of KnownCustomEvent's literal names, so it's intentionally absent from ChatStream's type -- including a generic fallback member would reintroduce the value: any poison for every other event on the stream. To read your own event with a type, annotate the stream as the wider StreamChunk instead of ChatStream for that branch; its generic CUSTOM member already types value as any, so no cast is needed there either.

Source: docs/protocol/custom-events.md

Common Mistakes

MEDIUM: Proxy buffering breaks SSE streaming

Reverse proxies (nginx, Cloudflare, AWS ALB) buffer SSE responses by default, causing events to arrive in batches instead of streaming token-by-token.

Fix: Set proxy-bypass headers on the response.

typescriptCopy code
toServerSentEventsResponse(stream, {
  headers: {
    'X-Accel-Buffering': 'no', // nginx
    'X-Content-Type-Options': 'nosniff', // Some CDNs
  },
  abortController,
})

For Cloudflare Workers, SSE streams automatically. For Cloudflare proxied origins, ensure "Response Buffering" is disabled in the dashboard.

Source: docs/protocol/sse-protocol.md

MEDIUM: Assuming all AG-UI events arrive in every response

Not all event types appear in every stream:

• STEP_STARTED / STEP_FINISHED only appear with thinking-enabled models (e.g., o3, claude-sonnet-4-5 with extended thinking). Standard models skip these entirely.
• TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END only appear when the model invokes tools. A text-only response has none.
• STATE_SNAPSHOT / STATE_DELTA only appear when server code explicitly emits them for stateful agent workflows.
• MESSAGES_SNAPSHOT only appears when the server explicitly sends a full transcript snapshot.
• CUSTOM events are application-defined and never emitted by default.

Code that expects a fixed sequence (e.g., always waiting for STEP_FINISHED before processing text) will hang or break on models that don't emit those events.

Source: docs/protocol/chunk-definitions.md

Tension

RESOLVED: TanStack AI is fully AG-UI compliant on both axes (server→client events AND client→server RunAgentInput). The wire format carries TanStack UIMessage anchors with their parts intact alongside AG-UI fan-out messages, so strict AG-UI servers see role-based messages while TanStack-aware servers read parts directly without transformation. See docs/migration/ag-ui-compliance.md for details.

Cross-References

• See also: ai-core/custom-backend-integration/SKILL.md -- Custom backends must implement SSE or HTTP stream format to work with TanStack AI client connection adapters.
• See also: ai-core/middleware/SKILL.md -- sandbox.file.diff's { path, diff } value (one of KnownCustomEvent's members) is populated from the same lazy before()/after()/diff() accessors documented there for onFile* middleware hooks.
• Full CUSTOM event taxonomy: docs/protocol/custom-events.md.