Type-safe streaming with async generators in TypeScript

Most LLM SDKs hand you a raw stream of unknown chunks and expect you to parse Server-Sent Events yourself. This tutorial shows how to wrap that mess in typescript async generator streaming so your call sites get typed tokens, finish reasons, and usage metadata without a single any. We’ll build a minimal client for any OpenAI-compatible chat completions endpoint and run it against a real API.

Prerequisites

• Node.js 18+ (global fetch and TextDecoder are available)
• TypeScript 5.2+ and tsx for running TS directly: npm i -D tsx
• An OpenAI-compatible endpoint URL and API key. For example, n4n.ai exposes one OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is degraded, so the same code works across vendors.
• Basic comfort with async/await and generator functions.

Create a file stream.ts and export your credentials:

export OPENAI_KEY=sk-...
# or point at another base URL
export BASE_URL=https://api.openai.com/v1
export MODEL=gpt-4o-mini

The wire format

OpenAI-compatible streaming returns newline-delimited data: frames. Each frame is a JSON StreamChunk. The stream ends with data: [DONE]. We encode the contract in TypeScript up front.

// types.ts
export interface ChatMessage {
  role: 'system' | 'user' | 'assistant';
  content: string;
}

export interface Usage {
  prompt_tokens: number;
  completion_tokens: number;
  total_tokens: number;
}

export interface StreamChunk {
  id: string;
  object: 'chat.completion.chunk';
  choices: {
    delta: { role?: string; content?: string };
    index: number;
    finish_reason: string | null;
  }[];
  usage?: Usage;
}

If you skip this step, you’re back to guessing field names at runtime. The whole point of typescript async generator streaming is to push those guesses to the edge.

Building the typed generator

We’ll yield a discriminated union instead of raw strings. That lets consumers switch on type and get full type narrowing.

// stream.ts
import type { ChatMessage, StreamChunk, Usage } from './types';

export type StreamEvent =
  | { type: 'token'; text: string }
  | { type: 'finish'; reason: string | null }
  | { type: 'usage'; usage: Usage };

export async function* streamChatEvents(
  messages: ChatMessage[],
  opts: { apiKey: string; baseUrl: string; model: string; signal?: AbortSignal }
): AsyncGenerator<StreamEvent, void, unknown> {
  const res = await fetch(`${opts.baseUrl}/chat/completions`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${opts.apiKey}`,
    },
    body: JSON.stringify({ model: opts.model, messages, stream: true }),
    signal: opts.signal,
  });

  if (!res.ok) throw new Error(`HTTP ${res.status}: ${await res.text()}`);
  if (!res.body) throw new Error('Empty response body');

  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n');
    buffer = lines.pop() ?? '';

    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed.startsWith('data:')) continue;
      const data = trimmed.slice(5).trim();
      if (data === '[DONE]') return;
      const chunk = JSON.parse(data) as StreamChunk;
      for (const choice of chunk.choices) {
        if (choice.delta.content) yield { type: 'token', text: choice.delta.content };
        if (choice.finish_reason) yield { type: 'finish', reason: choice.finish_reason };
      }
      if (chunk.usage) yield { type: 'usage', usage: chunk.usage };
    }
  }
}

The generator yields exactly three event shapes. There is no any in the loop—JSON.parse is cast once at the boundary, which is the correct place for that risk.

Checkpoint: stream raw tokens

Run a quick script to verify the shape:

// run.ts
import { streamChatEvents } from './stream';

const opts = {
  apiKey: process.env.OPENAI_KEY!,
  baseUrl: process.env.BASE_URL!,
  model: process.env.MODEL!,
};

for await (const ev of streamChatEvents(
  [{ role: 'user', content: 'Say "hello" in one word.' }],
  opts
)) {
  if (ev.type === 'token') process.stdout.write(ev.text);
  if (ev.type === 'finish') console.log(`\n[finish: ${ev.reason}]`);
}

npx tsx run.ts

Expected stdout (approximate):

hello
[finish: stop]

If you see the token printed incrementally and the finish line, your typescript async generator streaming pipeline is working.

Testing with a mock stream

Before hitting a paid API, mock fetch to return a local SSE string. This validates your parser offline.

// mock.ts
const fakeSse = [
  'data: {"choices":[{"delta":{"content":"Hello"}}]}\n',
  'data: {"choices":[{"delta":{"content":" world"}}]}\n',
  'data: {"choices":[{"finish_reason":"stop"}]}\n',
  'data: [DONE]\n',
].join('');

global.fetch = async () => {
  const body = new ReadableStream({
    start(c) { c.enqueue(new TextEncoder().encode(fakeSse)); c.close(); }
  });
  return new Response(body, { status: 200 });
};

// import and run streamChatEvents against the mock

Expected output when running against the mock:

Hello world
[finish: stop]

This proves the streaming logic without network access.

Collecting a full response

Streaming is great for UX, but sometimes you want the complete string plus usage after the stream ends. Wrap the generator in a promise-returning helper.

export interface CompletionResult {
  content: string;
  usage?: Usage;
}

export async function complete(
  messages: ChatMessage[],
  opts: { apiKey: string; baseUrl: string; model: string; signal?: AbortSignal }
): Promise<CompletionResult> {
  let content = '';
  let usage: Usage | undefined;
  for await (const ev of streamChatEvents(messages, opts)) {
    if (ev.type === 'token') content += ev.text;
    if (ev.type === 'usage') usage = ev.usage;
  }
  return { content, usage };
}

Now call sites get a typed object:

const result = await complete(
  [{ role: 'user', content: 'Explain async generators in 10 words.' }],
  opts
);
console.log(result.content);
console.log(result.usage?.total_tokens ?? 'no usage returned');

Some providers omit usage on streaming chunks unless you send stream_options: { include_usage: true }. Add that to the request body if you need per-token metering:

body: JSON.stringify({
  model: opts.model,
  messages,
  stream: true,
  stream_options: { include_usage: true },
}),

Cancellation and timeouts

A generator that ignores AbortSignal is a leak waiting to happen. We already pass signal to fetch. To abort mid-stream, call controller.abort() and the for await loop will throw AbortError, which you catch at the call site.

const controller = new AbortController();
setTimeout(() => controller.abort(), 5000);

try {
  for await (const ev of streamChatEvents(messages, { ...opts, signal: controller.signal })) {
    if (ev.type === 'token') process.stdout.write(ev.text);
  }
} catch (err) {
  if ((err as Error).name === 'AbortError') console.log('\nTimed out');
  else throw err;
}

Because the generator respects the signal, no stray reads continue after the timeout.

Composing generators

The real win of typescript async generator streaming is composition. You can map, filter, or merge streams with standard async iteration. Example: drop everything except tokens and uppercase them.

async function* upperTokens(
  messages: ChatMessage[],
  opts: Parameters<typeof streamChatEvents>[1]
): AsyncGenerator<string> {
  for await (const ev of streamChatEvents(messages, opts)) {
    if (ev.type === 'token') yield ev.text.toUpperCase();
  }
}

TypeScript infers ev as StreamEvent inside the loop, so ev.text is only accessible in the token branch. That’s the safety we wanted.

Pointing at a multi-model gateway

If you swap BASE_URL to an OpenAI-compatible gateway like n4n.ai, the same streamChatEvents function addresses 240+ models and inherits automatic fallback when a backend is rate-limited. You don’t change a line of parsing code; you only change the model string and key. That’s the benefit of coding against the contract rather than a vendor SDK.

const opts = {
  apiKey: process.env.N4N_KEY!,
  baseUrl: 'https://api.n4n.ai/v1', // OpenAI-compatible
  model: 'anthropic/claude-3.5-sonnet',
};

The generator doesn’t care. It yields the same StreamEvents.

Common pitfalls

• Line splitting across chunks: TCP can split a frame mid-line. We keep a buffer and only process complete lines after split('\n'), preserving the tail.
• Double JSON parsing: Parse once at the StreamChunk cast. Don’t re-parse inside the loop.
• Missing [DONE]: Some proxies send it, some close the stream. Our loop handles both: done from the reader and the explicit token.

Final checklist

• Defined wire types at the boundary (StreamChunk, Usage).
• Yielded a discriminated StreamEvent union from an AsyncGenerator.
• Consumed it with for await, narrowing by type.
• Added AbortSignal for cancellation.
• Wrapped streaming into a typed complete() for batch callers.
• Validated with a mock SSE stream before spending money.

You now have a reusable, type-safe streaming core that compiles to clean JavaScript and works against any compliant endpoint. The next step is to add retries with exponential backoff—but because you yield structured events, a retry wrapper can simply re-invoke the generator and concatenate tokens without touching your UI code.