Real-Time Data Streaming with Claude Code

This guide explores patterns and best practices for building real-time applications with Claude Code, focusing on WebSocket and Server-Sent Events (SSE) integration for live data streaming, collaborative features, and responsive AI interactions.

Overview

Real-time data streaming with Claude Code enables:

• Live AI response streaming with typewriter effects
• Real-time collaboration features
• Live data analysis and monitoring
• Interactive dashboards with AI insights
• Streaming code generation and execution

Streaming Technologies Comparison

Server-Sent Events (SSE)

• Direction: Server → Client only
• Protocol: HTTP/HTTPS
• Reconnection: Automatic
• Best for: Live updates, notifications, streaming AI responses

WebSockets

• Direction: Bidirectional
• Protocol: WS/WSS
• Reconnection: Manual
• Best for: Chat, collaboration, real-time interaction

Claude Streaming

• Native Support: SSE-based streaming
• Token Streaming: Real-time token generation
• Tool Calls: Streaming execution updates

Implementation Patterns

1. Basic SSE Streaming with Claude

// server/claude-sse.ts
import { Anthropic } from '@anthropic-ai/sdk';
import express from 'express';
 
const app = express();
const anthropic = new Anthropic();
 
app.get('/api/claude-stream', async (req, res) => {
  // Set up SSE headers
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'Access-Control-Allow-Origin': '*'
  });
 
  const { prompt } = req.query;
 
  try {
    const stream = await anthropic.messages.create({
      model: 'claude-3-sonnet-20240229',
      messages: [{ role: 'user', content: prompt as string }],
      max_tokens: 1024,
      stream: true
    });
 
    for await (const event of stream) {
      if (event.type === 'content_block_delta') {
        const data = {
          type: 'delta',
          content: event.delta.text,
          index: event.index
        };
        res.write(`data: ${JSON.stringify(data)}\n\n`);
      } else if (event.type === 'message_stop') {
        res.write('data: [DONE]\n\n');
        res.end();
      }
    }
  } catch (error) {
    res.write(`data: ${JSON.stringify({ error: error.message })}\n\n`);
    res.end();
  }
});
 
// Client-side consumption
class ClaudeSSEClient {
  private eventSource: EventSource | null = null;
  
  connect(prompt: string, onMessage: (data: any) => void) {
    const url = `/api/claude-stream?prompt=${encodeURIComponent(prompt)}`;
    this.eventSource = new EventSource(url);
    
    this.eventSource.onmessage = (event) => {
      if (event.data === '[DONE]') {
        this.disconnect();
        return;
      }
      
      const data = JSON.parse(event.data);
      onMessage(data);
    };
    
    this.eventSource.onerror = (error) => {
      console.error('SSE error:', error);
      this.disconnect();
    };
  }
  
  disconnect() {
    if (this.eventSource) {
      this.eventSource.close();
      this.eventSource = null;
    }
  }
}

2. WebSocket Integration for Bidirectional Communication

// server/claude-websocket.ts
import { WebSocketServer } from 'ws';
import { Anthropic } from '@anthropic-ai/sdk';
 
interface ClaudeSession {
  id: string;
  anthropic: Anthropic;
  context: any[];
  activeStream?: any;
}
 
class ClaudeWebSocketServer {
  private wss: WebSocketServer;
  private sessions: Map<string, ClaudeSession> = new Map();
  
  constructor(port: number) {
    this.wss = new WebSocketServer({ port });
    this.setupHandlers();
  }
  
  private setupHandlers() {
    this.wss.on('connection', (ws) => {
      const sessionId = this.generateSessionId();
      const session: ClaudeSession = {
        id: sessionId,
        anthropic: new Anthropic(),
        context: []
      };
      
      this.sessions.set(sessionId, session);
      
      ws.on('message', async (data) => {
        const message = JSON.parse(data.toString());
        await this.handleMessage(ws, session, message);
      });
      
      ws.on('close', () => {
        // Clean up session
        if (session.activeStream) {
          session.activeStream.controller?.abort();
        }
        this.sessions.delete(sessionId);
      });
      
      // Send initial connection confirmation
      ws.send(JSON.stringify({
        type: 'connected',
        sessionId
      }));
    });
  }
  
  private async handleMessage(ws: any, session: ClaudeSession, message: any) {
    switch (message.type) {
      case 'query':
        await this.handleQuery(ws, session, message);
        break;
      case 'stop':
        if (session.activeStream) {
          session.activeStream.controller?.abort();
        }
        break;
      case 'context':
        session.context = message.context || [];
        break;
    }
  }
  
  private async handleQuery(ws: any, session: ClaudeSession, message: any) {
    try {
      const controller = new AbortController();
      
      const stream = await session.anthropic.messages.create({
        model: 'claude-3-sonnet-20240229',
        messages: [
          ...session.context,
          { role: 'user', content: message.prompt }
        ],
        max_tokens: 1024,
        stream: true
      }, {
        signal: controller.signal
      });
      
      session.activeStream = { stream, controller };
      
      for await (const event of stream) {
        if (event.type === 'content_block_delta') {
          ws.send(JSON.stringify({
            type: 'delta',
            content: event.delta.text,
            index: event.index
          }));
        } else if (event.type === 'message_stop') {
          ws.send(JSON.stringify({
            type: 'complete',
            usage: event.usage
          }));
          
          // Update context
          session.context.push(
            { role: 'user', content: message.prompt },
            { role: 'assistant', content: event.message.content }
          );
        }
      }
    } catch (error) {
      ws.send(JSON.stringify({
        type: 'error',
        error: error.message
      }));
    }
  }
  
  private generateSessionId(): string {
    return Math.random().toString(36).substring(7);
  }
}
 
// Client-side WebSocket handler
class ClaudeWebSocketClient {
  private ws: WebSocket | null = null;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 5;
  
  connect(url: string, handlers: {
    onMessage?: (data: any) => void;
    onError?: (error: any) => void;
    onConnect?: (sessionId: string) => void;
  }) {
    this.ws = new WebSocket(url);
    
    this.ws.onopen = () => {
      this.reconnectAttempts = 0;
      console.log('WebSocket connected');
    };
    
    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      
      switch (data.type) {
        case 'connected':
          handlers.onConnect?.(data.sessionId);
          break;
        case 'delta':
        case 'complete':
          handlers.onMessage?.(data);
          break;
        case 'error':
          handlers.onError?.(data.error);
          break;
      }
    };
    
    this.ws.onerror = (error) => {
      handlers.onError?.(error);
    };
    
    this.ws.onclose = () => {
      this.handleReconnect(url, handlers);
    };
  }
  
  private handleReconnect(url: string, handlers: any) {
    if (this.reconnectAttempts < this.maxReconnectAttempts) {
      this.reconnectAttempts++;
      const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 10000);
      
      setTimeout(() => {
        console.log(`Reconnecting... (attempt ${this.reconnectAttempts})`);
        this.connect(url, handlers);
      }, delay);
    }
  }
  
  sendQuery(prompt: string) {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify({
        type: 'query',
        prompt
      }));
    }
  }
  
  stop() {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify({ type: 'stop' }));
    }
  }
  
  disconnect() {
    if (this.ws) {
      this.ws.close();
      this.ws = null;
    }
  }
}

3. Real-Time Collaboration Pattern

// Collaborative editing with Claude assistance
interface CollaborativeSession {
  documentId: string;
  participants: Map<string, Participant>;
  document: Document;
  claudeContext: any[];
}
 
class CollaborativeClaudeEditor {
  private sessions: Map<string, CollaborativeSession> = new Map();
  private anthropic: Anthropic;
  
  constructor() {
    this.anthropic = new Anthropic();
  }
  
  async handleCollaborativeAction(
    sessionId: string,
    userId: string,
    action: any
  ) {
    const session = this.sessions.get(sessionId);
    if (!session) return;
    
    switch (action.type) {
      case 'edit':
        // Apply edit
        this.applyEdit(session, action.edit);
        
        // Broadcast to other participants
        this.broadcast(session, userId, {
          type: 'edit',
          edit: action.edit,
          userId
        });
        break;
        
      case 'ai-assist':
        // Stream AI assistance to all participants
        await this.streamAIAssistance(session, action.prompt);
        break;
        
      case 'cursor-move':
        // Update cursor position
        this.updateCursor(session, userId, action.position);
        break;
    }
  }
  
  private async streamAIAssistance(
    session: CollaborativeSession,
    prompt: string
  ) {
    const stream = await this.anthropic.messages.create({
      model: 'claude-3-sonnet-20240229',
      messages: [
        {
          role: 'system',
          content: 'You are assisting with collaborative document editing.'
        },
        ...session.claudeContext,
        {
          role: 'user',
          content: `Document context: ${session.document.content}\n\nRequest: ${prompt}`
        }
      ],
      stream: true
    });
    
    let fullResponse = '';
    
    for await (const event of stream) {
      if (event.type === 'content_block_delta') {
        fullResponse += event.delta.text;
        
        // Broadcast delta to all participants
        this.broadcastToAll(session, {
          type: 'ai-delta',
          content: event.delta.text
        });
      }
    }
    
    // Update context
    session.claudeContext.push(
      { role: 'user', content: prompt },
      { role: 'assistant', content: fullResponse }
    );
  }
  
  private broadcast(
    session: CollaborativeSession,
    excludeUserId: string,
    message: any
  ) {
    session.participants.forEach((participant, userId) => {
      if (userId !== excludeUserId && participant.ws) {
        participant.ws.send(JSON.stringify(message));
      }
    });
  }
  
  private broadcastToAll(session: CollaborativeSession, message: any) {
    session.participants.forEach((participant) => {
      if (participant.ws) {
        participant.ws.send(JSON.stringify(message));
      }
    });
  }
}

4. Live Data Analysis Pattern

// Real-time data analysis with Claude
class LiveDataAnalyzer {
  private dataStream: EventSource;
  private analysisQueue: any[] = [];
  private batchSize = 10;
  private analysisInterval = 5000; // 5 seconds
  private anthropic: Anthropic;
  
  constructor(dataStreamUrl: string) {
    this.anthropic = new Anthropic();
    this.dataStream = new EventSource(dataStreamUrl);
    this.setupDataStream();
    this.startAnalysisLoop();
  }
  
  private setupDataStream() {
    this.dataStream.onmessage = (event) => {
      const data = JSON.parse(event.data);
      this.analysisQueue.push(data);
      
      // Trigger immediate analysis for critical data
      if (data.priority === 'critical') {
        this.analyzeImmediate(data);
      }
    };
  }
  
  private async startAnalysisLoop() {
    setInterval(async () => {
      if (this.analysisQueue.length >= this.batchSize) {
        const batch = this.analysisQueue.splice(0, this.batchSize);
        await this.analyzeBatch(batch);
      }
    }, this.analysisInterval);
  }
  
  private async analyzeBatch(batch: any[]) {
    const prompt = `Analyze the following real-time data batch and identify patterns, anomalies, and insights:
    
    ${JSON.stringify(batch, null, 2)}
    
    Provide:
    1. Key patterns observed
    2. Any anomalies detected
    3. Recommended actions
    4. Predictions for next data points`;
    
    try {
      const response = await this.anthropic.messages.create({
        model: 'claude-3-sonnet-20240229',
        messages: [{ role: 'user', content: prompt }],
        max_tokens: 500
      });
      
      this.broadcastAnalysis({
        timestamp: new Date().toISOString(),
        batchSize: batch.length,
        analysis: response.content[0].text,
        dataRange: {
          start: batch[0].timestamp,
          end: batch[batch.length - 1].timestamp
        }
      });
    } catch (error) {
      console.error('Analysis error:', error);
    }
  }
  
  private async analyzeImmediate(data: any) {
    const prompt = `CRITICAL DATA ALERT: Analyze this data point immediately:
    ${JSON.stringify(data, null, 2)}
    
    Provide immediate assessment and recommended actions.`;
    
    const response = await this.anthropic.messages.create({
      model: 'claude-3-opus-20240229', // Use faster model for critical
      messages: [{ role: 'user', content: prompt }],
      max_tokens: 200
    });
    
    this.broadcastAlert({
      type: 'critical',
      data,
      analysis: response.content[0].text,
      timestamp: new Date().toISOString()
    });
  }
  
  private broadcastAnalysis(analysis: any) {
    // Broadcast to connected clients via WebSocket/SSE
    console.log('Broadcasting analysis:', analysis);
  }
  
  private broadcastAlert(alert: any) {
    // Broadcast critical alerts
    console.log('Broadcasting alert:', alert);
  }
}

5. Streaming Code Generation

// Real-time code generation with execution feedback
class StreamingCodeGenerator {
  private anthropic: Anthropic;
  private executionEnvironment: CodeExecutor;
  
  async generateAndExecute(
    specification: string,
    onUpdate: (update: any) => void
  ) {
    // Stream code generation
    const stream = await this.anthropic.messages.create({
      model: 'claude-3-sonnet-20240229',
      messages: [{
        role: 'user',
        content: `Generate TypeScript code for: ${specification}`
      }],
      stream: true
    });
    
    let codeBuffer = '';
    let inCodeBlock = false;
    
    for await (const event of stream) {
      if (event.type === 'content_block_delta') {
        const text = event.delta.text;
        codeBuffer += text;
        
        // Detect code blocks
        if (text.includes('```typescript')) {
          inCodeBlock = true;
          onUpdate({ type: 'code_start' });
        } else if (text.includes('```') && inCodeBlock) {
          inCodeBlock = false;
          
          // Extract and execute code
          const code = this.extractCode(codeBuffer);
          onUpdate({ type: 'code_complete', code });
          
          // Execute in sandbox
          const result = await this.executionEnvironment.execute(code);
          onUpdate({ type: 'execution_result', result });
          
          codeBuffer = '';
        } else if (inCodeBlock) {
          onUpdate({ type: 'code_delta', delta: text });
        } else {
          onUpdate({ type: 'text_delta', delta: text });
        }
      }
    }
  }
  
  private extractCode(buffer: string): string {
    const match = buffer.match(/```typescript\n([\s\S]*?)\n```/);
    return match ? match[1] : '';
  }
}

Best Practices

1. Connection Management

• Implement automatic reconnection for WebSockets
• Use exponential backoff for retry attempts
• Handle connection state in UI
• Provide offline functionality

2. Performance Optimization

• Batch updates to reduce render cycles
• Implement virtual scrolling for long streams
• Use Web Workers for heavy processing
• Throttle/debounce user inputs

3. Error Handling

class StreamErrorHandler {
  private errorQueue: Error[] = [];
  private maxRetries = 3;
  
  async handleStreamError(
    error: Error,
    retryFn: () => Promise<any>,
    attempt = 1
  ): Promise<any> {
    this.errorQueue.push(error);
    
    if (error.message.includes('rate_limit')) {
      // Handle rate limiting
      const delay = this.calculateRateLimitDelay(error);
      await this.delay(delay);
      return retryFn();
    } else if (error.message.includes('timeout')) {
      // Handle timeouts with exponential backoff
      if (attempt <= this.maxRetries) {
        const delay = Math.pow(2, attempt) * 1000;
        await this.delay(delay);
        return this.handleStreamError(error, retryFn, attempt + 1);
      }
    }
    
    throw error;
  }
  
  private calculateRateLimitDelay(error: Error): number {
    // Extract retry-after header or use default
    return 60000; // 1 minute default
  }
  
  private delay(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

4. Security Considerations

• Validate all incoming messages
• Implement authentication for WebSocket connections
• Use HTTPS/WSS in production
• Rate limit client requests
• Sanitize AI responses before rendering

5. Testing Strategies

// Mock streaming for tests
class MockClaudeStream {
  async *streamResponse(response: string, chunkSize = 10) {
    for (let i = 0; i < response.length; i += chunkSize) {
      yield {
        type: 'content_block_delta',
        delta: {
          text: response.slice(i, i + chunkSize)
        }
      };
      
      // Simulate network delay
      await new Promise(resolve => setTimeout(resolve, 50));
    }
    
    yield { type: 'message_stop' };
  }
}

UI/UX Patterns

Typewriter Effect

class TypewriterRenderer {
  private container: HTMLElement;
  private speed = 30; // ms per character
  
  async render(text: string) {
    this.container.innerHTML = '';
    
    for (const char of text) {
      this.container.innerHTML += char;
      await new Promise(resolve => setTimeout(resolve, this.speed));
    }
  }
  
  async renderStream(stream: AsyncIterable<any>) {
    for await (const event of stream) {
      if (event.type === 'content_block_delta') {
        for (const char of event.delta.text) {
          this.container.innerHTML += char;
          await new Promise(resolve => setTimeout(resolve, this.speed));
        }
      }
    }
  }
}

Progress Indicators

interface StreamProgress {
  tokensGenerated: number;
  estimatedTotal: number;
  timeElapsed: number;
}
 
class StreamProgressTracker {
  private startTime: number;
  private tokensGenerated = 0;
  
  start() {
    this.startTime = Date.now();
    this.tokensGenerated = 0;
  }
  
  update(tokens: number): StreamProgress {
    this.tokensGenerated += tokens;
    const timeElapsed = Date.now() - this.startTime;
    const tokensPerSecond = this.tokensGenerated / (timeElapsed / 1000);
    
    return {
      tokensGenerated: this.tokensGenerated,
      estimatedTotal: Math.round(this.tokensGenerated * 1.2), // Estimate
      timeElapsed
    };
  }
}

Conclusion

Real-time data streaming with Claude Code opens up possibilities for interactive AI applications. Whether using SSE for simple streaming or WebSockets for complex bidirectional communication, these patterns provide a foundation for building responsive, real-time AI experiences.

See Also

• Streaming Patterns in TypeScript SDK
• Performance Optimization
• Real-Time Collaboration Patterns
• Error Recovery Patterns

External Resources

• Anthropic Streaming Documentation
• WebSocket API MDN
• Server-Sent Events MDN
• Model Context Protocol