lukasa1993/tanstack-skills-extracted/tanstack-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

import { chat, toServerSentEventsResponse } from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'
export async function POST(request: Request) {
  const { messages, threadId } = await request.json()
  const stream = chat({
    adapter: openaiText('gpt-6.1-sol'),
    messages,
    threadId,
  })
  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

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-6.1-sol'),
    messages: params.messages,
    threadId: params.threadId,
    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. The last event is RUN_FINISHED (or RUN_ERROR). There is no [DONE] line.

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-6.1-sol'),
  messages,
  threadId: 'thread-1',
  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):

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).

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-6.1-sol'),
  messages,
  threadId: 'thread-1',
  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:

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

Typical event sequence with tool calls:

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). For event type names, use the EventType enum, or ` ${EventType} ` for the string union.

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:

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-6.1-sol'),
  messages,
  threadId: 'thread-1',
})
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.