How to Build a Streaming Chatbot API in Python with FastAPI and SSE

Most chatbot UIs feel slow when they wait for the complete LLM response before showing anything. Server-Sent Events (SSE) fix this: the backend streams tokens as they're generated, and users see output appear character by character. Here's how to wire that up with FastAPI.

Why SSE and Not WebSockets?

WebSockets are bidirectional. For a chatbot where the client sends one message and the server streams back a response, that's more complexity than you need. SSE is:

• Unidirectional (server → client), which matches the LLM streaming model exactly
• HTTP/1.1 compatible — no upgrade handshake, works through most proxies and CDNs
• Automatically reconnecting in browsers via the native EventSource API
• Simpler to implement on the server side — one endpoint, no handshake state

If you need the client to send continuous data while receiving, use WebSockets. For a chatbot, SSE is the right tool.

Setting Up the Project

pip install fastapi uvicorn httpx python-dotenv

Project layout:

chatbot-api/
├── main.py
├── llm.py
└── .env

.env:

LLM_API_KEY=your_key_here
LLM_API_URL=https://api.your-llm-provider.com/v1/chat/completions
LLM_MODEL=your-model-name

Building the SSE Endpoint with FastAPI

FastAPI supports SSE via StreamingResponse. The SSE wire format is strict: each chunk must be prefixed with data: and end with \n\n. Miss the double newline and the browser won't parse the event.

# main.py
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from llm import stream_chat_completion

app = FastAPI()

class ChatRequest(BaseModel):
    messages: list[dict]
    temperature: float = 0.7

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    async def event_generator():
        try:
            async for token in stream_chat_completion(req.messages, req.temperature):
                chunk = json.dumps({"token": token})
                yield f"data: {chunk}\n\n"
            yield "data: [DONE]\n\n"
        except GeneratorExit:
            return
        except Exception as e:
            error = json.dumps({"error": str(e)})
            yield f"data: {error}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Disable Nginx buffering
        },
    )

The X-Accel-Buffering: no header is easy to miss. Without it, Nginx buffers your SSE chunks and the client won't see them until the buffer fills — which defeats the entire purpose of streaming.

Streaming from the Language Model

Most language model APIs support a stream: true parameter that returns chunks in the OpenAI format. Here's an async generator that handles this for any OpenAI-compatible endpoint:

# llm.py
import os
import json
import httpx
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("LLM_API_KEY")
API_URL = os.getenv("LLM_API_URL")
MODEL = os.getenv("LLM_MODEL")

async def stream_chat_completion(
    messages: list[dict],
    temperature: float = 0.7,
):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": MODEL,
        "messages": messages,
        "temperature": temperature,
        "stream": True,
    }

    async with httpx.AsyncClient(timeout=60.0) as client:
        async with client.stream("POST", API_URL, headers=headers, json=payload) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line or line == "data: [DONE]":
                    continue
                if line.startswith("data: "):
                    raw = line[6:]  # strip "data: " prefix
                    try:
                        chunk = json.loads(raw)
                        delta = chunk["choices"][0]["delta"]
                        if "content" in delta and delta["content"]:
                            yield delta["content"]
                    except (json.JSONDecodeError, KeyError):
                        continue

Swap LLM_API_URL and LLM_MODEL in .env to target different providers. The generator works with any endpoint that follows the OpenAI streaming format.

Handling Disconnects and Timeouts

Two failure modes you'll encounter in production:

Client disconnects mid-stream. FastAPI raises GeneratorExit when the client drops the connection. The except GeneratorExit: return in event_generator() handles this cleanly — it exits without attempting to write to a closed connection.

Provider timeout. Set an explicit timeout in httpx.AsyncClient. Sixty seconds covers most cloud-hosted models. If you're running inference locally, bump it to 120+. A timeout that's too short silently truncates responses with no error surfaced to the user.

You can also check await request.is_disconnected() inside the loop to cancel expensive server-side work (abort a database query, cancel a downstream API call) when the client leaves.

Testing the Endpoint

Start the server:

uvicorn main:app --reload

Test with curl — --no-buffer shows chunks as they arrive rather than waiting for the full response:

curl -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  --no-buffer \
  -d '{"messages": [{"role": "user", "content": "Explain SSE in one sentence"}]}'

Expected output:

data: {"token": "SSE"}
data: {"token": " stands"}
data: {"token": " for"}
data: {"token": " Server-Sent"}
data: {"token": " Events"}
data: [DONE]

On the frontend, the native EventSource API only supports GET requests. For POST (which you need to send the message payload), use the fetch API with ReadableStream, or the @microsoft/fetch-event-source library:

import { fetchEventSource } from '@microsoft/fetch-event-source';

await fetchEventSource('/chat/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ messages: [{ role: 'user', content: userInput }] }),
  onmessage(event) {
    if (event.data === '[DONE]') return;
    const { token } = JSON.parse(event.data);
    outputEl.textContent += token;
  },
});

Production Considerations

Three things to address before shipping:

Authentication. Add a FastAPI dependency that validates Bearer tokens or API keys on /chat/stream. SSE connections are long-lived HTTP connections — an open, unauthenticated streaming endpoint is a resource drain. For a practical checklist covering API authentication patterns including token validation, rate limiting, and header security, see our free security hardening checklists.

Rate limiting. Each streaming connection holds open a connection for the full generation duration — sometimes 30–60 seconds. Without per-client limits on concurrent requests, a single user can exhaust your server's thread pool.

Logging. Standard request/response logging won't capture the full response text since you're streaming. Buffer tokens server-side during generation and write the complete assembled response to your log store after the stream ends.

The Takeaway

Streaming chatbot APIs with FastAPI and SSE come down to a few concrete things:

1. Use StreamingResponse with media_type="text/event-stream"
2. Format every chunk as data: <json>\n\n — both newlines are required
3. Add X-Accel-Buffering: no if you're behind Nginx
4. Catch GeneratorExit for clean client disconnect handling
5. Set explicit timeouts on your HTTP client — silent truncation is harder to debug than a visible error

The stream: true flag is supported by virtually every major language model API. Once you have this async generator pattern working, switching between providers is a URL and a key swap in .env.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.