FastAPI Streaming LLM Responses with Server-Sent Events (SSE) in Python

In modern AI applications, users should never stare at a blank screen waiting 10 seconds for a full 500-token response. Streaming tokens in real-time as they are generated by the model is the expected UX standard.

In this tutorial, we build a production FastAPI application that streams LLM tokens to web clients using Server-Sent Events (SSE), complete with client disconnection detection and error recovery.

1. Architecture: Server-Sent Events Streaming Lifecycle

┌─────────────────┐             HTTP GET /stream             ┌─────────────────┐
│                 ├─────────────────────────────────────────►│                 │
│                 │   200 OK (Content-Type: text/event-stream)│     FastAPI     │
│                 │◄─────────────────────────────────────────┤  Async Engine   │
│   Web Browser   │               data: "Hello"              │                 │
│  (EventSource)  │◄─────────────────────────────────────────┤        │        │
│                 │               data: " World!"            │        ▼        │
│                 │◄─────────────────────────────────────────┤   Local LLM /   │
│                 │               data: [DONE]               │   Ollama Stream │
│                 │◄─────────────────────────────────────────┤                 │
└─────────────────┘                                          └─────────────────┘

2. Environment Setup

pip install fastapi>=0.111.0 uvicorn>=0.30.0 httpx>=0.27.0

3. Implementation: FastAPI SSE Streamer

Create main.py:

import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="TechJunkGigs SSE Streamer")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

async def token_generator(prompt: str, request: Request):
    # Simulates real-time token streaming from an LLM inference engine.
    tokens = [
        "In ", "2026, ", "building ", "production-grade ", "AI ",
        "applications ", "requires ", "resilient ", "streaming ",
        "architectures ", "with ", "proper ", "SSE ", "formatting."
    ]

    for token in tokens:
        # Check if the client disconnected (e.g. closed the browser tab)
        if await request.is_disconnected():
            print("Client disconnected early; stopping inference loop.")
            break

        # Format payload following SSE specification (data: <payload>\n\n)
        payload = json.dumps({"token": token})
        yield f"data: {payload}\n\n"
        await asyncio.sleep(0.08) # Simulate token generation latency

    yield "data: [DONE]\n\n"

@app.get("/api/v1/chat/stream")
async def chat_stream(prompt: str, request: Request):
    return StreamingResponse(
        token_generator(prompt, request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no" # Prevents NGINX from buffering chunked responses
        }
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

4. Frontend Client Implementation

Consume the stream in JavaScript using the native EventSource API:

const evtSource = new EventSource("/api/v1/chat/stream?prompt=Explain+FastAPI");
const outputDiv = document.getElementById("chat-output");

evtSource.onmessage = function(event) {
    if (event.data === "[DONE]") {
        evtSource.close();
        return;
    }
    const parsed = JSON.parse(event.data);
    outputDiv.textContent += parsed.token;
};

5. Frequently Asked Questions (FAQ)

Why use SSE instead of WebSockets?

Server-Sent Events run over standard HTTP, require zero custom protocol handshakes, handle automatic reconnection out of the box, and pass cleanly through enterprise firewalls. WebSockets are only necessary if you need bidirectional streaming simultaneously.