Build a Streaming AI Chat Backend with FastAPI and Server-Sent Events (Step-by-Step)

Most AI chat tutorials stop at a single blocking API call. Real products need streaming: users see tokens appear the moment the model generates them, not ten seconds later. In this tutorial you'll build a production-ready streaming chat backend in Python — FastAPI on the backend, Server-Sent Events (SSE) to the browser, token-by-token output from any model on the Qubax API.

By the end you'll have a working endpoint that streams responses from any chat model, handles errors and timeouts gracefully, and stays cheap even under load.

What You'll Need

Python 3.11+
A Qubax API key (sign up at qubax.ai)
Basic FastAPI familiarity

bash

pip install fastapi uvicorn httpx sse-starlette

Why Server-Sent Events?

For AI streaming you have three options:

WebSockets — bidirectional, but overkill: chat streaming is one-way after the request
Chunked HTTP — works, but browsers can't consume it natively
SSE — one-way server→client streaming, native browser support, automatic reconnection

SSE is the right tool: it's just HTTP, passes through every proxy and load balancer, and the browser's EventSource API handles reconnection for free.

Step 1: The Streaming Client

First, a thin async client that streams from the Qubax API using the OpenAI-compatible endpoint:

python

# llm_client.py
import httpx
from typing import AsyncIterator

QUBAX_BASE = "https://api.qubax.ai/v1"

async def stream_chat(
    api_key: str,
    model: str,
    messages: list[dict],
    timeout: float = 120.0,
) -> AsyncIterator[str]:
    """Yield content deltas from the model as they arrive."""
    async with httpx.AsyncClient(timeout=timeout) as client:
        async with client.stream(
            "POST",
            f"{QUBAX_BASE}/chat/completions",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "model": model,
                "messages": messages,
                "stream": True,
            },
        ) as response:
            response.raise_for_status()
            async for line in response.aiter_lines():
                if not line.startswith("data: "):
                    continue
                data = line.removeprefix("data: ").strip()
                if data == "[DONE]":
                    return
                delta = parse_delta(data)
                if delta:
                    yield delta

def parse_delta(chunk_json: str) -> str | None:
    import json
    try:
        chunk = json.loads(chunk_json)
        return chunk["choices"][0]["delta"].get("content")
    except (json.JSONDecodeError, KeyError, IndexError):
        return None

The two details people get wrong:

`client.stream()` context manager — without it, httpx buffers the entire response and you get no streaming at all.
`aiter_lines()` — SSE is line-delimited; each data: line is one JSON chunk ending with a final [DONE] sentinel.

Step 2: The SSE Endpoint

Now expose it through FastAPI:

python

# main.py
import os, json
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import httpx
from llm_client import stream_chat

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    model: str = "deepseek-v4-flash"
    history: list[dict] = []

@app.post("/api/chat")
async def chat(req: ChatRequest):
    api_key = os.environ["QUBAX_API_KEY"]
    if not req.message.strip():
        raise HTTPException(400, "message is required")
    if len(req.history) > 20:
        raise HTTPException(400, "history too long")

    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        *req.history[-10:],
        {"role": "user", "content": req.message},
    ]

    async def event_stream():
        try:
            async for delta in stream_chat(api_key, req.model, messages):
                yield f"data: {json.dumps({'t': delta})}\n\n"
        except httpx.TimeoutException:
            yield f"data: {json.dumps({'error': 'model timeout'})}\n\n"
        except httpx.HTTPStatusError as e:
            yield f"data: {json.dumps({'error': f'upstream {e.response.status_code}'})}\n\n"
        yield "data: [DONE]\n\n"

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

Note the X-Accel-Buffering: no header. If you deploy behind nginx, it buffers responses by default and your beautiful token streaming arrives as one giant blob at the end. This header disables that per-response.

Step 3: The Frontend

The browser side is minimal:

javascript

async function streamChat(message, history) {
  const res = await fetch("/api/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ message, history }),
  });

  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });

    const lines = buffer.split("\n\n");
    buffer = lines.pop();  // keep incomplete chunk in buffer

    for (const line of lines) {
      if (!line.startsWith("data: ")) continue;
      const data = line.slice(6);
      if (data === "[DONE]") return;
      const { t, error } = JSON.parse(data);
      if (error) { showError(error); return; }
      if (t) appendToken(t);  // your UI update
    }
  }
}

The buffering dance (lines.pop()) matters: network chunks don't respect SSE message boundaries, so you must accumulate partial messages.

Step 4: Cost Control

Streaming makes it easy to forget you're paying per token. Three habits keep bills sane:

Cap output tokens — add "max_tokens": 2048 to your request payload.
Trim history — the code above sends only the last 10 turns; older context is usually dead weight.
Route by task — use a fast model like DeepSeek V4 Flash (around $0.03/M input tokens on Qubax) for chat, and save frontier models for hard reasoning. A simple two-tier router can cut costs 10x: see our cost-router tutorial.

Testing It

bash

uvicorn main:app --reload
curl -N -X POST localhost:8000/api/chat \
  -H 'Content-Type: application/json' \
  -d '{"message": "Explain SSE in one paragraph"}'

The -N flag disables curl buffering so you see tokens arrive live.

Production Checklist

Before shipping:

[ ] Set max_tokens on every request
[ ] Add per-user rate limiting (e.g., slowapi)
[ ] Log token usage per request for cost attribution
[ ] Handle client disconnects — cancel the upstream stream when the user closes the tab (check request.is_disconnected())
[ ] Keep the API key server-side only; never expose it to the browser

This whole stack runs on any model available on Qubax — swap the model field and nothing else changes. Browse models and live pricing at qubax.ai/models, and read the full API reference at qubax.ai/docs.

FAQ

Why SSE instead of WebSockets?

Chat streaming is one-directional after the initial request. SSE is plain HTTP, simpler to deploy, passes corporate proxies, and gives you free browser reconnection. Use WebSockets only when the client also streams data continuously (e.g., voice).

Does this work with any model?

Yes — the Qubax API is OpenAI-compatible, so any chat model on qubax.ai/models works with the same code. Just change the model name.

How do I measure tokens for billing?

The final streamed chunk includes usage data in the Qubax API, or make a follow-up non-streaming usage call. Log prompt_tokens and completion_tokens per request.

What about multi-model streaming in one conversation?

Keep the history array in your frontend and pass a different model per request — the backend is already model-agnostic.