Client Protocol

BASE_URL="https://api.trigger.dev"   # or your local webapp
SECRET_KEY="tr_dev_..."              # secret API key for the env
TASK_ID="ai-chat"                    # your chat.agent task id
CHAT_ID=$(uuidgen | tr '[:upper:]' '[:lower:]')

# 1. Create session + trigger first run with the user's first message.
RESP=$(curl -sS -X POST "$BASE_URL/api/v1/sessions" \
  -H "Authorization: Bearer $SECRET_KEY" \
  -H "Content-Type: application/json" \
  -d @- <<JSON
{
  "type": "chat.agent",
  "externalId": "$CHAT_ID",
  "taskIdentifier": "$TASK_ID",
  "triggerConfig": {
    "basePayload": {
      "chatId": "$CHAT_ID",
      "trigger": "submit-message",
      "message": {
        "id": "u1",
        "role": "user",
        "parts": [{ "type": "text", "text": "Reply with the single word: pong." }]
      },
      "metadata": { "userId": "demo-user" }
    }
  }
}
JSON
)
SESSION_ID=$(echo "$RESP" | jq -r .id)
PAT=$(echo "$RESP" | jq -r .publicAccessToken)
echo "Session: $SESSION_ID  PAT: ${PAT:0:24}..."

# 2. Subscribe to .out and read until the turn-complete control record.
#    The doc's self-contained parser (Step 2) shows full SSE handling;
#    this recipe just greps for `text-delta` data records and the
#    `trigger-control` header on the `turn-complete` control record,
#    then extracts the last seq_num so step 4 can resume from it.
SSE=$(curl -sS --max-time 30 -N \
  -H "Authorization: Bearer $PAT" \
  -H "Accept: text/event-stream" \
  -H "Timeout-Seconds: 20" \
  "$BASE_URL/realtime/v1/sessions/$SESSION_ID/out")
echo "$SSE" | grep -E 'text-delta|trigger-control' | head -2
LAST_SEQ=$(echo "$SSE" | grep -oE '"seq_num":[0-9]+' | tail -1 | grep -oE '[0-9]+')
echo "lastSeq: $LAST_SEQ"

# 3. Send a follow-up via .in/append.
curl -sS -X POST "$BASE_URL/realtime/v1/sessions/$SESSION_ID/in/append" \
  -H "Authorization: Bearer $PAT" \
  -H "Content-Type: application/json" \
  -d @- <<JSON
{
  "kind": "message",
  "payload": {
    "chatId": "$CHAT_ID",
    "trigger": "submit-message",
    "message": {
      "id": "u2",
      "role": "user",
      "parts": [{ "type": "text", "text": "Now reply with: echo." }]
    },
    "metadata": { "userId": "demo-user" }
  }
}
JSON

# 4. Re-subscribe with Last-Event-ID so we resume past turn 1's records
#    and read turn 2 only.
curl -sS --max-time 30 -N \
  -H "Authorization: Bearer $PAT" \
  -H "Accept: text/event-stream" \
  -H "Timeout-Seconds: 20" \
  -H "Last-Event-ID: $LAST_SEQ" \
  "$BASE_URL/realtime/v1/sessions/$SESSION_ID/out" \
  | grep -m 2 -E 'text-delta|trigger-control'

HTTP/1.1 201 Created
content-type: application/json; charset=utf-8
x-trigger-jwt: eyJhbGciOi...
x-trigger-jwt-claims: {"sub":"...","scopes":["read:runs:run_abc123","write:inputStreams:run_abc123"]}

{
  "id": "session_cm4z2plfh000abcd1efgh",
  "externalId": "conversation-123",
  "type": "chat.agent",
  "taskIdentifier": "ai-chat",
  "triggerConfig": { "basePayload": { /* echoed back */ } },
  "currentRunId": "run_abc123",
  "tags": ["chat:conversation-123"],
  "metadata": null,
  "closedAt": null,
  "closedReason": null,
  "expiresAt": null,
  "createdAt": "2026-04-24T09:00:00.000Z",
  "updatedAt": "2026-04-24T09:00:00.000Z",
  "runId": "run_abc123",
  "publicAccessToken": "eyJhbGciOi...",
  "isCached": false
}

GET /realtime/v1/sessions/{sessionId}/out
Authorization: Bearer <publicAccessToken>
Accept: text/event-stream

GET /realtime/v1/sessions/{sessionId}/out
Authorization: Bearer <publicAccessToken>
Accept: text/event-stream
Timeout-Seconds: 30

id: 0,1,106
event: batch
data: {"records":[{"seq_num":0,"timestamp":1712150400000,"body":"{\"data\":{\"type\":\"text-delta\",\"id\":\"msg_1\",\"delta\":\"pong\"},\"id\":\"abc\"}"}],"tail":{"seq_num":10,"timestamp":1712150400500}}

{
  "records": [
    {
      "seq_num": 0,
      "timestamp": 1712150400000,
      "body": "{\"data\":{\"type\":\"text-delta\",\"id\":\"msg_1\",\"delta\":\"pong\"},\"id\":\"abc\"}"
    }
  ],
  "tail": {
    "seq_num": 10,
    "timestamp": 1712150400500
  }
}

// Always advance the resume cursor — even for records you skip.
lastEventId = String(record.seq_num);

// S2 command record: bump cursor, don't dispatch.
if (record.headers?.[0]?.[0] === "") continue;

// Trigger control record: route by `trigger-control` value, don't
// dispatch as a UIMessageChunk.
const controlValue = record.headers?.find(([name]) => name === "trigger-control")?.[1];
if (controlValue === "turn-complete") {
  const token = record.headers.find(([name]) => name === "public-access-token")?.[1];
  // ...fire your turn-complete handler with the optional refreshed token...
  continue;
}
if (controlValue === "upgrade-required") {
  // ...your upgrade flow, if any. The server has already swapped the run
  // by the time this arrives — subsequent chunks are from the new run...
  continue;
}

// Otherwise: data record. Parse the body, dispatch the UIMessageChunk.
const { data: chunk } = JSON.parse(record.body);

import { SSEStreamSubscription, controlSubtype } from "@trigger.dev/core/v3";

const subscription = new SSEStreamSubscription(
  `${baseUrl}/realtime/v1/sessions/${sessionId}/out`,
  {
    headers: { Authorization: `Bearer ${publicAccessToken}` },
    timeoutInSeconds: 120,
    lastEventId,
  }
);

const stream = await subscription.subscribe();
const reader = stream.getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  // value is { id, chunk, timestamp, headers }. S2 command records are
  // filtered out of this stream entirely (cursor still advances). Trigger
  // control records pass through with `chunk === undefined` and a
  // `trigger-control` header.
  const control = controlSubtype(value.headers);
  if (control === "turn-complete") break;
  if (control === "upgrade-required") continue;

  const chunk = value.chunk as { type?: string; delta?: string } | undefined;
  if (chunk?.type === "text-delta") process.stdout.write(chunk.delta ?? "");
}

async function* readSessionOut(
  url: string,
  publicAccessToken: string,
  opts: { lastEventId?: string; timeoutSeconds?: number } = {}
) {
  const headers: Record<string, string> = {
    Authorization: `Bearer ${publicAccessToken}`,
    Accept: "text/event-stream",
  };
  if (opts.lastEventId) headers["Last-Event-ID"] = opts.lastEventId;
  if (opts.timeoutSeconds) headers["Timeout-Seconds"] = String(opts.timeoutSeconds);

  const res = await fetch(url, { headers });
  if (!res.ok || !res.body) throw new Error(`SSE failed: ${res.status}`);

  const decoder = new TextDecoder();
  const reader = res.body.getReader();
  let buf = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) return;
    buf += decoder.decode(value, { stream: true });

    // SSE events are separated by blank lines (CRLF or LF).
    const events = buf.split(/\r?\n\r?\n/);
    buf = events.pop() ?? ""; // last chunk is incomplete

    for (const raw of events) {
      let eventType = "message"; // SSE default
      const dataLines: string[] = [];
      for (const line of raw.split(/\r?\n/)) {
        if (line.startsWith("event:")) eventType = line.slice(6).trim();
        else if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart());
        // We deliberately ignore `id:` — use record.seq_num for resume cursors.
      }
      const data = dataLines.join("\n");
      if (!data) continue;

      if (eventType === "ping") continue;
      if (data === "[DONE]") return;

      if (eventType === "batch") {
        const batch = JSON.parse(data) as {
          records: Array<{
            seq_num: number;
            timestamp: number;
            body: string;
            headers?: Array<[string, string]>;
          }>;
        };
        for (const record of batch.records) {
          const firstHeaderName = record.headers?.[0]?.[0];

          // S2 command record (trim/fence) — bump cursor, skip dispatch.
          if (firstHeaderName === "") {
            yield { seqNum: record.seq_num, timestamp: record.timestamp, kind: "command" };
            continue;
          }

          // Trigger control record (turn-complete, upgrade-required) —
          // semantics live in headers, body is empty. Route by header.
          const controlValue = record.headers?.find(([n]) => n === "trigger-control")?.[1];
          if (controlValue) {
            const token = record.headers?.find(([n]) => n === "public-access-token")?.[1];
            yield {
              seqNum: record.seq_num,
              timestamp: record.timestamp,
              kind: "control",
              subtype: controlValue,
              publicAccessToken: token,
            };
            continue;
          }

          // Data record — UIMessageChunk wrapped in `{ data, id }`.
          const inner = JSON.parse(record.body) as { data: unknown; id: string };
          yield {
            seqNum: record.seq_num, // use this for Last-Event-ID on resume
            timestamp: record.timestamp,
            kind: "data",
            chunk: inner.data, // the actual UIMessageChunk
          };
        }
      }
    }
  }
}

let lastSeq: string | undefined;
for await (const ev of readSessionOut(sseUrl, publicAccessToken)) {
  lastSeq = String(ev.seqNum);                       // always advance the cursor

  if (ev.kind === "command") continue;               // S2 trim/fence — skip
  if (ev.kind === "control") {
    if (ev.subtype === "turn-complete") break;       // turn done
    if (ev.subtype === "upgrade-required") continue; // run swap handled server-side
    continue;
  }

  // ev.kind === "data" — the UIMessageChunk
  const chunk = ev.chunk as { type: string; delta?: string };
  if (chunk.type === "text-delta") process.stdout.write(chunk.delta ?? "");
}
// On reconnect, pass `lastEventId: lastSeq` to resume from the next record.

headers:
  ["trigger-control", "turn-complete"]
  ["public-access-token", "eyJ..."]   // optional, refreshed JWT
  ["session-in-event-id", "42"]       // optional, agent-internal resume cursor
body: ""

headers:
  ["trigger-control", "upgrade-required"]
body: ""

GET /realtime/v1/sessions/{sessionId}/out
Authorization: Bearer <publicAccessToken>
Accept: text/event-stream
Last-Event-ID: 42

// Reconnect path (page reload)
const response = await fetch(sseUrl, {
  headers: {
    Authorization: `Bearer ${publicAccessToken}`,
    "X-Peek-Settled": "1",
    "Last-Event-ID": lastEventId,
  },
});
const settled = response.headers.get("X-Session-Settled") === "true";
// ...subscribe as normal; if settled and nothing arrives, you're done.

// Active send path — no X-Peek-Settled, keep long-poll semantics
const liveResponse = await fetch(sseUrl, {
  headers: {
    Authorization: `Bearer ${publicAccessToken}`,
    "Last-Event-ID": lastEventId,
  },
});

POST /realtime/v1/sessions/{sessionId}/in/append
Authorization: Bearer <publicAccessToken>
Content-Type: application/json

type ChatInputChunk =
  | { kind: "message"; payload: ChatTaskWirePayload }
  | { kind: "stop"; message?: string };

type ChatTaskWirePayload<TMessage extends UIMessage = UIMessage, TMetadata = unknown> = {
  /**
   * The new message for this turn — at most ONE per record.
   *  - "submit-message": the new user message, OR a tool-approval-responded
   *    assistant message (with `state: "approval-responded"` tool parts).
   *  - "regenerate-message": omitted (the server trims its own tail).
   *  - "preload" / "close" / "action": omitted.
   *  - "handover-prepare": omitted (use `headStartMessages` instead — see below).
   */
  message?: TMessage;

  /**
   * Escape hatch for chat.headStart. Ships full UIMessage history on the
   * very first turn — before any snapshot exists. Used ONLY by
   * trigger: "handover-prepare" against the customer's own HTTP route
   * handler. The server ignores this field on any other trigger.
   */
  headStartMessages?: TMessage[];

  chatId: string;
  trigger:
    | "submit-message"
    | "regenerate-message"
    | "preload"
    | "close"
    | "action"
    | "handover-prepare";
  messageId?: string;
  /**
   * Wire envelope for the agent's typed `clientData` (declared via
   * `chat.withClientData({ schema })`). Whatever you put here is parsed
   * against that schema at the agent boundary. If the agent declares
   * `clientData: { userId: string }`, then `metadata.userId` is required.
   */
  metadata?: TMetadata;
  action?: unknown;
  /**
   * Informational — the server sets this automatically on continuation
   * runs (when the prior run is dead). Clients don't need to send it.
   * Read by the agent's boot gate to skip `onChatStart` and trigger
   * snapshot read + replay.
   */
  continuation?: boolean;
  /**
   * Informational — paired with `continuation: true`, set by the server
   * from the prior run's friendly ID. Surfaced to the agent in
   * `ctx.previousRunId`. Clients don't need to send it.
   */
  previousRunId?: string;
  idleTimeoutInSeconds?: number;
  sessionId?: string;
};

POST /realtime/v1/sessions/{sessionId}/in/append
Authorization: Bearer <publicAccessToken>
Content-Type: application/json

{
  "kind": "message",
  "payload": {
    "message": {
      "id": "msg-2",
      "role": "user",
      "parts": [{ "type": "text", "text": "Tell me more" }]
    },
    "chatId": "conversation-123",
    "trigger": "submit-message",
    "metadata": { "userId": "user-456" }
  }
}

{ "kind": "stop" }

{ "kind": "stop", "message": "user cancelled" }

{
  "kind": "message",
  "payload": {
    "chatId": "conversation-123",
    "trigger": "action",
    "action": { "type": "undo" },
    "metadata": { "userId": "user-456" }
  }
}

{
  "kind": "message",
  "payload": {
    "chatId": "conversation-123",
    "trigger": "regenerate-message",
    "metadata": { "userId": "user-456" }
  }
}

{
  "kind": "message",
  "payload": {
    "message": {
      "id": "asst-msg-1",
      "role": "assistant",
      "parts": [
        { "type": "text", "text": "I'll send that email for you." },
        {
          "type": "tool-sendEmail",
          "toolCallId": "call-1",
          "state": "approval-responded",
          "input": { "to": "user@example.com", "subject": "Hello" },
          "approval": { "id": "approval-1", "approved": true }
        }
      ]
    },
    "chatId": "conversation-123",
    "trigger": "submit-message",
    "metadata": { "userId": "user-456" }
  }
}

{
  "kind": "message",
  "payload": {
    "headStartMessages": [
      { "id": "u1", "role": "user", "parts": [/* ... */] },
      { "id": "a1", "role": "assistant", "parts": [/* ... */] }
    ],
    "chatId": "conversation-123",
    "trigger": "handover-prepare",
    "metadata": { "userId": "user-456" }
  }
}

{
  "kind": "message",
  "payload": {
    "message": {
      "id": "u-42",
      "role": "user",
      "parts": [{ "type": "text", "text": "Where were we?" }]
    },
    "chatId": "conversation-123",
    "trigger": "submit-message",
    "metadata": { "userId": "user-456" }
  }
}

POST /api/v1/sessions/{sessionId}/close
Authorization: Bearer <secret-key-or-jwt>
Content-Type: application/json

{ "reason": "user-ended" }