Server-Sent Events (SSE) Implementation Guide

This guide provides a comprehensive tutorial on implementing Server-Sent Events (SSE) for streaming AI responses, with a focus on Claude Code integration.

Overview

Server-Sent Events (SSE) is the preferred technology for streaming AI responses from Claude and other LLMs. It provides:

• Unidirectional server-to-client communication
• Automatic reconnection handling
• Simple HTTP-based protocol
• Wide browser support
• Lower overhead than WebSockets for one-way streaming

Basic SSE Implementation

Server Setup (Node.js/Express)

import express from 'express';
import { Anthropic } from '@anthropic-ai/sdk';
 
const app = express();
const anthropic = new Anthropic();
 
app.post('/api/stream', async (req, res) => {
  // Configure SSE headers
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no', // Disable Nginx buffering
    'Access-Control-Allow-Origin': '*' // Configure CORS as needed
  });
 
  try {
    const stream = await anthropic.messages.create({
      model: 'claude-opus-4-20250514',
      messages: req.body.messages,
      max_tokens: req.body.max_tokens || 1024,
      stream: true
    });
 
    for await (const event of stream) {
      // Format as SSE event
      res.write(`event: ${event.type}\n`);
      res.write(`data: ${JSON.stringify(event)}\n\n`);
      
      // Ensure immediate delivery
      if (res.flush) res.flush();
    }
 
    // Send completion event
    res.write('event: complete\n');
    res.write('data: {"status": "success"}\n\n');
    res.end();
  } catch (error) {
    res.write('event: error\n');
    res.write(`data: ${JSON.stringify({ error: error.message })}\n\n`);
    res.end();
  }
});
 
app.listen(3000, () => {
  console.log('SSE server running on port 3000');
});

Client Implementation

class SSEClient {
  private eventSource: EventSource | null = null;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 5;
  private reconnectDelay = 1000;
 
  connect(url: string, options: {
    onMessage?: (data: any) => void;
    onError?: (error: any) => void;
    onComplete?: () => void;
  }) {
    this.eventSource = new EventSource(url);
 
    this.eventSource.addEventListener('message', (event) => {
      try {
        const data = JSON.parse(event.data);
        options.onMessage?.(data);
      } catch (error) {
        console.error('Failed to parse SSE data:', error);
      }
    });
 
    this.eventSource.addEventListener('content_block_delta', (event) => {
      const data = JSON.parse(event.data);
      options.onMessage?.({
        type: 'delta',
        content: data.delta.text
      });
    });
 
    this.eventSource.addEventListener('complete', () => {
      options.onComplete?.();
      this.disconnect();
    });
 
    this.eventSource.addEventListener('error', (event) => {
      if (this.eventSource?.readyState === EventSource.CLOSED) {
        this.handleReconnect(url, options);
      } else {
        options.onError?.(event);
      }
    });
 
    this.eventSource.addEventListener('open', () => {
      this.reconnectAttempts = 0;
      console.log('SSE connection established');
    });
  }
 
  private handleReconnect(url: string, options: any) {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      options.onError?.(new Error('Max reconnection attempts reached'));
      return;
    }
 
    this.reconnectAttempts++;
    const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
 
    console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
    
    setTimeout(() => {
      this.connect(url, options);
    }, delay);
  }
 
  disconnect() {
    if (this.eventSource) {
      this.eventSource.close();
      this.eventSource = null;
    }
  }
}

Advanced SSE Patterns

1. Buffered Streaming for Smooth UI

class BufferedSSERenderer {
  private buffer: string[] = [];
  private flushInterval: number;
  private flushTimer: NodeJS.Timeout | null = null;
  private onFlush: (content: string) => void;
 
  constructor(options: {
    flushInterval?: number;
    onFlush: (content: string) => void;
  }) {
    this.flushInterval = options.flushInterval || 50;
    this.onFlush = options.onFlush;
  }
 
  addChunk(chunk: string) {
    this.buffer.push(chunk);
    
    if (!this.flushTimer) {
      this.flushTimer = setTimeout(() => this.flush(), this.flushInterval);
    }
  }
 
  private flush() {
    if (this.buffer.length > 0) {
      const content = this.buffer.join('');
      this.buffer = [];
      this.onFlush(content);
    }
    
    this.flushTimer = null;
  }
 
  forceFlush() {
    if (this.flushTimer) {
      clearTimeout(this.flushTimer);
      this.flushTimer = null;
    }
    this.flush();
  }
}
 
// Usage
const renderer = new BufferedSSERenderer({
  flushInterval: 30,
  onFlush: (content) => {
    document.getElementById('output').innerHTML += content;
  }
});
 
sseClient.connect('/api/stream', {
  onMessage: (data) => {
    if (data.type === 'delta') {
      renderer.addChunk(data.content);
    }
  },
  onComplete: () => {
    renderer.forceFlush();
  }
});

2. Progress Tracking

interface StreamMetrics {
  startTime: number;
  bytesReceived: number;
  chunksReceived: number;
  tokensGenerated: number;
  averageChunkSize: number;
  throughput: number; // bytes/second
}
 
class SSEProgressTracker {
  private metrics: StreamMetrics = {
    startTime: 0,
    bytesReceived: 0,
    chunksReceived: 0,
    tokensGenerated: 0,
    averageChunkSize: 0,
    throughput: 0
  };
 
  private onProgress?: (metrics: StreamMetrics) => void;
 
  constructor(onProgress?: (metrics: StreamMetrics) => void) {
    this.onProgress = onProgress;
  }
 
  start() {
    this.metrics = {
      startTime: Date.now(),
      bytesReceived: 0,
      chunksReceived: 0,
      tokensGenerated: 0,
      averageChunkSize: 0,
      throughput: 0
    };
  }
 
  updateChunk(chunk: string) {
    const chunkSize = new TextEncoder().encode(chunk).length;
    this.metrics.bytesReceived += chunkSize;
    this.metrics.chunksReceived++;
    this.metrics.tokensGenerated += this.estimateTokens(chunk);
    
    const elapsed = (Date.now() - this.metrics.startTime) / 1000;
    this.metrics.throughput = this.metrics.bytesReceived / elapsed;
    this.metrics.averageChunkSize = 
      this.metrics.bytesReceived / this.metrics.chunksReceived;
    
    this.onProgress?.(this.metrics);
  }
 
  private estimateTokens(text: string): number {
    // Rough estimation: 1 token ≈ 4 characters
    return Math.ceil(text.length / 4);
  }
 
  getMetrics(): StreamMetrics {
    return { ...this.metrics };
  }
}

3. Error Recovery and Resilience

class ResilientSSEClient {
  private eventSource: EventSource | null = null;
  private connectionState: 'disconnected' | 'connecting' | 'connected' = 'disconnected';
  private errorHistory: Array<{ timestamp: number; error: any }> = [];
  private maxErrorHistory = 10;
 
  async connectWithFallback(primaryUrl: string, fallbackUrls: string[], options: any) {
    const urls = [primaryUrl, ...fallbackUrls];
    
    for (const url of urls) {
      try {
        await this.attemptConnection(url, options);
        return; // Success
      } catch (error) {
        console.error(`Failed to connect to ${url}:`, error);
        this.recordError(error);
      }
    }
    
    throw new Error('All connection attempts failed');
  }
 
  private attemptConnection(url: string, options: any): Promise<void> {
    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        this.disconnect();
        reject(new Error('Connection timeout'));
      }, 10000); // 10 second timeout
 
      this.connectionState = 'connecting';
      this.eventSource = new EventSource(url);
 
      this.eventSource.addEventListener('open', () => {
        clearTimeout(timeout);
        this.connectionState = 'connected';
        this.errorHistory = []; // Clear errors on successful connection
        resolve();
      });
 
      this.eventSource.addEventListener('error', (event) => {
        clearTimeout(timeout);
        this.connectionState = 'disconnected';
        reject(event);
      });
 
      // Set up other event handlers
      Object.entries(options).forEach(([event, handler]) => {
        if (event.startsWith('on') && typeof handler === 'function') {
          const eventName = event.substring(2).toLowerCase();
          this.eventSource!.addEventListener(eventName, handler);
        }
      });
    });
  }
 
  private recordError(error: any) {
    this.errorHistory.push({
      timestamp: Date.now(),
      error
    });
    
    if (this.errorHistory.length > this.maxErrorHistory) {
      this.errorHistory.shift();
    }
  }
 
  getConnectionState() {
    return this.connectionState;
  }
 
  getErrorRate(windowMs: number = 60000): number {
    const now = Date.now();
    const recentErrors = this.errorHistory.filter(
      e => now - e.timestamp <= windowMs
    );
    return recentErrors.length;
  }
 
  disconnect() {
    if (this.eventSource) {
      this.eventSource.close();
      this.eventSource = null;
      this.connectionState = 'disconnected';
    }
  }
}

SSE with Authentication

// Server-side authentication
app.post('/api/authenticated-stream', authenticate, async (req, res) => {
  const userId = req.user.id;
  const streamId = crypto.randomUUID();
  
  // Store active streams for user management
  activeStreams.set(streamId, { userId, startTime: Date.now() });
  
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Stream-ID': streamId
  });
 
  // Set up cleanup on disconnect
  req.on('close', () => {
    activeStreams.delete(streamId);
    console.log(`Stream ${streamId} closed for user ${userId}`);
  });
 
  // Stream implementation...
});
 
// Client-side with auth token
class AuthenticatedSSEClient {
  connect(url: string, token: string, options: any) {
    // EventSource doesn't support custom headers, so pass token in URL
    const authUrl = `${url}?token=${encodeURIComponent(token)}`;
    
    // Or use a polyfill that supports headers
    const eventSource = new EventSourcePolyfill(url, {
      headers: {
        'Authorization': `Bearer ${token}`
      }
    });
    
    // Set up handlers...
  }
}

Testing SSE Implementations

// Mock SSE server for testing
class MockSSEServer {
  private clients: Set<MockSSEClient> = new Set();
 
  async simulateStream(messages: any[], delayMs: number = 50) {
    for (const message of messages) {
      this.broadcast(message);
      await new Promise(resolve => setTimeout(resolve, delayMs));
    }
    
    this.broadcast({ type: 'complete' });
  }
 
  private broadcast(message: any) {
    this.clients.forEach(client => {
      client.receive(message);
    });
  }
 
  connect(client: MockSSEClient) {
    this.clients.add(client);
    return () => this.clients.delete(client);
  }
}
 
// Test example
describe('SSE Streaming', () => {
  it('should handle streaming response', async () => {
    const received: any[] = [];
    const mockServer = new MockSSEServer();
    
    const client = new MockSSEClient({
      onMessage: (data) => received.push(data)
    });
    
    mockServer.connect(client);
    
    await mockServer.simulateStream([
      { type: 'delta', content: 'Hello' },
      { type: 'delta', content: ' World' }
    ]);
    
    expect(received).toHaveLength(3); // 2 deltas + complete
    expect(received[0].content).toBe('Hello');
    expect(received[1].content).toBe(' World');
    expect(received[2].type).toBe('complete');
  });
});

Best Practices

1. Connection Management

• Implement automatic reconnection with exponential backoff
• Monitor connection health and provide user feedback
• Clean up connections properly on page unload
• Handle browser tab suspension/resumption

2. Performance

• Use compression for large payloads
• Implement client-side buffering for smooth rendering
• Monitor and limit memory usage for long streams
• Consider pagination for very long responses

3. Error Handling

• Provide clear error messages to users
• Implement fallback mechanisms
• Log errors for debugging
• Handle rate limiting gracefully

4. Security

• Always use HTTPS in production
• Implement proper authentication
• Validate and sanitize streamed content
• Set appropriate CORS headers

Common Issues and Solutions

Issue: Nginx Buffering

# Disable buffering for SSE
location /api/stream {
    proxy_pass http://backend;
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header X-Accel-Buffering no;
}

Issue: Connection Timeout

// Implement heartbeat to keep connection alive
setInterval(() => {
  res.write(':heartbeat\n\n');
}, 30000); // Every 30 seconds

Issue: Memory Leaks

// Properly clean up event listeners
window.addEventListener('beforeunload', () => {
  sseClient.disconnect();
});

🔗 Related Resources

• Streaming Deep Dive
• WebSocket Patterns
• Streaming Performance
• Claude Code Documentation

📚 External References

• MDN: Server-Sent Events
• EventSource Specification
• Anthropic Streaming Docs