Streaming Api Patterns

Streaming API Patterns

Overview

When to use this skill:

• Streaming LLM responses (ChatGPT-style interfaces)
• Real-time notifications and updates
• Live data feeds (stock prices, analytics)
• Chat applications
• Progress updates for long-running tasks
• Collaborative editing features

Core Technologies

1. Server-Sent Events (SSE)

Best for: Server-to-client streaming (LLM responses, notifications)

// Next.js Route Handler
export async function GET(req: Request) {
  const encoder = new TextEncoder()

  const stream = new ReadableStream({
    async start(controller) {
      // Send data
      controller.enqueue(encoder.encode('data: Hello\n\n'))

      // Keep connection alive
      const interval = setInterval(() => {
        controller.enqueue(encoder.encode(': keepalive\n\n'))
      }, 30000)

      // Cleanup
      req.signal.addEventListener('abort', () => {
        clearInterval(interval)
        controller.close()
      })
    }
  })

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    }
  })
}

// Client
const eventSource = new EventSource('/api/stream')
eventSource.onmessage = (event) => {
  console.log(event.data)
}

2. WebSockets

Best for: Bidirectional real-time communication (chat, collaboration)

// WebSocket Server (Next.js with ws)
import { WebSocketServer } from 'ws'

const wss = new WebSocketServer({ port: 8080 })

wss.on('connection', (ws) => {
  ws.on('message', (data) => {
    // Broadcast to all clients
    wss.clients.forEach((client) => {
      if (client.readyState === WebSocket.OPEN) {
        client.send(data)
      }
    })
  })
})

// Client
const ws = new WebSocket('ws://localhost:8080')
ws.onmessage = (event) => console.log(event.data)
ws.send(JSON.stringify({ type: 'message', text: 'Hello' }))

3. ReadableStream API

Best for: Processing large data streams with backpressure

async function* generateData() {
  for (let i = 0; i < 1000; i++) {
    await new Promise(resolve => setTimeout(resolve, 100))
    yield "data-" + i
  }
}

const stream = new ReadableStream({
  async start(controller) {
    for await (const chunk of generateData()) {
      controller.enqueue(new TextEncoder().encode(chunk + '\n'))
    }
    controller.close()
  }
})

LLM Streaming Pattern

// Server
import OpenAI from 'openai'

const openai = new OpenAI()

export async function POST(req: Request) {
  const { messages } = await req.json()

  const stream = await openai.chat.completions.create({
    model: 'gpt-5.2',
    messages,
    stream: true
  })

  const encoder = new TextEncoder()

  return new Response(
    new ReadableStream({
      async start(controller) {
        for await (const chunk of stream) {
          const content = chunk.choices[0]?.delta?.content
          if (content) {
            controller.enqueue(encoder.encode("data: " + JSON.stringify({ content }) + "\n\n"))
          }
        }
        controller.enqueue(encoder.encode('data: [DONE]\n\n'))
        controller.close()
      }
    }),
    {
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache'
      }
    }
  )
}

// Client
async function streamChat(messages) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ messages })
  })

  const reader = response.body.getReader()
  const decoder = new TextDecoder()

  while (true) {
    const { done, value } = await reader.read()
    if (done) break

    const chunk = decoder.decode(value)
    const lines = chunk.split('\n')

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = line.slice(6)
        if (data === '[DONE]') return

        const json = JSON.parse(data)
        console.log(json.content) // Stream token
      }
    }
  }
}

Reconnection Strategy

class ReconnectingEventSource {
  private eventSource: EventSource | null = null
  private reconnectDelay = 1000
  private maxReconnectDelay = 30000

  constructor(private url: string, private onMessage: (data: string) => void) {
    this.connect()
  }

  private connect() {
    this.eventSource = new EventSource(this.url)

    this.eventSource.onmessage = (event) => {
      this.reconnectDelay = 1000 // Reset on success
      this.onMessage(event.data)
    }

    this.eventSource.onerror = () => {
      this.eventSource?.close()

      // Exponential backoff
      setTimeout(() => this.connect(), this.reconnectDelay)
      this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxReconnectDelay)
    }
  }

  close() {
    this.eventSource?.close()
  }
}

Python Async Generator Cleanup ( Best Practice)

CRITICAL: Async generators can leak resources if not properly cleaned up. Python 3.10+ provides aclosing() from contextlib to guarantee cleanup.

The Problem

# ❌ DANGEROUS: Generator not closed if exception occurs mid-iteration
async def stream_analysis():
    async for chunk in external_api_stream():  # What if exception here?
        yield process(chunk)  # Generator may be garbage collected without cleanup

# ❌ ALSO DANGEROUS: Using .aclose() manually is error-prone
gen = stream_analysis()
try:
    async for chunk in gen:
        process(chunk)
finally:
    await gen.aclose()  # Easy to forget, verbose

The Solution: aclosing()

from contextlib import aclosing

# ✅ CORRECT: aclosing() guarantees cleanup
async def stream_analysis():
    async with aclosing(external_api_stream()) as stream:
        async for chunk in stream:
            yield process(chunk)

# ✅ CORRECT: Using aclosing() at consumption site
async def consume_stream():
    async with aclosing(stream_analysis()) as gen:
        async for chunk in gen:
            handle(chunk)

Real-World Pattern: LLM Streaming

from contextlib import aclosing
from langchain_core.runnables import RunnableConfig

async def stream_llm_response(prompt: str, config: RunnableConfig | None = None):
    """Stream LLM tokens with guaranteed cleanup."""
    async with aclosing(llm.astream(prompt, config=config)) as stream:
        async for chunk in stream:
            yield chunk.content

# Consumption with proper cleanup
async def generate_response(user_input: str):
    result_chunks = []
    async with aclosing(stream_llm_response(user_input)) as response:
        async for token in response:
            result_chunks.append(token)
            yield token  # Stream to client

    # Post-processing after stream completes
    full_response = "".join(result_chunks)
    await log_response(full_response)

Database Connection Pattern

from contextlib import aclosing
from typing import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession

async def stream_large_query(
    session: AsyncSession,
    batch_size: int = 1000
) -> AsyncIterator[Row]:
    """Stream large query results with automatic connection cleanup."""
    result = await session.execute(
        select(Model).execution_options(stream_results=True)
    )

    async with aclosing(result.scalars()) as stream:
        async for row in stream:
            yield row

When to Use aclosing()

Anti-Patterns to Avoid

# ❌ NEVER: Consuming without aclosing
async for chunk in stream_analysis():
    process(chunk)

# ❌ NEVER: Manual try/finally (verbose, error-prone)
gen = stream_analysis()
try:
    async for chunk in gen:
        process(chunk)
finally:
    await gen.aclose()

# ❌ NEVER: Assuming GC will handle cleanup
gen = stream_analysis()
# ... later gen goes out of scope without close

Testing Async Generators

import pytest
from contextlib import aclosing

@pytest.mark.asyncio
async def test_stream_cleanup_on_error():
    """Test that cleanup happens even when exception raised."""
    cleanup_called = False

    async def stream_with_cleanup():
        nonlocal cleanup_called
        try:
            yield "data"
            yield "more"
        finally:
            cleanup_called = True

    with pytest.raises(ValueError):
        async with aclosing(stream_with_cleanup()) as gen:
            async for chunk in gen:
                raise ValueError("simulated error")

    assert cleanup_called, "Cleanup must run even on exception"

Best Practices

SSE

• ✅ Use for one-way server-to-client streaming
• ✅ Implement automatic reconnection
• ✅ Send keepalive messages every 30s
• ✅ Handle browser connection limits (6 per domain)
• ✅ Use HTTP/2 for better performance

WebSockets

• ✅ Use for bidirectional real-time communication
• ✅ Implement heartbeat/ping-pong
• ✅ Handle reconnection with exponential backoff
• ✅ Validate and sanitize messages
• ✅ Implement message queuing for offline periods

Backpressure

• ✅ Use ReadableStream with proper flow control
• ✅ Monitor buffer sizes
• ✅ Pause production when consumer is slow
• ✅ Implement timeouts for slow consumers

Performance

• ✅ Compress data (gzip/brotli)
• ✅ Batch small messages
• ✅ Use binary formats (MessagePack, Protobuf) for large data
• ✅ Implement client-side buffering
• ✅ Monitor connection count and resource usage

Resources

• Server-Sent Events Specification
• WebSocket Protocol
• Streams API
• Vercel AI SDK

Related Skills

• llm-streaming - LLM-specific streaming patterns for token-by-token responses
• api-design-framework - REST API design patterns for streaming endpoints
• caching-strategies - Cache invalidation patterns for real-time data updates
• edge-computing-patterns - Edge function streaming for low-latency delivery

Key Decisions

Capability Details

sse

Keywords: sse, server-sent events, event stream, one-way stream Solves:

• How do I implement SSE?
• Stream data from server to client
• Real-time notifications

sse-protocol

Keywords: sse protocol, event format, event types, sse headers Solves:

• SSE protocol fundamentals
• Event format and types
• SSE HTTP headers

sse-buffering

Keywords: event buffering, sse race condition, late subscriber, buffer events Solves:

• How do I buffer SSE events?
• Fix SSE race condition
• Handle late-joining subscribers

sse-reconnection

Keywords: sse reconnection, reconnect, last-event-id, retry, exponential backoff Solves:

• How do I handle SSE reconnection?
• Implement automatic reconnection
• Resume from Last-Event-ID

orchestkit-sse

Keywords: orchestkit sse, event broadcaster, workflow events, analysis progress Solves:

• How does OrchestKit SSE work?
• EventBroadcaster implementation
• Real-world SSE example

websocket

Keywords: websocket, ws, bidirectional, real-time chat, socket Solves:

• How do I set up WebSocket server?
• Build a chat application
• Bidirectional real-time communication

llm-streaming

Keywords: llm stream, chatgpt stream, ai stream, token stream, openai stream Solves:

• How do I stream LLM responses?
• ChatGPT-style streaming interface
• Stream tokens as they arrive

backpressure

Keywords: backpressure, flow control, buffer, readable stream, transform stream Solves:

• Handle slow consumers
• Implement backpressure
• Stream large files efficiently

reconnection

Keywords: reconnect, connection lost, retry, resilient, heartbeat Solves:

• Handle connection drops
• Implement automatic reconnection
• Keep-alive and heartbeat

Streaming API Implementation Checklist

Server-Sent Events (SSE)

Setup

• [ ] Set correct Content-Type: text/event-stream
• [ ] Disable caching with Cache-Control: no-cache
• [ ] Set Connection: keep-alive
• [ ] Disable nginx buffering with X-Accel-Buffering: no

Data Format

• [ ] Prefix messages with data:
• [ ] End messages with \n\n
• [ ] Send keepalive comments every 30s (: keepalive\n\n)
• [ ] Use JSON for structured data
• [ ] Send [DONE] marker when complete

Error Handling

• [ ] Handle client disconnection (req.signal.aborted)
• [ ] Catch and send errors as SSE messages
• [ ] Close stream properly in finally block
• [ ] Implement server-side reconnection logic

Client

• [ ] Use EventSource API
• [ ] Handle automatic reconnection
• [ ] Implement error handling
• [ ] Close connection when done
• [ ] Handle browser connection limits (max 6 per domain)

WebSockets

Server

• [ ] Implement ping/pong heartbeat
• [ ] Handle connection upgrades
• [ ] Validate incoming messages
• [ ] Broadcast to multiple clients efficiently
• [ ] Track active connections

Client

• [ ] Implement reconnection with exponential backoff
• [ ] Queue messages during disconnection
• [ ] Handle connection states (connecting, open, closing, closed)
• [ ] Implement timeout for slow connections

Security

• [ ] Validate origin headers
• [ ] Authenticate connections
• [ ] Rate limit messages per client
• [ ] Sanitize message content
• [ ] Implement message size limits

Backpressure & Performance

• [ ] Monitor buffer sizes
• [ ] Implement flow control
• [ ] Pause stream when consumer is slow
• [ ] Use chunking for large data
• [ ] Compress data (gzip/brotli)
• [ ] Batch small messages
• [ ] Set appropriate timeouts

LLM Streaming

• [ ] Stream tokens as they arrive
• [ ] Handle partial Unicode characters
• [ ] Implement stop generation
• [ ] Show typing indicator
• [ ] Handle stream interruption
• [ ] Measure time-to-first-token

Testing

• [ ] Test with slow networks
• [ ] Test reconnection scenarios
• [ ] Test with multiple concurrent clients
• [ ] Load test with expected traffic
• [ ] Test error cases (server crash, network issues)
• [ ] Verify memory doesn't leak over long connections

OrchestKit SSE Implementation

Real-world Server-Sent Events implementation from OrchestKit, documenting the EventBroadcaster service, SSE endpoint handler, event buffering, and workflow integration.

Project Context

OrchestKit: Multi-agent analysis workflow with real-time progress streaming

Tech Stack:

• Backend: FastAPI + sse-starlette 3.0.3
• Frontend: React 19 with native EventSource API
• Event Bus: Custom EventBroadcaster (in-memory pub/sub)

SSE Endpoint: GET /api/v1/analyze/{analysis_id}/stream

Architecture Overview

┌─────────────────┐
│ Workflow Tasks  │ (extraction, analysis, chunking, etc.)
└────────┬────────┘
         │ publish events
         ↓
┌─────────────────────────────────┐
│ EventBroadcaster (Pub/Sub)      │
│ - In-memory queues              │
│ - Event buffering (deque)       │
│ - Channel-based routing         │
└────────┬────────────────────────┘
         │ subscribe
         ↓
┌─────────────────┐
│ SSE Handler     │ (sse_handler.py)
└────────┬────────┘
         │ stream over HTTP
         ↓
┌─────────────────┐
│ Frontend Client │ (EventSource)
└─────────────────┘

EventBroadcaster Service

Location: backend/app/shared/services/messaging/broadcaster.py

Core Implementation

"""Event broadcaster service for pub/sub messaging.

Provides in-memory pub/sub functionality for SSE events using asyncio.Queue.
Channels are keyed by string identifiers (e.g., "workflow:{analysis_id}").

Issue #SSE-RACE: Added event buffering to solve the race condition where
events are published before subscribers connect. Recent events are buffered
per channel and replayed to new subscribers.
"""

import asyncio
from collections import defaultdict, deque
from collections.abc import AsyncIterator
from contextlib import suppress
from datetime import UTC, datetime, timedelta

from app.core.logging import get_logger

logger = get_logger(__name__)

# Buffer configuration
MAX_BUFFER_SIZE = 100  # Max events to buffer per channel
BUFFER_TTL_SECONDS = 300  # 5 minutes - events older than this are dropped


class EventBroadcaster:
    """In-memory pub/sub broadcaster for SSE events with event buffering.

    Issue #SSE-RACE: Implements event buffering to solve race condition where
    workflow starts emitting events before frontend SSE connection is established.
    Recent events are buffered per channel and replayed to new subscribers.
    """

    def __init__(self) -> None:
        """Initialize event broadcaster with empty channels and buffers."""
        self._channels: dict[str, list[asyncio.Queue]] = defaultdict(list)
        self._buffers: dict[str, deque[tuple[datetime, dict]]] = defaultdict(
            lambda: deque(maxlen=MAX_BUFFER_SIZE)
        )
        self._lock = asyncio.Lock()

    async def publish(self, channel: str, message: dict) -> None:
        """Publish message to all subscribers of a channel.

        Issue #SSE-RACE: Events are now buffered for late-joining subscribers.
        Even if no subscribers exist, events are stored in the buffer.
        """
        now = datetime.now(UTC)

        async with self._lock:
            queues = self._channels.get(channel, [])

            # Always buffer the event (even if no subscribers)
            # This solves the race condition where workflow starts before SSE connects
            self._buffers[channel].append((now, message))

            # Clean up old events from buffer (older than TTL)
            cutoff = now - timedelta(seconds=BUFFER_TTL_SECONDS)
            while self._buffers[channel] and self._buffers[channel][0][0] < cutoff:
                self._buffers[channel].popleft()

        if not queues:
            logger.debug(
                "publish_buffered_no_subscribers",
                channel=channel,
                buffer_size=len(self._buffers[channel])
            )
            return

        # Broadcast to all subscribers
        for queue in queues:
            try:
                await queue.put(message)
            except (asyncio.CancelledError, RuntimeError) as e:
                logger.warning(
                    "publish_failed",
                    channel=channel,
                    error=str(e),
                    exc_info=True
                )

        logger.debug(
            "publish_success",
            channel=channel,
            subscribers=len(queues),
            buffer_size=len(self._buffers[channel])
        )

    async def subscribe(self, channel: str) -> AsyncIterator[dict]:
        """Subscribe to a channel and yield messages.

        Issue #SSE-RACE: New subscribers first receive all buffered events
        (events published before the subscriber connected), then receive
        live events as they are published.
        """
        queue: asyncio.Queue = asyncio.Queue()
        buffered_events: list[dict] = []

        # Add queue to channel subscribers and capture buffered events
        async with self._lock:
            self._channels[channel].append(queue)

            # Capture buffered events for replay (copy to avoid mutation during iteration)
            if channel in self._buffers:
                buffered_events = [event for _, event in self._buffers[channel]]

        logger.debug(
            "subscribe_created",
            channel=channel,
            buffered_events_count=len(buffered_events)
        )

        try:
            # First, replay all buffered events to catch up the subscriber
            for event in buffered_events:
                yield event

            if buffered_events:
                logger.debug(
                    "subscribe_buffer_replayed",
                    channel=channel,
                    events_replayed=len(buffered_events)
                )

            # Then, yield live events as they arrive
            while True:
                message = await queue.get()
                yield message
        except asyncio.CancelledError:
            logger.debug("subscribe_cancelled", channel=channel)
            raise
        finally:
            # Cleanup: remove queue from subscribers
            async with self._lock:
                if channel in self._channels:
                    with suppress(ValueError):
                        self._channels[channel].remove(queue)

                    # Clean up empty channels
                    if not self._channels[channel]:
                        del self._channels[channel]

            logger.debug("subscribe_cleaned", channel=channel)

    def get_subscriber_count(self, channel: str) -> int:
        """Get number of active subscribers for a channel."""
        return len(self._channels.get(channel, []))

    async def clear_buffer(self, channel: str) -> None:
        """Clear the event buffer for a channel.

        Call this when an analysis is complete to free memory.
        """
        async with self._lock:
            if channel in self._buffers:
                cleared_count = len(self._buffers[channel])
                del self._buffers[channel]
                logger.debug(
                    "buffer_cleared",
                    channel=channel,
                    events_cleared=cleared_count
                )


# Global broadcaster instance
broadcaster = EventBroadcaster()

Key Design Decisions

1. Event Buffering (Issue #SSE-RACE)

Problem: Workflow starts → Events published → Frontend connects (late) → Events lost

Solution: Buffer recent 100 events per channel with 5-minute TTL

• Late-joining subscribers receive buffered events first
• Then receive live events
• Bounded memory (deque with maxlen=100)
• Auto-cleanup (5-minute TTL)

2. Channel-Based Routing

Pattern: workflow:{analysis_id} → Each analysis has its own event channel

• Isolated event streams per analysis
• Multiple analyses can run concurrently
• No cross-contamination

3. Asyncio.Queue for Pub/Sub

Why not Redis Pub/Sub?

• In-memory is sufficient for single-instance deployment
• Lower latency (no network overhead)
• Simpler setup (no external dependency)

When to use Redis?

• Multi-instance backend (horizontal scaling)
• Event persistence across restarts
• Cross-service event bus

SSE Handler

Location: backend/app/api/v1/analysis/sse_handler.py

Core Implementation

"""SSE endpoint handler for analysis progress streaming."""

import asyncio
import json
import uuid
from collections.abc import AsyncIterator
from contextlib import aclosing
from datetime import UTC, datetime

from fastapi import Request
from sse_starlette.sse import EventSourceResponse

from app.core.logging import get_logger
from app.shared.services.messaging.broadcaster import broadcaster

logger = get_logger(__name__)


async def stream_analysis_progress(
    analysis_id: uuid.UUID,
    request: Request,
) -> EventSourceResponse:
    """Stream real-time analysis progress via Server-Sent Events (SSE).

    Establishes an SSE connection for the specified analysis and streams
    progress events as they occur during workflow execution. Events include
    stage updates, status changes, and completion notifications.

    Event Types:
        - progress: Stage status updates (running, complete)
        - error: Error notifications with error details
        - complete: Final workflow completion

    Example Server Events:

event: progress data: {"type": "progress", "stage": "extraction", "status": "running"}

event: progress data: {"type": "progress", "stage": "extraction", "status": "complete", "word_count": 5234}

event: complete data: {"type": "complete", "stage": "artifact_generation"}

    """
    channel = f"workflow:{analysis_id}"

    logger.info(
        "sse_connection_started",
        analysis_id=str(analysis_id),
        channel=channel
    )

    async def client_close_handler(message: dict) -> None:
        """Handle client disconnect with cleanup logging.

        Called automatically by sse-starlette 3.0.3 when client disconnects.
        """
        logger.info(
            "sse_client_disconnected",
            analysis_id=str(analysis_id),
            channel=channel,
            message=str(message)
        )

    async def event_generator() -> AsyncIterator[dict[str, str]]:
        """Generate SSE events from broadcaster subscription.

        Leverages sse-starlette 3.0.3 features:
        - Automatic disconnect detection (no manual checks needed)
        - Better exception propagation for clearer error messages
        - Improved cancellation handling with asyncio.CancelledError

        Uses aclosing() to ensure proper cleanup of the broadcaster subscription
        even if streaming is interrupted.
        """
        try:
            # Use aclosing() to ensure proper cleanup of async generator
            async with aclosing(broadcaster.subscribe(channel)) as subscription:
                async for event in subscription:
                    # Format event for SSE
                    event_type = str(event.get("type", "message"))
                    yield {
                        "event": event_type,
                        "data": json.dumps(event)
                    }

                    # Close connection on complete event
                    if event.get("type") == "complete":
                        logger.info(
                            "sse_complete_event_sent",
                            analysis_id=str(analysis_id),
                            channel=channel
                        )
                        break

        except asyncio.CancelledError:
            # sse-starlette 3.0.3 automatically cancels on client disconnect
            logger.info(
                "sse_connection_cancelled",
                analysis_id=str(analysis_id),
                channel=channel
            )
            raise
        except ConnectionError as e:
            logger.warning(
                "sse_connection_error",
                analysis_id=str(analysis_id),
                error=str(e)
            )
            # Send structured error event
            yield {
                "event": "error",
                "data": json.dumps({
                    "type": "error",
                    "error_type": "ConnectionError",
                    "analysis_id": str(analysis_id),
                    "error": "Connection error occurred",
                    "message": str(e),
                    "timestamp": datetime.now(UTC).isoformat()
                })
            }
        except Exception as e:
            logger.error(
                "sse_unexpected_error",
                analysis_id=str(analysis_id),
                error_type=type(e).__name__,
                error=str(e),
                exc_info=True
            )
            yield {
                "event": "error",
                "data": json.dumps({
                    "type": "error",
                    "error_type": type(e).__name__,
                    "analysis_id": str(analysis_id),
                    "error": "Unexpected error occurred",
                    "message": str(e),
                    "timestamp": datetime.now(UTC).isoformat()
                })
            }
        finally:
            logger.debug(
                "sse_event_generator_exiting",
                analysis_id=str(analysis_id),
                channel=channel
            )

    # Configure EventSourceResponse with sse-starlette 3.0.3 enhancements
    return EventSourceResponse(
        event_generator(),
        client_close_handler_callable=client_close_handler,
        send_timeout=30.0  # 30 seconds timeout for unresponsive clients
    )

Key Design Decisions

1. aclosing() for Cleanup

Why: Ensures broadcaster.subscribe() generator is properly closed even if SSE connection is interrupted

async with aclosing(broadcaster.subscribe(channel)) as subscription:
    async for event in subscription:
        yield event

2. Structured Error Events

Pattern: Send error as SSE event (not HTTP error)

yield {
    "event": "error",
    "data": json.dumps({
        "type": "error",
        "error_type": "ConnectionError",
        "message": str(e),
        "timestamp": datetime.now(UTC).isoformat()
    })
}

Benefit: Frontend can display error in UI without losing connection

3. Auto-close on Complete

Pattern: Break generator loop when complete event is received

if event.get("type") == "complete":
    logger.info("sse_complete_event_sent")
    break

Benefit: Clean connection closure, no lingering connections

Workflow Integration

Publishing Events from Workflow

Location: backend/app/domains/analysis/workflows/nodes/*.py

from app.shared.services.messaging.broadcaster import broadcaster

async def extraction_node(state: WorkflowState) -> dict[str, Any]:
    """Extract content from URL."""
    analysis_id = state["analysis_id"]
    channel = f"workflow:{analysis_id}"

    # Publish start event
    await broadcaster.publish(channel, {
        "type": "progress",
        "stage": "extraction",
        "status": "running",
        "timestamp": datetime.now(UTC).isoformat()
    })

    # Do extraction work
    try:
        content = await extract_content(state["url"])

        # Publish success event
        await broadcaster.publish(channel, {
            "type": "progress",
            "stage": "extraction",
            "status": "complete",
            "word_count": len(content.split()),
            "timestamp": datetime.now(UTC).isoformat()
        })

        return {"content": content}

    except Exception as e:
        # Publish error event
        await broadcaster.publish(channel, {
            "type": "error",
            "stage": "extraction",
            "error": str(e),
            "timestamp": datetime.now(UTC).isoformat()
        })
        raise

Final Complete Event

Location: backend/app/api/v1/analysis/workflow_runner.py

async def run_workflow_task(
    analysis_id: uuid.UUID,
    url: str,
    skill_level: str
) -> None:
    """Run analysis workflow and publish completion event."""
    channel = f"workflow:{analysis_id}"

    try:
        # Run workflow
        result = await workflow.ainvoke(initial_state)

        # Publish final complete event
        await broadcaster.publish(channel, {
            "type": "complete",
            "stage": "artifact_generation",
            "timestamp": datetime.now(UTC).isoformat()
        })

        # Clear buffer to free memory
        await broadcaster.clear_buffer(channel)

    except Exception as e:
        logger.error(
            "workflow_failed",
            analysis_id=str(analysis_id),
            error=str(e),
            exc_info=True
        )
        # Error event already published by failing node

Frontend Integration (React)

EventSource Hook

Location: frontend/src/features/analysis/hooks/useAnalysisProgress.ts

import { useEffect, useState } from 'react';

interface ProgressEvent {
  type: 'progress' | 'error' | 'complete';
  stage: string;
  status?: string;
  message?: string;
  timestamp: string;
}

export function useAnalysisProgress(analysisId: string) {
  const [events, setEvents] = useState<ProgressEvent[]>([]);
  const [isComplete, setIsComplete] = useState(false);
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    const eventSource = new EventSource(
      `http://localhost:8500/api/v1/analyze/${analysisId}/stream`
    );

    eventSource.addEventListener('progress', (e) => {
      const event = JSON.parse(e.data) as ProgressEvent;
      setEvents((prev) => [...prev, event]);
    });

    eventSource.addEventListener('complete', (e) => {
      const event = JSON.parse(e.data) as ProgressEvent;
      setEvents((prev) => [...prev, event]);
      setIsComplete(true);
      eventSource.close();
    });

    eventSource.addEventListener('error', (e) => {
      const event = JSON.parse(e.data) as ProgressEvent;
      setError(event.message || 'Unknown error');
      setEvents((prev) => [...prev, event]);
    });

    // Connection error handler
    eventSource.onerror = (error) => {
      console.error('EventSource failed:', error);
      eventSource.close();
    };

    // Cleanup on unmount
    return () => {
      eventSource.close();
    };
  }, [analysisId]);

  return { events, isComplete, error };
}

UI Component

import { useAnalysisProgress } from '@/features/analysis/hooks/useAnalysisProgress';

export function AnalysisProgress({ analysisId }: { analysisId: string }) {
  const { events, isComplete, error } = useAnalysisProgress(analysisId);

  if (error) {
    return <div className="error">Error: {error}</div>;
  }

  return (
    <div>
      <h3>Analysis Progress</h3>
      {events.map((event, idx) => (
        <div key={idx} className="event">
          <strong>{event.stage}</strong>: {event.status}
        </div>
      ))}
      {isComplete && <div className="success">✓ Analysis complete!</div>}
    </div>
  );
}

Testing

Manual Test (cURL)

curl -N -H "Accept: text/event-stream" \
  http://localhost:8500/api/v1/analyze/550e8400-e29b-41d4-a716-446655440000/stream

Expected output:

event: progress
data: {"type":"progress","stage":"extraction","status":"running","timestamp":"2025-12-21T10:30:15Z"}

event: progress
data: {"type":"progress","stage":"extraction","status":"complete","word_count":5234}

event: complete
data: {"type":"complete","stage":"artifact_generation","timestamp":"2025-12-21T10:32:45Z"}

Unit Test (Broadcaster)

Location: backend/tests/unit/test_event_broadcaster.py

import pytest
from app.shared.services.messaging.broadcaster import EventBroadcaster

@pytest.mark.asyncio
async def test_event_buffering():
    """Test that events are buffered for late-joining subscribers."""
    broadcaster = EventBroadcaster()
    channel = "test:123"

    # Publish events BEFORE subscriber connects
    await broadcaster.publish(channel, {"type": "event1"})
    await broadcaster.publish(channel, {"type": "event2"})

    # Subscribe AFTER events were published
    events = []
    async for event in broadcaster.subscribe(channel):
        events.append(event)
        if len(events) == 2:
            break

    # Verify subscriber received buffered events
    assert len(events) == 2
    assert events[0]["type"] == "event1"
    assert events[1]["type"] == "event2"

Event Types

1. Progress Event

{
  "type": "progress",
  "stage": "extraction",
  "status": "running",
  "timestamp": "2025-12-21T10:30:15Z"
}

2. Progress Complete Event

{
  "type": "progress",
  "stage": "extraction",
  "status": "complete",
  "word_count": 5234,
  "timestamp": "2025-12-21T10:30:45Z"
}

3. Error Event

{
  "type": "error",
  "stage": "extraction",
  "error": "Failed to fetch URL: Connection timeout",
  "timestamp": "2025-12-21T10:30:20Z"
}

4. Complete Event

{
  "type": "complete",
  "stage": "artifact_generation",
  "timestamp": "2025-12-21T10:32:45Z"
}

Best Practices Learned

1. Always Buffer Events

Lesson: Even with "fast" workflows, race conditions happen Solution: Always buffer recent events (OrchestKit: 100 events, 5-minute TTL)

2. Use aclosing() for Generators

Lesson: Async generators don't auto-close on exception Solution: Wrap with aclosing() for guaranteed cleanup

3. Send Errors as Events

Lesson: HTTP errors close SSE connection Solution: Send errors as SSE events, keep connection alive

4. Include Timestamps

Lesson: Hard to debug timing issues without timestamps Solution: Include ISO 8601 timestamp in every event

5. Clear Buffers After Completion

Lesson: Buffers consume memory indefinitely Solution: Clear buffer after complete event

Related Files

• Broadcaster: backend/app/shared/services/messaging/broadcaster.py
• SSE Handler: backend/app/api/v1/analysis/sse_handler.py
• Endpoint: backend/app/api/v1/analysis/endpoints.py
• Workflow Runner: backend/app/api/v1/analysis/workflow_runner.py
• Tests: backend/tests/unit/test_event_broadcaster.py

References

• See references/sse-deep-dive.md for protocol details
• See scripts/sse-endpoint-template.ts for TypeScript client template
• See sse-starlette documentation: https://github.com/sysid/sse-starlette

Server-Sent Events (SSE) Deep Dive

Comprehensive guide to Server-Sent Events including protocol details, reconnection strategies, event types, buffering, backpressure handling, and production patterns.

SSE Protocol Fundamentals

What is SSE?

Server-Sent Events (SSE) is a web standard for server-to-client unidirectional streaming over HTTP. Unlike WebSockets (bidirectional), SSE uses regular HTTP and provides automatic reconnection.

Use cases:

• Real-time progress updates (file uploads, analysis workflows)
• Live notifications (chat messages, alerts)
• Streaming LLM responses (ChatGPT-style interfaces)
• Live data feeds (stock prices, sports scores)

Not suitable for:

• Bidirectional communication (use WebSockets)
• Binary data (SSE is text-based)
• Low latency requirements (<50ms) (use WebSockets)

HTTP Response Format

Server response headers:

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
X-Accel-Buffering: no  # Disable nginx buffering

Event format:

event: message
data: {"content": "Hello"}
id: 123
retry: 3000

Key fields:

• event: Event type (default: "message")
• data: Payload (can span multiple lines)
• id: Event ID for reconnection
• retry: Reconnection delay in milliseconds
• Terminator: Two newlines (\n\n) end each event

Multi-line Data

event: progress
data: {
data:   "stage": "extraction",
data:   "status": "complete"
data: }

Parser concatenates: {"stage": "extraction", "status": "complete"}

Client Implementation

Basic EventSource

const eventSource = new EventSource('/api/v1/analyze/123/stream');

// Listen to default "message" events
eventSource.onmessage = (event) => {
  console.log('Received:', event.data);
};

// Listen to custom event types
eventSource.addEventListener('progress', (event) => {
  const data = JSON.parse(event.data);
  console.log(`Stage: ${data.stage}, Status: ${data.status}`);
});

eventSource.addEventListener('error', (event) => {
  const data = JSON.parse(event.data);
  console.error('Error:', data.message);
});

eventSource.addEventListener('complete', (event) => {
  console.log('Complete!');
  eventSource.close();  // Close connection when done
});

// Handle connection errors
eventSource.onerror = (error) => {
  console.error('EventSource failed:', error);
};

React Hook Example

import { useEffect, useState } from 'react';

interface ProgressEvent {
  type: 'progress' | 'error' | 'complete';
  stage: string;
  status: string;
  message?: string;
}

export function useAnalysisProgress(analysisId: string) {
  const [events, setEvents] = useState<ProgressEvent[]>([]);
  const [isComplete, setIsComplete] = useState(false);

  useEffect(() => {
    const eventSource = new EventSource(
      `/api/v1/analyze/${analysisId}/stream`
    );

    eventSource.addEventListener('progress', (e) => {
      const event = JSON.parse(e.data) as ProgressEvent;
      setEvents((prev) => [...prev, event]);
    });

    eventSource.addEventListener('complete', (e) => {
      const event = JSON.parse(e.data) as ProgressEvent;
      setEvents((prev) => [...prev, event]);
      setIsComplete(true);
      eventSource.close();
    });

    eventSource.addEventListener('error', (e) => {
      const event = JSON.parse(e.data) as ProgressEvent;
      setEvents((prev) => [...prev, event]);
    });

    // Cleanup on unmount
    return () => {
      eventSource.close();
    };
  }, [analysisId]);

  return { events, isComplete };
}

Server Implementation (FastAPI)

Basic SSE Endpoint

from fastapi import APIRouter
from sse_starlette.sse import EventSourceResponse
from collections.abc import AsyncIterator
import asyncio
import json

router = APIRouter()

async def event_generator() -> AsyncIterator[dict[str, str]]:
    """Generate SSE events."""
    for i in range(10):
        await asyncio.sleep(1)

        yield {
            "event": "progress",
            "data": json.dumps({
                "type": "progress",
                "count": i,
                "timestamp": datetime.now(UTC).isoformat()
            })
        }

    # Final event
    yield {
        "event": "complete",
        "data": json.dumps({"type": "complete"})
    }

@router.get("/stream")
async def stream_events():
    return EventSourceResponse(event_generator())

SSE with Pub/Sub (Production Pattern)

Use case: Multiple workflow tasks publish events → Single SSE endpoint subscribes and streams to client

from app.shared.services.messaging.broadcaster import broadcaster

@router.get("/analyze/{analysis_id}/stream")
async def stream_analysis_progress(
    analysis_id: uuid.UUID,
    request: Request
) -> EventSourceResponse:
    """Stream analysis progress via SSE."""
    channel = f"workflow:{analysis_id}"

    logger.info(
        "sse_connection_started",
        analysis_id=str(analysis_id),
        channel=channel
    )

    async def event_generator() -> AsyncIterator[dict[str, str]]:
        try:
            # Subscribe to broadcaster channel
            async with aclosing(broadcaster.subscribe(channel)) as subscription:
                async for event in subscription:
                    # Format event for SSE
                    event_type = str(event.get("type", "message"))
                    yield {
                        "event": event_type,
                        "data": json.dumps(event)
                    }

                    # Close connection on complete event
                    if event.get("type") == "complete":
                        logger.info(
                            "sse_complete_event_sent",
                            analysis_id=str(analysis_id)
                        )
                        break

        except asyncio.CancelledError:
            # Client disconnected
            logger.info(
                "sse_connection_cancelled",
                analysis_id=str(analysis_id)
            )
            raise
        except Exception as e:
            logger.error(
                "sse_unexpected_error",
                analysis_id=str(analysis_id),
                error=str(e),
                exc_info=True
            )
            # Send error event to client
            yield {
                "event": "error",
                "data": json.dumps({
                    "type": "error",
                    "error_type": type(e).__name__,
                    "message": str(e),
                    "timestamp": datetime.now(UTC).isoformat()
                })
            }

    return EventSourceResponse(
        event_generator(),
        send_timeout=30.0  # Timeout for unresponsive clients
    )

Event Buffering (Race Condition Solution)

Problem: SSE Race Condition

Scenario: Workflow starts emitting events BEFORE client connects to SSE endpoint

POST /analyze         → Analysis created, workflow started
  ↓
  Workflow emits events: extraction_started, extraction_complete
  ↓
GET /stream          → Client connects (LATE!)
  ↓
  Client receives: chunking_started, chunking_complete
  ✗ Client never sees: extraction events (LOST!)

Result: Frontend shows "Waiting for agent activity..." while backend runs

Solution: Event Buffering

Pattern: Buffer recent events per channel, replay to new subscribers

from collections import deque
from datetime import datetime, timedelta, UTC

MAX_BUFFER_SIZE = 100  # Max events to buffer per channel
BUFFER_TTL_SECONDS = 300  # 5 minutes - events older than this are dropped

class EventBroadcaster:
    def __init__(self) -> None:
        self._channels: dict[str, list[asyncio.Queue]] = defaultdict(list)
        self._buffers: dict[str, deque[tuple[datetime, dict]]] = defaultdict(
            lambda: deque(maxlen=MAX_BUFFER_SIZE)
        )
        self._lock = asyncio.Lock()

    async def publish(self, channel: str, message: dict) -> None:
        """Publish message to channel AND buffer it."""
        now = datetime.now(UTC)

        async with self._lock:
            # Always buffer the event (even if no subscribers)
            self._buffers[channel].append((now, message))

            # Clean up old events (older than TTL)
            cutoff = now - timedelta(seconds=BUFFER_TTL_SECONDS)
            while self._buffers[channel] and self._buffers[channel][0][0] < cutoff:
                self._buffers[channel].popleft()

            # Broadcast to active subscribers
            queues = self._channels.get(channel, [])
            for queue in queues:
                await queue.put(message)

    async def subscribe(self, channel: str) -> AsyncIterator[dict]:
        """Subscribe to channel, replaying buffered events first."""
        queue = asyncio.Queue()
        buffered_events = []

        async with self._lock:
            # Add subscriber
            self._channels[channel].append(queue)

            # Capture buffered events for replay
            if channel in self._buffers:
                buffered_events = [event for _, event in self._buffers[channel]]

        logger.debug(
            "subscribe_created",
            channel=channel,
            buffered_events_count=len(buffered_events)
        )

        try:
            # 1. Replay buffered events (catch up)
            for event in buffered_events:
                yield event

            # 2. Stream live events
            while True:
                message = await queue.get()
                yield message

        finally:
            # Cleanup on disconnect
            async with self._lock:
                if channel in self._channels:
                    self._channels[channel].remove(queue)
                    if not self._channels[channel]:
                        del self._channels[channel]

Benefits:

• Clients receive ALL events, even if they connect late
• No race condition
• Bounded memory (maxlen=100)
• Auto-cleanup (5 minute TTL)

Reconnection Strategies

Automatic Reconnection (Built-in)

EventSource automatically reconnects on connection failure:

const eventSource = new EventSource('/api/v1/stream');

// Browser automatically:
// 1. Detects connection loss
// 2. Waits 'retry' milliseconds (default: 3000ms)
// 3. Reconnects with Last-Event-ID header

Server can control retry delay:

yield {
    "event": "message",
    "data": json.dumps({"status": "ok"}),
    "retry": 5000  # Client will wait 5 seconds before reconnecting
}

Custom Reconnection (Exponential Backoff)

For more control, implement custom reconnection:

class ReconnectingEventSource {
  private eventSource: EventSource | null = null;
  private reconnectDelay = 1000;
  private maxReconnectDelay = 30000;
  private reconnectAttempts = 0;

  constructor(
    private url: string,
    private onMessage: (data: string) => void,
    private onError?: (error: Event) => void
  ) {
    this.connect();
  }

  private connect() {
    this.eventSource = new EventSource(this.url);

    this.eventSource.onmessage = (event) => {
      this.reconnectDelay = 1000; // Reset on success
      this.reconnectAttempts = 0;
      this.onMessage(event.data);
    };

    this.eventSource.onerror = (error) => {
      console.warn(
        `SSE error (attempt ${this.reconnectAttempts}):`,
        error
      );

      this.eventSource?.close();
      this.reconnectAttempts++;

      // Exponential backoff with max delay
      setTimeout(() => this.connect(), this.reconnectDelay);
      this.reconnectDelay = Math.min(
        this.reconnectDelay * 2,
        this.maxReconnectDelay
      );

      this.onError?.(error);
    };
  }

  close() {
    this.eventSource?.close();
    this.eventSource = null;
  }
}

// Usage
const sse = new ReconnectingEventSource(
  '/api/v1/stream',
  (data) => console.log('Message:', data),
  (error) => console.error('Error:', error)
);

Last-Event-ID for Resume

Server sends event IDs:

event_id = 0

async def event_generator():
    global event_id
    for item in data:
        event_id += 1
        yield {
            "id": str(event_id),
            "event": "progress",
            "data": json.dumps(item)
        }

Client reconnects with Last-Event-ID:

// Browser automatically includes header on reconnect:
// Last-Event-ID: 42

Server resumes from last ID:

@router.get("/stream")
async def stream_events(request: Request):
    last_event_id = request.headers.get("Last-Event-ID")
    start_id = int(last_event_id) if last_event_id else 0

    async def event_generator():
        for event_id in range(start_id + 1, 100):
            yield {
                "id": str(event_id),
                "data": json.dumps({"count": event_id})
            }

    return EventSourceResponse(event_generator())

Keep-Alive / Heartbeat

Problem: Proxy/Load Balancer Timeout

Scenario: No events for 60 seconds → Nginx/ALB closes connection

Solution: Send Keep-Alive Comments

async def event_generator():
    last_event_time = time.time()

    while True:
        # Send heartbeat every 30 seconds
        if time.time() - last_event_time > 30:
            yield {
                "comment": "keepalive"  # Special field: ignored by client
            }
            last_event_time = time.time()

        # Or send actual event
        if has_event():
            yield {
                "event": "progress",
                "data": json.dumps(get_event())
            }
            last_event_time = time.time()

        await asyncio.sleep(1)

SSE comment format:

: keepalive

Lines starting with : are comments (ignored by EventSource parser).

Backpressure Handling

Problem: Slow Consumer

Scenario: Server produces events faster than client can consume

Solution: Monitor Queue Size

MAX_QUEUE_SIZE = 100

async def event_generator(channel: str):
    queue = asyncio.Queue(maxsize=MAX_QUEUE_SIZE)

    async def producer():
        try:
            async for event in data_source():
                try:
                    # Non-blocking put with timeout
                    await asyncio.wait_for(
                        queue.put(event),
                        timeout=5.0
                    )
                except asyncio.TimeoutError:
                    logger.warning(
                        "backpressure_detected",
                        queue_size=queue.qsize()
                    )
                    # Drop event or pause producer
        finally:
            await queue.put(None)  # Sentinel

    # Start producer in background
    asyncio.create_task(producer())

    # Consume from queue
    while True:
        event = await queue.get()
        if event is None:  # Sentinel
            break
        yield event

Error Handling

Send Structured Error Events

try:
    async for event in subscription:
        yield {"event": "progress", "data": json.dumps(event)}
except ConnectionError as e:
    yield {
        "event": "error",
        "data": json.dumps({
            "type": "error",
            "error_type": "ConnectionError",
            "message": str(e),
            "timestamp": datetime.now(UTC).isoformat()
        })
    }

Client handling:

eventSource.addEventListener('error', (e) => {
  const error = JSON.parse(e.data);
  console.error(`Error (${error.error_type}):`, error.message);

  // Maybe close connection on specific errors
  if (error.error_type === 'FatalError') {
    eventSource.close();
  }
});

Browser Limits

Connection Limit: 6 per Domain

Problem: Browser limits SSE connections to 6 per domain (HTTP/1.1)

Solutions: 1. Use HTTP/2: No connection limit (multiplexing) 2. Close old connections: When opening new SSE, close previous 3. Use single connection: Multiplex multiple streams over one SSE

// Close old connection before opening new one
if (window.currentSSE) {
  window.currentSSE.close();
}
window.currentSSE = new EventSource('/stream');

Production Checklist

Server Configuration

return EventSourceResponse(
    event_generator(),
    headers={
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",  # Disable nginx buffering
    },
    send_timeout=30.0,  # Timeout for unresponsive clients
    ping=15.0,  # Send ping every 15 seconds
)

Nginx Configuration

location /api/v1/stream {
    proxy_pass http://backend;
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding on;
    proxy_read_timeout 3600s;  # Long timeout for SSE
}

Load Balancer (AWS ALB)

# Target group settings
idle_timeout: 3600  # 1 hour (default: 60s)
deregistration_delay: 30

Testing SSE Endpoints

cURL Test

curl -N -H "Accept: text/event-stream" \
  http://localhost:8500/api/v1/analyze/123/stream

Output:

event: progress
data: {"type":"progress","stage":"extraction","status":"running"}

event: progress
data: {"type":"progress","stage":"extraction","status":"complete"}

event: complete
data: {"type":"complete"}

Python Test

import httpx

async def test_sse():
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "GET",
            "http://localhost:8500/api/v1/analyze/123/stream",
            headers={"Accept": "text/event-stream"}
        ) as response:
            async for line in response.aiter_lines():
                print(line)

Related Files

• See examples/orchestkit-sse-implementation.md for OrchestKit-specific patterns
• See scripts/sse-endpoint-template.ts for TypeScript client template
• See SKILL.md for WebSocket comparison and LLM streaming patterns