streaming-llm-responses

name	streaming-llm-responses
description	Implement backend SSE (Server-Sent Events) infrastructure for real-time AI response streaming. Use when building the Stream Broker service, Redis Pub/Sub listeners, or SSE endpoints. Covers sse-starlette integration, connection management, heartbeats, and event routing. NOT for frontend consumption patterns - this is backend infrastructure only.

Streaming LLM Responses (Backend SSE Infrastructure)

Build the real-time event delivery layer using FastAPI, sse-starlette, and Redis Pub/Sub.

Quick Start

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
import redis.asyncio as redis

app = FastAPI()

@app.get("/events/stream/{user_id}")
async def stream_events(user_id: str):
    async def event_generator():
        pubsub = redis_client.pubsub()
        await pubsub.subscribe(f"events:stream:{user_id}")
        try:
            async for message in pubsub.listen():
                if message["type"] == "message":
                    yield {"event": "delta", "data": message["data"]}
        finally:
            await pubsub.unsubscribe(f"events:stream:{user_id}")

    return EventSourceResponse(event_generator())

---

Architecture: The Fan-Out Pattern

┌─────────────────┐    Redis Pub/Sub    ┌─────────────────┐
│ Chat Orchestrator│ ──publish──────────► │  Stream Broker  │
│ Tool Executor    │   (fire & forget)   │  (SSE Server)   │
│ Media Processor  │                     │                 │
└─────────────────┘                     └────────┬────────┘
                                                 │
                                          SSE connections
                                                 │
                                    ┌────────────┼────────────┐
                                    ▼            ▼            ▼
                               [Client A]  [Client B]   [Client C]

Key Principle: Backend services publish events without knowing about HTTP clients. The Stream Broker handles all connection management.

---

Core Patterns

1. SSE Endpoint with sse-starlette

from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize Redis connection pool on startup
    app.state.redis = redis.Redis.from_url(
        settings.redis_url,
        encoding="utf-8",
        decode_responses=True
    )
    yield
    # Cleanup on shutdown
    await app.state.redis.close()

app = FastAPI(lifespan=lifespan)

@app.get("/events/stream/{user_id}")
async def stream_events(user_id: str, request: Request):
    """SSE endpoint for real-time event streaming."""

    async def event_generator():
        pubsub = request.app.state.redis.pubsub()
        channel = f"events:stream:{user_id}"

        await pubsub.subscribe(channel)

        try:
            async for message in pubsub.listen():
                # Check if client disconnected
                if await request.is_disconnected():
                    break

                if message["type"] == "message":
                    event = parse_event(message["data"])
                    yield {
                        "event": event.type,
                        "data": event.model_dump_json(),
                        "id": event.id,
                    }
        finally:
            await pubsub.unsubscribe(channel)
            await pubsub.close()

    return EventSourceResponse(
        event_generator(),
        media_type="text/event-stream"
    )

2. Heartbeat Keep-Alive

Prevent load balancers from terminating idle connections:

import asyncio
from typing import AsyncGenerator

HEARTBEAT_INTERVAL = 15  # seconds

async def event_generator_with_heartbeat(
    pubsub: redis.client.PubSub,
    channel: str,
    request: Request
) -> AsyncGenerator[dict, None]:
    """Generator that yields events and periodic heartbeats."""

    await pubsub.subscribe(channel)

    try:
        while True:
            if await request.is_disconnected():
                break

            try:
                # Non-blocking check for messages with timeout
                message = await asyncio.wait_for(
                    pubsub.get_message(ignore_subscribe_messages=True),
                    timeout=HEARTBEAT_INTERVAL
                )

                if message and message["type"] == "message":
                    event = parse_event(message["data"])
                    yield {
                        "event": event.type,
                        "data": event.model_dump_json(),
                        "id": event.id,
                    }
            except asyncio.TimeoutError:
                # No message received, send heartbeat
                yield {"event": "heartbeat", "data": ""}

    finally:
        await pubsub.unsubscribe(channel)
        await pubsub.close()

3. Event Publishing (Other Services)

Services publish events without knowing about SSE:

# In Chat Orchestrator or Tool Executor
import redis.asyncio as redis
import json

class EventPublisher:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    async def publish_status(self, user_id: str, status: str, message: str):
        """Publish a status update event."""
        event = {
            "id": str(uuid.uuid4()),
            "type": "status",
            "payload": {"status": status, "message": message},
            "timestamp": datetime.utcnow().isoformat()
        }
        await self.redis.publish(
            f"events:stream:{user_id}",
            json.dumps(event)
        )

    async def publish_delta(self, user_id: str, content: str, message_id: str):
        """Publish a token delta event."""
        event = {
            "id": str(uuid.uuid4()),
            "type": "delta",
            "payload": {"content": content, "message_id": message_id},
            "timestamp": datetime.utcnow().isoformat()
        }
        await self.redis.publish(
            f"events:stream:{user_id}",
            json.dumps(event)
        )

    async def publish_tool_event(
        self,
        user_id: str,
        event_type: Literal["tool_start", "tool_end"],
        tool_name: str,
        tool_call_id: str,
        result: dict | None = None
    ):
        """Publish tool lifecycle events."""
        event = {
            "id": str(uuid.uuid4()),
            "type": event_type,
            "payload": {
                "tool_name": tool_name,
                "tool_call_id": tool_call_id,
                "result": result
            },
            "timestamp": datetime.utcnow().isoformat()
        }
        await self.redis.publish(
            f"events:stream:{user_id}",
            json.dumps(event)
        )

---

OmniChat Event Protocol

All SSE events follow this structure:

SSE Frame Format

event: <event_type>
id: <unique_event_id>
data: <json_payload>

Event Types

Event Type	Purpose	Payload Schema
status	Processing state updates	`{"status": "thinking"
delta	Token-by-token content	{"content": str, "message_id": str}
tool_start	Tool execution begins	{"tool_name": str, "tool_call_id": str}
tool_end	Tool execution complete	{"tool_name": str, "tool_call_id": str, "result": any}
error	Error notification	{"code": str, "message": str, "recoverable": bool}
heartbeat	Keep-alive signal	"" (empty)

Pydantic Event Models

from pydantic import BaseModel, Field
from typing import Literal
from datetime import datetime
import uuid

class BaseEvent(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = Field(default_factory=datetime.utcnow)

class StatusEvent(BaseEvent):
    type: Literal["status"] = "status"
    payload: "StatusPayload"

class StatusPayload(BaseModel):
    status: Literal["thinking", "searching", "generating"]
    message: str

class DeltaEvent(BaseEvent):
    type: Literal["delta"] = "delta"
    payload: "DeltaPayload"

class DeltaPayload(BaseModel):
    content: str
    message_id: str

class ToolStartEvent(BaseEvent):
    type: Literal["tool_start"] = "tool_start"
    payload: "ToolStartPayload"

class ToolStartPayload(BaseModel):
    tool_name: str
    tool_call_id: str

class ToolEndEvent(BaseEvent):
    type: Literal["tool_end"] = "tool_end"
    payload: "ToolEndPayload"

class ToolEndPayload(BaseModel):
    tool_name: str
    tool_call_id: str
    result: dict | None = None

class ErrorEvent(BaseEvent):
    type: Literal["error"] = "error"
    payload: "ErrorPayload"

class ErrorPayload(BaseModel):
    code: str
    message: str
    recoverable: bool = True

# Union type for parsing incoming events
StreamEvent = StatusEvent | DeltaEvent | ToolStartEvent | ToolEndEvent | ErrorEvent

---

Connection Management

Tracking Active Connections

from collections import defaultdict
import asyncio

class ConnectionManager:
    def __init__(self):
        self._connections: dict[str, set[asyncio.Task]] = defaultdict(set)
        self._lock = asyncio.Lock()

    async def add_connection(self, user_id: str, task: asyncio.Task):
        async with self._lock:
            self._connections[user_id].add(task)

    async def remove_connection(self, user_id: str, task: asyncio.Task):
        async with self._lock:
            self._connections[user_id].discard(task)
            if not self._connections[user_id]:
                del self._connections[user_id]

    async def get_connection_count(self, user_id: str) -> int:
        async with self._lock:
            return len(self._connections.get(user_id, set()))

    async def get_total_connections(self) -> int:
        async with self._lock:
            return sum(len(conns) for conns in self._connections.values())

Graceful Shutdown

from contextlib import asynccontextmanager
import signal

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Setup
    app.state.redis = redis.Redis.from_url(settings.redis_url)
    app.state.connections = ConnectionManager()
    app.state.shutdown_event = asyncio.Event()

    yield

    # Teardown - signal all connections to close
    app.state.shutdown_event.set()

    # Wait for connections to drain (with timeout)
    for _ in range(10):  # 10 second timeout
        if await app.state.connections.get_total_connections() == 0:
            break
        await asyncio.sleep(1)

    await app.state.redis.close()

---

Redis Channel Naming Convention

events:stream:{user_id}      # Per-user event channel
events:broadcast:{scope}     # System-wide broadcasts (maintenance, etc.)

Use the channel format defined in omnichat.shared.constants when available.

---

Testing SSE Endpoints

With curl

# Connect to SSE stream
curl -N -H "Accept: text/event-stream" \
  "http://localhost:8000/events/stream/user123"

# In another terminal, publish a test event
redis-cli PUBLISH "events:stream:user123" \
  '{"id":"test-1","type":"status","payload":{"status":"thinking","message":"Processing..."}}'

With pytest

import pytest
from httpx import AsyncClient
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_sse_receives_published_events(app, redis_client):
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Start SSE connection in background
        async with client.stream("GET", "/events/stream/user123") as response:
            # Publish event
            await redis_client.publish(
                "events:stream:user123",
                '{"id":"1","type":"status","payload":{"status":"thinking","message":"Test"}}'
            )

            # Read first event (skip heartbeats)
            async for line in response.aiter_lines():
                if line.startswith("event: status"):
                    break

            assert "status" in line

---

Anti-Patterns

1. Not handling disconnects - Always check request.is_disconnected() to avoid resource leaks
2. Missing heartbeats - Load balancers will kill idle connections after ~60s
3. Blocking in generator - Use asyncio.wait_for with timeouts, never block
4. Not cleaning up pubsub - Always unsubscribe and close in finally block
5. Storing state in SSE handler - Keep the broker stateless; use Redis for shared state
6. Missing error events - Always send error event before closing on failure

---

Verification

Run: python scripts/verify.py

Expected: [OK] streaming-llm-responses skill ready

If Verification Fails

1. Check: references/ folder has sse-patterns.md
2. Stop and report if still failing

References

• references/sse-patterns.md - Complete SSE and Redis Pub/Sub patterns