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.