Stream events from a turn

Goal: render a live UI from a turn's SSE event stream — phases, agent timings, validator verdicts, the answer streaming in token by token. The "watch Amy think" feel from the CLI, in any client.

Goal: render a live UI from a turn's SSE event stream — phases, agent timings, validator verdicts, the answer streaming in token by token. The "watch Amy think" feel from the CLI, in any client.

This recipe goes deeper than Ask a question. It covers the full event catalog, the right code for browsers / Node / RN, the reconnect protocol, graceful cancellation, and a worked CLI-style trace renderer.

The SDK wraps every SSE frame as { type, id, data }. type mirrors the SSE event: field, id mirrors id: (a monotonic per-turn integer as a string), and data is the parsed JSON body. Every typed field lives under event.data — there are no top-level shortcuts.

STEP 1 — Know which event drives which UI element

The cloud workflow synthesizes the three lifecycle frames (turn.started, turn.completed, turn.failed); everything else is emitted by the multi-agent orchestrator running inside the turn.

Event	Fields you read	Drives in the UI
`turn.started`	`data.turn_id`, `data.at`	Open the chat bubble. Start the elapsed-time counter.
`phase`	`data.agent`, `data.phase`, `data.detail?`	A progress marker. ~5–30 of these per turn — render as a small stack or just the most recent.
`routing`	`data.decision.main_agent`, `data.decision.supporting_agents`	"Data Science + Domain Expert are working on this."
`rephrase`	`data.main`, `data.supporting`	Optional — show the per-agent sub-question if you have a debug pane.
`agent_start`	`data.agent`, `data.question`	Header label: "Data Science is running…". Spinner on.
`agent_end`	`data.agent`, `data.cost_usd`, `data.duration_ms`, `data.preview?`	Header label: "Data Science done in 64s". Optional collapse.
`validation_start`	`data.finding_id`, `data.claim`	Pill: "Checking: average HRV over 30 days…".
`validation_end`	`data.verdict`, `data.gates_passed`, `data.gates_total`, `data.reason`	Pill: VALIDATED · CONDITIONAL · REJECTED.
`synthesis_delta`	`data.text`	The token stream of the final answer. Append `data.text` to the visible bubble.
`fact_check`	`data.issues[]`	Soft warning bar if any numeric in the rendered reply didn't match the fact sheet.
`memory`	`data.entries[]`	"Amy remembered 2 things from this turn." (optional confirmation toast)
`cost` / `cost_warning`	`data.usd` / `data.cumulative_usd`	Live cost meter (debug).
`turn.completed`	`data.turn_id`, `data.result`	Close the bubble. Spinner off. Render `data.result.answer` as the canonical final text.
`turn.failed`	`data.turn_id`, `data.error`	Red banner with `data.error.message`. Stop the spinner.

Rule of thumb: phase and agent_* drive the "who's working" line; validation_* drives the trust indicators; synthesis_delta drives the typewriter; turn.* drives the lifecycle.

Full payload shapes: SDK: TypeScript / Event types.

The SDK works in Node, Bun, browsers, and React Native. It uses fetch + ReadableStream under the hood — no EventSource polyfill required, even on React Native (provided your runtime ships a global fetch, which Hermes does in Expo SDK 53+).

Runtime	What to use	Auth headers
Node / Bun / Deno	`amy.turns.stream(id)` async iterator	Handled by the SDK
Browser	`amy.turns.stream(id)` async iterator	Handled by the SDK (cannot use native `EventSource` — it can't set `Authorization`)
React Native (Expo)	`amy.turns.stream(id)` async iterator	Handled by the SDK; works with `apiKey` or `apiKeyProvider`

Node / Bun

import { Amy } from "@amy/sdk";

const amy = new Amy({
  apiKey: process.env.AMY_API_KEY!,
  baseUrl: process.env.AMY_BASE_URL,
});

const turn = await amy.turns.create({
  messages: [{ role: "user", content: "How's my recovery?" }],
});

for await (const event of amy.turns.stream(turn.id)) {
  switch (event.type) {
    case "turn.started":
      console.log("→ turn started");
      break;
    case "agent_start":
      console.log(`→ ${event.data.agent} running…`);
      break;
    case "synthesis_delta":
      process.stdout.write(event.data.text);
      break;
    case "agent_end":
      console.log(`\n  ${event.data.agent} done in ${event.data.duration_ms}ms`);
      break;
    case "validation_end":
      console.log(`  ${event.data.verdict} (${event.data.gates_passed}/${event.data.gates_total})`);
      break;
    case "turn.completed":
      console.log("\n✓", event.data.result.answer);
      break;
    case "turn.failed":
      console.error("✗", event.data.error.message);
      break;
  }
}

Browser (via the SDK)

The browser's native EventSource can't set Authorization headers, so the SDK's iterator (which uses fetch) is the only sane path.

import { Amy } from "@amy/sdk";

// Mobile / web — pass Clerk's getToken() through:
import { useAuth } from "@clerk/react";
const { getToken } = useAuth();
const amy = new Amy({ apiKeyProvider: () => getToken() });

const turn = await amy.turns.create({
  messages: [{ role: "user", content: "How's my recovery?" }],
});

const $bubble = document.querySelector("#answer")!;

for await (const event of amy.turns.stream(turn.id)) {
  if (event.type === "synthesis_delta") {
    $bubble.textContent += event.data.text;
  }
  if (event.type === "turn.completed") {
    // Canonical final text — swap in to recover from any dropped tokens.
    $bubble.textContent = event.data.result.answer;
  }
}

React Native (Expo)

Identical code — same SDK iterator. No EventSource polyfill needed.

import { useAuth } from "@clerk/expo";
import { Amy } from "@amy/sdk";

const { getToken } = useAuth();
const amy = new Amy({ apiKeyProvider: () => getToken() });

const turn = await amy.turns.create({
  messages: [{ role: "user", content }],
});

for await (const event of amy.turns.stream(turn.id)) {
  if (event.type === "synthesis_delta") {
    setAnswer((prev) => prev + event.data.text);
  }
  if (event.type === "turn.completed") {
    setAnswer(event.data.result.answer);
  }
}

Token resolution timing. With apiKeyProvider, the SDK calls getToken() once when the stream opens — the resulting connection keeps that token until it closes. Clerk session JWTs default to a 60-second TTL, but getToken() auto-refreshes, so this only matters if the same connection runs across an expiry. For multi-minute turns, reconnect with Last-Event-Id on any disconnect (see Step 3) and a fresh token will be issued.

STEP 3 — Handle reconnects with `Last-Event-Id`

Streams break. Networks idle out, phones go to sleep, server processes restart. The SSE protocol has a built-in resume mechanism: the server sends an id: line with every event, and the client passes the last one it saw back via the Last-Event-Id header on reconnect.

The server replays from that ID forward. Replays are available for 1 hour after turn completion.

Manually with the SDK

The SDK does not auto-reconnect in v1 — it surfaces a single stream_closed_unexpectedly error on disconnect and you decide what to do. Track the last event id you saw and pass it on retry:

let lastId: string | number | null = null;
let attempt = 0;

while (attempt < 5) {
  try {
    for await (const event of amy.turns.stream(turn.id, {
      lastEventId: lastId,
    })) {
      lastId = event.id ?? lastId;
      handleEvent(event);
      if (event.type === "turn.completed" || event.type === "turn.failed") return;
    }
    return; // stream ended cleanly
  } catch (err) {
    if (err instanceof AmyApiError && err.code === "stream_closed_unexpectedly") {
      attempt++;
      const backoff = Math.min(30_000, 500 * 2 ** attempt) * (0.5 + Math.random());
      await new Promise((r) => setTimeout(r, backoff));
      continue;
    }
    throw err;
  }
}

Manually with curl

# Initial subscribe
curl -N -H "Authorization: Bearer $AMY_TOKEN" \
  "$AMY_BASE_URL/v1/turns/$TURN_ID/events"

# …network blip, last id seen was 42 …

# Resume from event 43
curl -N -H "Authorization: Bearer $AMY_TOKEN" \
  -H "Last-Event-Id: 42" \
  "$AMY_BASE_URL/v1/turns/$TURN_ID/events"

Server replays from event 43 forward.

STEP 4 — Cancel a stream gracefully

You'll need this when the user closes the chat, navigates away, or hits a stop button. The SDK's iterator accepts an AbortSignal:

const controller = new AbortController();

// Wire to your UI's "stop" button
stopButton.onclick = () => controller.abort();

try {
  for await (const event of amy.turns.stream(turn.id, { signal: controller.signal })) {
    handleEvent(event);
  }
} catch (err) {
  if (err instanceof AmyApiError && err.code === "aborted") {
    console.log("user cancelled the stream");
  } else {
    throw err;
  }
}

Aborting only closes the SSE connection on the client. The turn keeps running on the server, its result is still saved in D1 and retrievable later via GET /v1/turns/:id. There is no POST /v1/turns/:id/cancel endpoint in v1. If you want to ignore the result, just don't read it.

Worked example — CLI-style trace renderer

Below is a ~70-line script that produces output close to the trace you see in the README's real-turn snippets — phases, agents, validator verdicts, costs, and the final answer streamed in.

// trace.ts
import { Amy } from "@amy/sdk";

const amy = new Amy({
  apiKey: process.env.AMY_API_KEY!,
  baseUrl: process.env.AMY_BASE_URL,
});

const turn = await amy.turns.create({
  messages: [{ role: "user", content: process.argv[2] ?? "How's my recovery?" }],
});

const start = Date.now();
const ts = () => {
  const ms = Date.now() - start;
  return `[+${String(Math.floor(ms / 1000)).padStart(3, " ")}s]`;
};
const ANSI = { dim: "\x1b[2m", reset: "\x1b[0m", green: "\x1b[32m", red: "\x1b[31m", yellow: "\x1b[33m" };
const verdictColor = (v: string) =>
  v === "validated" ? ANSI.green : v === "rejected" ? ANSI.red : ANSI.yellow;
const pad = (s: string, n = 20) => s.padEnd(n);

for await (const event of amy.turns.stream(turn.id)) {
  switch (event.type) {
    case "turn.started":
      console.log(`${ANSI.dim}${ts()} ${pad("orchestrator")} → turn started${ANSI.reset}`);
      break;

    case "phase":
      console.log(
        `${ANSI.dim}${ts()} ${pad(event.data.agent)} • ${event.data.phase}${event.data.detail ? ` (${event.data.detail})` : ""}${ANSI.reset}`,
      );
      break;

    case "agent_start":
      console.log(`${ANSI.dim}${ts()} ${pad(event.data.agent)} → running${ANSI.reset}`);
      break;

    case "agent_end":
      console.log(
        `${ANSI.dim}${ts()} ${pad(event.data.agent)} ✓ done in ${event.data.duration_ms}ms ($${event.data.cost_usd.toFixed(4)})${ANSI.reset}`,
      );
      break;

    case "validation_end":
      console.log(
        `${ANSI.dim}${ts()} ${pad("validator")} ${verdictColor(event.data.verdict)}■ ${event.data.verdict.toUpperCase().padEnd(11)}${ANSI.reset}${ANSI.dim} ${event.data.finding_id}${ANSI.reset}`,
      );
      break;

    case "synthesis_delta":
      process.stdout.write(event.data.text);
      break;

    case "synthesis_done":
      process.stdout.write("\n");
      break;

    case "turn.completed":
      console.log(
        `${ANSI.dim}${ts()} ${pad("orchestrator")} ✓ turn complete ($${event.data.result.cost_usd.toFixed(4)})${ANSI.reset}`,
      );
      break;

    case "turn.failed":
      console.error(`\n${ANSI.red}✗ turn failed: ${event.data.error.message}${ANSI.reset}`);
      process.exit(1);
  }
}

Run it:

bun trace.ts "is my -7.3% sleep score drop clinically meaningful?"

You'll get output like:

[+  0s] orchestrator         → turn started
[+  0s] orchestrator         • classifying query vagueness
[+  1s] orchestrator         • vagueness = low (specific — direct routing)
[+  1s] Data Science Agent   → running
[+ 64s] Data Science Agent   ✓ done in 64200ms ($0.0895)
[+ 64s] validator            ■ VALIDATED   ds-001
[+138s] Domain Expert Agent  → running
[+201s] Domain Expert Agent  ✓ done in 63100ms ($0.0341)
Short answer: no — by the most defensible read of your own data…
[+222s] orchestrator         ✓ turn complete ($0.1288)

Common mistakes

Reaching for `event.delta` / `event.agent` instead of `event.data.*`

Every typed field lives under event.data. The SDK wraps each SSE frame as { type, id, data }; accessing event.delta will be undefined.

Filtering for an `agent.thought` event with `agent === "synthesis"`

agent.thought and a "synthesis" agent name are both legacy. The final answer streams as synthesis_delta frames with data.text chunks. Per-agent thoughts are not streamed at the token level in v1.

Treating the streamed tokens as the final answer

synthesis_delta is a best-effort token feed. The authoritative answer is event.data.result.answer on turn.completed — that's what was committed to D1 and what GET /v1/turns/:id returns. If you display the streamed tokens, swap them out for result.answer on turn.completed to handle any dropped chunks or fact-check rewrites.

Forgetting that browser `EventSource` can't set Authorization

The browser's built-in EventSource API has no way to set request headers. If you try to use it directly with Authorization: Bearer …, it silently sends no header and the server returns 401 (which EventSource reports as a generic onerror). Use the SDK's iterator — it uses fetch + ReadableStream and sets the header normally.

Reconnecting from event id 0 after a disconnect

If you reconnect without Last-Event-Id, the server replays the entire stream from the start, and you'll re-render every event, re-stream every token, and confuse your UI. Track the last id you saw on every event and pass it on reconnect.

Holding the connection open after `turn.completed`

The server closes the stream after turn.completed or turn.failed. If you keep your client's iterator alive, you'll eventually hit "stream closed" on the next read. Break out of the loop on the terminal events:

for await (const event of amy.turns.stream(turn.id)) {
  handleEvent(event);
  if (event.type === "turn.completed" || event.type === "turn.failed") break;
}

(The SDK iterator already does this internally; the break is only needed if you're parsing raw SSE yourself.)

Where to next

Just want the answer with no UI? Recipe: Ask a question.
Building a chat UI from scratch? Recipe: Build a mobile app — uses this stream for the chat screen.
Full event schema: SDK: TypeScript / Streaming protocol.
How the events get produced in the first place: Architecture: Streaming.

Stream events from a turn

On this page