Streaming Patterns

Introduction to Streaming

The Claude Code TypeScript SDK uses async generators for streaming responses, providing memory-efficient, real-time processing of Claude’s outputs. This approach is essential for:

Long-running operations
Real-time user feedback
Memory-constrained environments
Interactive applications

Note: While the classic async generator API is still supported, the fluent API is now recommended for new projects as of 2025. The fluent API provides a chainable interface that’s more intuitive and powerful.

Basic Streaming Concepts

Modern Fluent API (Recommended for 2025)

The SDK now provides a fluent, chainable TypeScript API that offers better developer experience:

import { claude } from '@anthropic-ai/claude-code';
 
// Fluent API example
const result = await claude('Write a haiku about foo.py')
  .allowTools('Read', 'Write')
  .setModel('claude-sonnet-4')
  .onMessage(msg => console.log('Processing:', msg))
  .onToolUse(tool => console.log('Using tool:', tool))
  .asText();
 
console.log(result);

Key features of the fluent API:

Configure models, enable tools, stream events in one chain
Multi-level logging with live callbacks
Deep, CLI-compatible observability
Less memory usage compared to classic API

Classic Async Iterator Pattern (Legacy)

import { query, type SDKMessage } from "@anthropic-ai/claude-code";
 
// The query function returns an AsyncGenerator
async function* streamMessages(prompt: string): AsyncGenerator<SDKMessage> {
  const abortController = new AbortController();
  
  for await (const message of query({ prompt, abortController })) {
    yield message;
  }
}
 
// Consuming the stream
async function consume() {
  for await (const message of streamMessages("Hello")) {
    console.log(message);
  }
}

Streaming Patterns

1. Real-Time Display Pattern

async function realtimeDisplay(prompt: string) {
  const abortController = new AbortController();
  let buffer = "";
  
  for await (const message of query({ prompt, abortController })) {
    if (message.type === 'assistant') {
      const content = message.data.message.content;
      
      if (typeof content === 'string') {
        // Clear previous line and write new content
        process.stdout.write('\r' + ' '.repeat(buffer.length) + '\r');
        buffer = content;
        process.stdout.write(buffer);
      }
    }
  }
  
  console.log(); // New line after completion
}

2. Token Streaming Pattern

interface TokenStream {
  token: string;
  position: number;
  timestamp: number;
}
 
async function* tokenStream(prompt: string): AsyncGenerator<TokenStream> {
  const abortController = new AbortController();
  let position = 0;
  let lastContent = "";
  
  for await (const message of query({ prompt, abortController })) {
    if (message.type === 'assistant') {
      const content = message.data.message.content;
      
      if (typeof content === 'string') {
        // Extract new tokens
        const newContent = content.slice(lastContent.length);
        
        for (const token of newContent) {
          yield {
            token,
            position: position++,
            timestamp: Date.now()
          };
        }
        
        lastContent = content;
      }
    }
  }
}
 
// Usage
async function displayTokens() {
  for await (const { token, position } of tokenStream("Generate a story")) {
    process.stdout.write(token);
    
    // Add dramatic pauses for effect
    if (token === '.' || token === '!' || token === '?') {
      await new Promise(resolve => setTimeout(resolve, 300));
    }
  }
}

3. Buffered Streaming Pattern

class BufferedStream {
  private buffer: SDKMessage[] = [];
  private bufferSize: number;
  
  constructor(bufferSize: number = 10) {
    this.bufferSize = bufferSize;
  }
  
  async *stream(prompt: string): AsyncGenerator<SDKMessage[]> {
    const abortController = new AbortController();
    
    for await (const message of query({ prompt, abortController })) {
      this.buffer.push(message);
      
      if (this.buffer.length >= this.bufferSize) {
        yield [...this.buffer];
        this.buffer = [];
      }
    }
    
    // Yield remaining messages
    if (this.buffer.length > 0) {
      yield [...this.buffer];
    }
  }
}
 
// Usage
const buffered = new BufferedStream(5);
for await (const batch of buffered.stream("Large task")) {
  await processBatch(batch);
}

4. Transform Stream Pattern

async function* transformStream<T>(
  prompt: string,
  transformer: (message: SDKMessage) => T | null
): AsyncGenerator<T> {
  const abortController = new AbortController();
  
  for await (const message of query({ prompt, abortController })) {
    const transformed = transformer(message);
    if (transformed !== null) {
      yield transformed;
    }
  }
}
 
// Example: Extract only code blocks
async function* codeBlockStream(prompt: string): AsyncGenerator<string> {
  yield* transformStream(prompt, (message) => {
    if (message.type === 'assistant') {
      const content = message.data.message.content;
      if (typeof content === 'string') {
        const codeMatch = content.match(/```[\s\S]*?```/g);
        return codeMatch ? codeMatch[0] : null;
      }
    }
    return null;
  });
}

5. Parallel Stream Processing

class ParallelStreamProcessor {
  async processMultiple(prompts: string[]) {
    const streams = prompts.map(prompt => ({
      prompt,
      iterator: query({ 
        prompt, 
        abortController: new AbortController() 
      })
    }));
    
    // Process all streams concurrently
    const results = await Promise.all(
      streams.map(async ({ prompt, iterator }) => {
        const messages: SDKMessage[] = [];
        
        for await (const message of iterator) {
          messages.push(message);
          
          // Emit progress events
          this.emitProgress(prompt, message);
        }
        
        return { prompt, messages };
      })
    );
    
    return results;
  }
  
  private emitProgress(prompt: string, message: SDKMessage) {
    console.log(`[${prompt.slice(0, 20)}...]:`, message.type);
  }
}

Advanced Streaming Techniques

1. Cancellable Streams

class CancellableStream {
  private abortController: AbortController;
  
  constructor() {
    this.abortController = new AbortController();
  }
  
  async *stream(prompt: string): AsyncGenerator<SDKMessage> {
    try {
      for await (const message of query({ 
        prompt, 
        abortController: this.abortController 
      })) {
        yield message;
      }
    } catch (error) {
      if (error.name === 'AbortError') {
        console.log('Stream cancelled');
        // See [[docs/core/typescript-sdk/error-handling|Error Handling]] for more details.
      } else {
        throw error;
      }
    }
  }
  
  cancel() {
    this.abortController.abort();
  }
}
 
// Usage with timeout
async function streamWithTimeout(prompt: string, timeoutMs: number = 30000) {
  const stream = new CancellableStream();
  
  const timeout = setTimeout(() => {
    console.log('Timeout reached, cancelling stream');
    stream.cancel();
  }, timeoutMs);
  
  try {
    for await (const message of stream.stream(prompt)) {
      console.log(message);
    }
  } finally {
    clearTimeout(timeout);
  }
}

2. Stream Multiplexing

class StreamMultiplexer {
  private subscribers: ((message: SDKMessage) => void)[] = [];
  
  subscribe(callback: (message: SDKMessage) => void) {
    this.subscribers.push(callback);
    
    return () => {
      const index = this.subscribers.indexOf(callback);
      if (index > -1) {
        this.subscribers.splice(index, 1);
      }
    };
  }
  
  async multiplex(prompt: string) {
    const abortController = new AbortController();
    
    for await (const message of query({ prompt, abortController })) {
      // Broadcast to all subscribers
      this.subscribers.forEach(callback => callback(message));
    }
  }
}
 
// Usage
const multiplexer = new StreamMultiplexer();
 
// Subscribe multiple handlers
multiplexer.subscribe(msg => console.log('Logger:', msg.type));
multiplexer.subscribe(msg => {
  if (msg.type === 'result') {
    console.log('Cost tracker:', msg.data.total_cost_usd);
  }
});
 
await multiplexer.multiplex("Complex task");

3. Stream Filtering and Mapping

class StreamPipeline {
  static async *filter(
    stream: AsyncGenerator<SDKMessage>,
    predicate: (msg: SDKMessage) => boolean
  ): AsyncGenerator<SDKMessage> {
    for await (const message of stream) {
      if (predicate(message)) {
        yield message;
      }
    }
  }
  
  static async *map<T>(
    stream: AsyncGenerator<SDKMessage>,
    mapper: (msg: SDKMessage) => T
  ): AsyncGenerator<T> {
    for await (const message of stream) {
      yield mapper(message);
    }
  }
  
  static async *take(
    stream: AsyncGenerator<SDKMessage>,
    count: number
  ): AsyncGenerator<SDKMessage> {
    let taken = 0;
    for await (const message of stream) {
      if (taken >= count) break;
      yield message;
      taken++;
    }
  }
}
 
// Usage: Get first 5 assistant messages
async function getFirstAssistantMessages(prompt: string, count: number = 5) {
  const stream = query({ prompt, abortController: new AbortController() });
  
  const filtered = StreamPipeline.filter(
    stream,
    msg => msg.type === 'assistant'
  );
  
  const limited = StreamPipeline.take(filtered, count);
  
  const messages: string[] = [];
  
  for await (const message of limited) {
    if (message.type === 'assistant') {
      messages.push(message.data.message.content as string);
    }
  }
  
  return messages;
}

4. Stream State Management

class StatefulStream {
  private state: {
    messageCount: number;
    totalCost: number;
    startTime: number;
    errors: Error[];
  };
  
  constructor() {
    this.state = {
      messageCount: 0,
      totalCost: 0,
      startTime: Date.now(),
      errors: []
    };
  }
  
  async *stream(prompt: string): AsyncGenerator<{
    message: SDKMessage;
    state: typeof this.state;
  }> {
    const abortController = new AbortController();
    
    try {
      for await (const message of query({ prompt, abortController })) {
        this.updateState(message);
        
        yield {
          message,
          state: { ...this.state }
        };
      }
    } catch (error) {
      this.state.errors.push(error as Error);
      throw error;
    }
  }
  
  private updateState(message: SDKMessage) {
    this.state.messageCount++;
    
    if (message.type === 'result' && message.data.total_cost_usd) {
      this.state.totalCost += message.data.total_cost_usd;
    }
  }
  
  getStats() {
    return {
      ...this.state,
      duration: Date.now() - this.state.startTime
    };
  }
}

Performance Optimization

1. Backpressure Handling

class BackpressureStream {
  private queue: SDKMessage[] = [];
  private processing = false;
  private maxQueueSize: number;
  
  constructor(maxQueueSize: number = 100) {
    this.maxQueueSize = maxQueueSize;
  }
  
  async stream(prompt: string, processor: (msg: SDKMessage) => Promise<void>) {
    const abortController = new AbortController();
    
    for await (const message of query({ prompt, abortController })) {
      // Apply backpressure if queue is full
      while (this.queue.length >= this.maxQueueSize) {
        await new Promise(resolve => setTimeout(resolve, 100));
      }
      
      this.queue.push(message);
      
      if (!this.processing) {
        this.processQueue(processor);
      }
    }
  }
  
  private async processQueue(processor: (msg: SDKMessage) => Promise<void>) {
    this.processing = true;
    
    while (this.queue.length > 0) {
      const message = this.queue.shift()!;
      await processor(message);
    }
    
    this.processing = false;
  }
}

2. Memory-Efficient Large Stream Processing

async function processLargeStream(prompt: string) {
  const abortController = new AbortController();
  const chunkSize = 1000; // Characters per chunk
  let currentChunk = "";
  
  for await (const message of query({ prompt, abortController })) {
    if (message.type === 'assistant') {
      const content = message.data.message.content;
      
      if (typeof content === 'string') {
        currentChunk += content;
        
        // Process and clear chunks to prevent memory buildup
        while (currentChunk.length >= chunkSize) {
          const chunk = currentChunk.slice(0, chunkSize);
          await processChunk(chunk);
          currentChunk = currentChunk.slice(chunkSize);
        }
      }
    }
  }
  
  // Process remaining content
  if (currentChunk) {
    await processChunk(currentChunk);
  }
}
 
async function processChunk(chunk: string) {
  // Process chunk without keeping in memory
  console.log(`Processed ${chunk.length} characters`);
}

Best Practices

Always use AbortController for cancellable operations
Process messages immediately instead of accumulating in memory
Implement proper error handling for stream interruptions
Use backpressure when processing can’t keep up with streaming
Monitor memory usage for long-running streams
Add timeouts for streams that might hang
Log stream progress for debugging and monitoring

Common Patterns Summary

AsyncIterator Pattern (from Patterns Library)

The SDK provides built-in streaming support through async iterators with these key benefits:

Clean Abstraction: AsyncGenerator provides elegant streaming interface
Type Safety: Maintained throughout stream lifecycle
Error Handling: Implemented at both stream and chunk level

Core Implementation Pattern

// Streaming implementation pattern for clean abstraction
async function* streamResponse<T>(
  client: ClaudeClient,
  prompt: string
): AsyncGenerator<T> {
  const stream = await client.messages.stream({
    model: 'claude-3-opus-20240229',
    messages: [{ role: 'user', content: prompt }],
    max_tokens: 1024,
  });
 
  for await (const chunk of stream) {
    if (chunk.type === 'content_block_delta') {
      yield chunk.delta as T;
    }
  }
}

Stream Transformations

type StreamTransformer<T, U> = (input: AsyncIterable<T>) => AsyncIterable<U>;
 
function mapStream<T, U>(fn: (item: T) => U): StreamTransformer<T, U> {
  return async function* (stream) {
    for await (const item of stream) {
      yield fn(item);
    }
  };
}

Retry Logic for Streams

async function* streamWithRetry<T>(
  streamFn: () => AsyncGenerator<T>,
  maxRetries = 3
): AsyncGenerator<T> {
  let retries = 0;
  while (retries < maxRetries) {
    try {
      yield* streamFn();
      break;
    } catch (error) {
      retries++;
      if (retries === maxRetries) throw error;
      await new Promise(resolve => setTimeout(resolve, 1000 * retries));
    }
  }
}

Stream Aggregation

async function* batchStream<T>(
  stream: AsyncIterable<T>,
  batchSize: number
): AsyncGenerator<T[]> {
  let batch: T[] = [];
  
  for await (const item of stream) {
    batch.push(item);
    if (batch.length === batchSize) {
      yield batch;
      batch = [];
    }
  }
  
  if (batch.length > 0) {
    yield batch;
  }
}

Type Inference Best Practices

Problem: TypeScript cannot infer types in deeply nested async generators.

Solution: Explicitly annotate generator return types:

// Explicit type annotation
async function* nestedStream(): AsyncGenerator<string> {
  const innerStream: AsyncGenerator<number> = getNumberStream();
  for await (const num of innerStream) {
    yield String(num);
  }
}

Next Steps

Explore Error Handling for robust stream error management
Learn about Tool Integration for streaming with tools
See Practical Examples for real-world streaming use cases

Tags: claude-code typescript streaming async-generators real-time fluent-api patterns

Verifications

Claude Code SDK Documentation: Verified against official Anthropic documentation at https://docs.anthropic.com/en/docs/claude-code/sdk (2025-07-20)
Fluent API: Added information about the new fluent API recommended for 2025 based on latest SDK updates from https://github.com/instantlyeasy/claude-code-sdk-ts
Streaming Best Practices: Confirmed current best practices from Anthropic’s engineering blog at https://www.anthropic.com/engineering/claude-code-best-practices
Pattern Library: Consolidated patterns from the TypeScript patterns library for unified reference

Claude Code Research

Explorer

Streaming Patterns

Streaming Patterns

Introduction to Streaming

Basic Streaming Concepts

Modern Fluent API (Recommended for 2025)

Classic Async Iterator Pattern (Legacy)

Streaming Patterns

1. Real-Time Display Pattern

2. Token Streaming Pattern

3. Buffered Streaming Pattern

4. Transform Stream Pattern

5. Parallel Stream Processing

Advanced Streaming Techniques

1. Cancellable Streams

2. Stream Multiplexing

3. Stream Filtering and Mapping

4. Stream State Management

Performance Optimization

1. Backpressure Handling

2. Memory-Efficient Large Stream Processing

Best Practices

Common Patterns Summary

AsyncIterator Pattern (from Patterns Library)

Core Implementation Pattern

Stream Transformations

Retry Logic for Streams

Stream Aggregation

Type Inference Best Practices

Next Steps

Verifications

Graph View

Table of Contents

Backlinks

Claude Code Research

Explorer

Streaming Patterns

Streaming Patterns

Introduction to Streaming

Basic Streaming Concepts

Modern Fluent API (Recommended for 2025)

Classic Async Iterator Pattern (Legacy)

Streaming Patterns

1. Real-Time Display Pattern

2. Token Streaming Pattern

3. Buffered Streaming Pattern

4. Transform Stream Pattern

5. Parallel Stream Processing

Advanced Streaming Techniques

1. Cancellable Streams

2. Stream Multiplexing

3. Stream Filtering and Mapping

4. Stream State Management

Performance Optimization

1. Backpressure Handling

2. Memory-Efficient Large Stream Processing

Best Practices

Common Patterns Summary

AsyncIterator Pattern (from Patterns Library)

Core Implementation Pattern

Stream Transformations

Retry Logic for Streams

Stream Aggregation

Type Inference Best Practices

Next Steps

Related Documentation

Verifications

Graph View

Table of Contents

Backlinks