Documents

Real-Time Streaming with SSE and FastAPI

6 min readApr 1, 2025Feb 22, 2026

SSE Streaming Responses - Real-Time Data Delivery with FastAPI

Because LLMs generate responses token by token, making users wait for the full response before seeing anything creates a poor experience. SSE (Server-Sent Events) lets you push each token to the client the moment it's generated.

What is SSE?

SSE is an HTTP-based protocol for unidirectional, real-time data streaming from server to client. Unlike WebSockets, it's one-way only, but that also makes it far simpler to implement and plays well with HTTP/2.

Implementing SSE in FastAPI

from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from typing import AsyncGenerator
import json

router = APIRouter()

async def stream_workflow_execution(
    workflow_id: str,
) -> AsyncGenerator[str, None]:
    executor = WorkflowExecutor(workflow_id)
    levels = executor.get_execution_order()

    for level_idx, level in enumerate(levels):
        # Level start event
        yield f"data: {json.dumps({'type': 'level_start', 'level': level_idx})}\n\n"

        for node_id in level:
            # Node execution start
            yield f"data: {json.dumps({'type': 'node_start', 'node_id': node_id})}\n\n"

            async for chunk in executor.execute_node_stream(node_id):
                yield f"data: {json.dumps({'type': 'chunk', 'node_id': node_id, 'content': chunk})}\n\n"

            # Node execution complete
            yield f"data: {json.dumps({'type': 'node_complete', 'node_id': node_id})}\n\n"

    yield f"data: {json.dumps({'type': 'workflow_complete'})}\n\n"

@router.post("/api/workflow/{workflow_id}/execute/stream")
async def execute_workflow_stream(workflow_id: str):
    return StreamingResponse(
        stream_workflow_execution(workflow_id),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Disable Nginx buffering
        },
    )

Integrating LLM Streaming

Here's how to bridge OpenAI API streaming responses into SSE.

async def stream_llm_response(
    messages: list, model: str = "gpt-4o"
) -> AsyncGenerator[str, None]:
    client = AsyncOpenAI()
    stream = await client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
    )

    async for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

Consuming SSE on the Frontend

const eventSource = new EventSource(`/api/workflow/${id}/execute/stream`);

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  switch (data.type) {
    case 'chunk':
      appendToNodeOutput(data.node_id, data.content);
      break;
    case 'node_complete':
      markNodeComplete(data.node_id);
      break;
    case 'workflow_complete':
      eventSource.close();
      break;
  }
};

Nginx Configuration Gotchas

When running SSE behind an Nginx reverse proxy, buffering settings require careful attention.

location /api/ {
    proxy_pass http://backend:8000;
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding off;
}

SSE is essentially a must-have for any LLM-based service. It looks straightforward, but error handling and connection management surface a surprising number of edge cases. I'll cover those in a dedicated Bug Fix post.

Tags
SSEStreamingFastAPIEventSourceReal-time