How to Build a Streaming AI Chat API with Token Fallbacks in Python (2026 Guide)

Every chat product streams tokens — users won't tolerate a spinner while a model composes a full answer. But streaming introduces a problem most tutorials skip: what happens when your model fails mid-stream? Rate limits, provider outages, and context-overflow errors all happen in production. This guide shows you how to build a streaming chat API with FastAPI that falls back gracefully to another model when the first one fails.

What You'll Build

• A FastAPI endpoint that streams model output via Server-Sent Events (SSE)
• A fallback chain: if the primary model errors before emitting tokens, automatically retry with a cheaper backup
• Proper client disconnect handling, so you don't pay for tokens nobody is reading

Stack: Python 3.11+, FastAPI, the OpenAI SDK (works with any OpenAI-compatible endpoint, including Qubax).

Step 1: Setup

pip install fastapi uvicorn openai
export QUBAX_API_KEY="sk-..."   # get one at qubax.ai

Step 2: The Streaming Client with Fallbacks

The key insight: with OpenAI-compatible APIs, you know a request probably works once the first chunk arrives. So the fallback logic is — try model A; if the connection or first chunk fails, try model B.

# chat.py
import os
from openai import OpenAI, APIError, RateLimitError

client = OpenAI(
    base_url="https://api.qubax.ai/v1",
    api_key=os.environ["QUBAX_API_KEY"],
)

FALLBACK_CHAIN = [
    "gpt-5.6-sol",        # primary — strong flagship
    "gpt-5.6-luna",       # backup — fast and cheap
    "glm-4.7-flash",      # last resort — ultra cheap
]

def stream_chat(messages: list[dict], model: str):
    """Yield text deltas from a model. Raises if the request fails."""
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
        max_tokens=1024,
    )
    got_any = False
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            got_any = True
            yield delta
    if not got_any:
        raise APIError("empty stream", request=None, body=None)

def stream_with_fallback(messages: list[dict]):
    """Try each model in order; stream from the first one that responds."""
    last_error = None
    for model in FALLBACK_CHAIN:
        try:
            yield from stream_chat(messages, model)
            return
        except (APIError, RateLimitError) as e:
            last_error = e
            continue
    raise RuntimeError(f"All models failed: {last_error}")

Step 3: The FastAPI Endpoint

FastAPI supports async generators with StreamingResponse. We use SSE formatting (data: ...\n\n) because it's trivial to parse in browsers and survives proxies better than raw chunked text.

# main.py
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from chat import stream_with_fallback

app = FastAPI()

async def sse_wrapper(messages, request: Request):
    async def gen():
        # stream_with_fallback is sync; run chunks through in a thread-friendly way
        for delta in stream_with_fallback(messages):
            if await request.is_disconnected():
                return  # stop paying for tokens nobody reads
            yield f"data: {json.dumps({'delta': delta})}\n\n"
        yield "data: [DONE]\n\n"
    return gen()

@app.post("/chat")
async def chat(request: Request):
    body = await request.json()
    messages = body.get("messages", [])
    if not messages:
        return {"error": "messages required"}

    generator = await sse_wrapper(messages, request)
    return StreamingResponse(
        generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Run it:

uvicorn main:app --host 0.0.0.0 --port 8000

Step 4: Test It

curl -N -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"messages":[{"role":"user","content":"Explain vector databases in two sentences."}]}'

You'll see deltas arriving one at a time:

data: {"delta": "A vector"}
data: {"delta": " database stores"}
data: {"delta": " text as embeddings..."}
data: [DONE]

Step 5: The Frontend (30 Seconds of JavaScript)

const res = await fetch("/chat", {
  method: "POST",
  headers: {"Content-Type": "application/json"},
  body: JSON.stringify({messages: conversation}),
});
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();
  for (const line of lines) {
    const payload = line.replace("data: ", "");
    if (payload === "[DONE]") break;
    chatBox.textContent += JSON.parse(payload).delta;
  }
}

Production Hardening Checklist

• Timeouts: wrap the stream with asyncio.wait_for so a stalled connection doesn't hang your worker.
• Token accounting: count usage from the final chunk (stream_options={"include_usage": true}) so your billing stays accurate across fallbacks.
• Client disconnects: the request.is_disconnected() check above matters — without it, background streams keep generating (and billing) after the user closes the tab.
• Sticky fallbacks: if the primary model fails for a user, prefer the backup for their next few requests instead of retrying the dead one each time.
• Rate-limit headers: pass provider rate-limit info through so your client can back off intelligently.

Common Pitfalls

Fallback after partial output. If model A emitted 200 tokens and then died, don't silently restart with model B — the user sees a garbled restart. Either buffer and only commit after the first N chunks succeed, or send an explicit "retrying" event.

Proxy buffering. Nginx buffers SSE by default; the X-Accel-Buffering: no header above fixes it. If you still see chunks arriving in bursts, this is why.

Invisible failures. Log which model served each request. When quality complaints come in, "your fallback chain degraded" is an answer you want data for, not a guess.

Take It Further

Once streaming and fallbacks work, the natural next step is routing by difficulty — send easy prompts to the cheap models in your chain and reserve the flagship for hard ones. Combined with streaming, that's where the real cost savings live.

Want to pick the right models for your chain? Compare latency, pricing and benchmarks across 300+ models at qubax.ai/models, and find more integration guides at qubax.ai/docs.

FAQ

Why use Server-Sent Events instead of WebSockets?

SSE is one-way (server to client), which is exactly what token streaming needs. It works over plain HTTP, reconnects automatically, and needs no special server infrastructure. Use WebSockets only if the client also streams data continuously.

Does streaming cost more than non-streaming?

No. Token pricing is identical either way — streaming is purely about delivery, not billing. Streaming actually saves money indirectly because users can cancel early.

How do I handle mid-stream failures?

Decide by how much output was already delivered: before the first tokens, fail over transparently; after, either notify the user and let them regenerate, or buffer the first few chunks before committing to a model.

Can I use this with any OpenAI-compatible API?

Yes. The code only assumes OpenAI-compatible chat completion endpoints, so it works with Qubax, OpenAI, OpenRouter, or self-hosted vLLM servers unchanged.