If your app waits for an entire LLM response before showing a single word, your users are staring at spinners. Streaming fixes that — tokens arrive as they're generated, so the first character appears within hundreds of milliseconds instead of seconds.
The good news: nearly every modern AI API supports Server-Sent Events (SSE) for streaming, and because Qubax exposes a fully OpenAI-compatible /v1/chat/completions endpoint, the same client code works across GLM, DeepSeek, Claude, GPT, Gemini, and Qwen — just by changing the model string.
This tutorial builds a complete Python streaming client from scratch: delta parsing, live token counting, usage accounting, per-request cost math, error handling, and reconnect logic. By the end you'll have production-ready code you can drop into any app.
What You'll Need
- Python 3.10+
- A Qubax API key (generate one at qubax.ai — your key works across every model)
- The
requestsandrichpackages:pip install requests rich
We'll use plain requests rather than an SDK so every byte of the protocol is visible. The same logic ports directly to httpx (for async) or aiohttp.
Step 1: The Streaming Request
SSE streaming is just a POST request with stream: true in the body. The server responds with Content-Type: text/event-stream and sends a sequence of data: lines, each containing a JSON object, ending with data: [DONE].
Here's the minimal version:
import json
import requests
QUBAX_API_KEY = "sk-qubax-..." # your key from qubax.ai
QUBAX_BASE_URL = "https://api.qubax.ai"
def stream_chat(model: str, messages: list[dict], **kwargs):
"""Yield content deltas from an OpenAI-compatible streaming endpoint."""
headers = {
"Authorization": f"Bearer {QUBAX_API_KEY}",
"Content-Type": "application/json",
}
payload = {
"model": model,
"messages": messages,
"stream": True,
"stream_options": {"include_usage": True}, # critical — see Step 4
**kwargs,
}
with requests.post(
f"{QUBAX_BASE_URL}/v1/chat/completions",
headers=headers,
json=payload,
stream=True,
timeout=120,
) as resp:
resp.raise_for_status()
for line in resp.iter_lines(decode_unicode=True):
if not line or not line.startswith("data: "):
continue
data = line[len("data: "):]
if data == "[DONE]":
break
yield json.loads(data)Calling it is straightforward:
for chunk in stream_chat(
model="glm-5.3-flash",
messages=[{"role": "user", "content": "Explain MoE in one sentence."}],
):
delta = chunk["choices"][0]["delta"].get("content", "")
print(delta, end="", flush=True)That's the whole protocol — but production needs more. Let's build it out.
Step 2: Parsing Deltas Correctly
Each SSE chunk is a JSON object shaped like:
{
"id": "chatcmpl-abc123",
"object": "chat.completion.chunk",
"model": "glm-5.3-flash",
"choices": [{"index": 0, "delta": {"content": "Mixture"}, "finish_reason": null}],
"usage": null
}The gotchas that bite people:
- `delta.content` is usually missing on the very first chunk (which carries only the role) and on the final chunk (which carries
finish_reason). Always.get("content", ""). - `choices` can be empty on the usage-only final chunk. Guard with
if chunk.get("choices"). - Tool-call deltas come under `delta.tool_calls`, not
delta.content. If you support function calling, accumulate those fragments separately.
A robust delta extractor:
def extract_delta(chunk: dict) -> str:
"""Safely pull text content from a streaming chunk."""
choices = chunk.get("choices") or []
if not choices:
return ""
delta = choices[0].get("delta", {})
return delta.get("content") or ""Step 3: A Live Console Renderer
For user-facing apps you want streaming text to render as it arrives, not print one character at a time. rich's Live display handles this elegantly:
from rich.console import Console
from rich.live import Live
from rich.text import Text
console = Console()
def stream_and_print(model: str, prompt: str) -> str:
full_text = Text()
with Live(full_text, console=console, refresh_per_second=20) as live:
for chunk in stream_chat(
model=model,
messages=[{"role": "user", "content": prompt}],
):
piece = extract_delta(chunk)
if piece:
full_text.append(piece)
live.update(full_text)
return str(full_text)Users see a smooth typing effect, and you still capture the full string for your database.
Step 4: Tracking Usage and Cost (The Important Part)
Non-streaming responses include a usage object with prompt_tokens, completion_tokens, and total_tokens. In streaming mode, that usage object arrives only on the final chunk — and only if you set stream_options.include_usage: true in the request (Step 1 includes it).
The usage chunk looks like this:
{
"choices": [],
"usage": {
"prompt_tokens": 14,
"completion_tokens": 42,
"total_tokens": 56
}
}Here's a helper that captures it and computes your real cost using per-million-token prices:
# Prices in USD per 1M tokens — check qubax.ai/models for live rates
PRICING = {
"glm-5.3-flash": {"input": 0.0156, "output": 0.052},
"deepseek-v4-flash": {"input": 0.006, "output": 0.0119},
"gpt-5.6-terra": {"input": 0.069, "output": 0.414},
}
def cost_for(model: str, prompt_tokens: int, completion_tokens: int) -> float:
rates = PRICING.get(model, {"input": 0, "output": 0})
return (prompt_tokens / 1_000_000 * rates["input"]
+ completion_tokens / 1_000_000 * rates["output"])Now wrap streaming in a function that returns both the text and the cost:
def stream_with_cost(model: str, messages: list[dict]) -> tuple[str, float]:
parts, usage = [], None
for chunk in stream_chat(model=model, messages=messages):
piece = extract_delta(chunk)
if piece:
parts.append(piece)
print(piece, end="", flush=True)
if chunk.get("usage"):
usage = chunk["usage"]
text = "".join(parts)
if usage:
cost = cost_for(model, usage["prompt_tokens"], usage["completion_tokens"])
print(f"\n\n[usage] in={usage['prompt_tokens']} "
f"out={usage['completion_tokens']} cost=${cost:.6f}")
return text, cost
return text, 0.0This is the pattern that makes model routing practical: you see exactly what each request cost, per model, in real time. Route 90% of traffic to a cheap model like deepseek-v4-flash and escalate only hard requests to gpt-5.6-terra, and your bill drops by an order of magnitude.
Step 5: Error Handling and Reconnects
Real APIs hiccup. Your client should handle three failure modes:
import time
def robust_stream(model, messages, max_retries=3):
last_error = None
for attempt in range(max_retries):
try:
return stream_with_cost(model, messages)
except requests.exceptions.ConnectionError as e:
last_error = e
wait = 2 ** attempt # exponential backoff
print(f"\n[retry] connection error, waiting {wait}s...")
time.sleep(wait)
except requests.exceptions.HTTPError as e:
status = e.response.status_code if e.response is not None else 0
# 429 = rate limit, 5xx = server error — retry these
if status in (429, 500, 502, 503, 504):
wait = 2 ** attempt
print(f"\n[retry] HTTP {status}, waiting {wait}s...")
time.sleep(wait)
last_error = e
else:
# 4xx (auth, bad request) — don't retry
raise
raise last_errorFor mid-stream failures (where you've already received partial output), the simplest robust strategy is to resume with the partial output as context: pass the accumulated text back to the model and ask it to continue. More advanced setups track the last received token and use it for provider-specific resumption headers.
Step 6: Putting It All Together
Here's the complete, copy-pasteable client:
"""qubax_stream.py — streaming chat client with live cost tracking."""
import json, time, requests
from rich.console import Console
from rich.live import Live
from rich.text import Text
QUBAX_API_KEY = "sk-qubax-..."
QUBAX_BASE_URL = "https://api.qubax.ai"
PRICING = {
"glm-5.3-flash": {"input": 0.0156, "output": 0.052},
"deepseek-v4-flash": {"input": 0.006, "output": 0.0119},
"gpt-5.6-terra": {"input": 0.069, "output": 0.414},
}
def _post_stream(model, messages, **kw):
headers = {"Authorization": f"Bearer {QUBAX_API_KEY}",
"Content-Type": "application/json"}
payload = {"model": model, "messages": messages, "stream": True,
"stream_options": {"include_usage": True}, **kw}
resp = requests.post(f"{QUBAX_BASE_URL}/v1/chat/completions",
headers=headers, json=payload, stream=True, timeout=120)
resp.raise_for_status()
for line in resp.iter_lines(decode_unicode=True):
if line and line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
yield json.loads(data)
def _delta(chunk):
ch = (chunk.get("choices") or [{}])[0]
return ch.get("delta", {}).get("content") or ""
def _cost(model, pt, ct):
r = PRICING.get(model, {"input": 0, "output": 0})
return pt / 1_000_000 * r["input"] + ct / 1_000_000 * r["output"]
def stream(model, messages, render=True):
parts, usage = [], None
console = Console()
live = Live(Text(), console=console, refresh_per_second=20) if render else None
if live:
live.start()
try:
for chunk in _post_stream(model, messages):
p = _delta(chunk)
if p:
parts.append(p)
if live:
live.update(Text("".join(parts)))
if chunk.get("usage"):
usage = chunk["usage"]
finally:
if live:
live.stop()
text = "".join(parts)
cost = _cost(model, usage["prompt_tokens"], usage["completion_tokens"]) if usage else 0
return text, cost, usage
if __name__ == "__main__":
text, cost, usage = stream(
"deepseek-v4-flash",
[{"role": "user", "content": "Write a haiku about streaming APIs."}],
)
print(f"\n[cost] ${cost:.6f} tokens={usage}")Run it and you'll see the poem type itself out, followed by the exact token count and dollar cost.
Why This Pattern Matters
Streaming isn't just a UX nicety — it's the foundation of cost-effective AI apps:
- First-token latency determines perceived speed far more than total latency does. A 5-second streamed response feels faster than a 2-second blocked one.
- Usage tracking per request is what makes model routing possible. You can't optimize what you don't measure.
- One OpenAI-compatible endpoint across all models means your routing code is literally one string change. No per-provider SDKs, no per-provider auth, no per-provider billing.
Qubax's API is designed around exactly this: one key, one endpoint, every model, usage-based pricing, and SSE streaming out of the box. Browse the full catalog at qubax.ai/models and check the API reference at qubax.ai/docs.
FAQ
What is SSE (Server-Sent Events) in the context of AI APIs?
SSE is a standard HTTP-based protocol where the server pushes data to the client over a long-lived connection. For LLM APIs, each generated token (or small group of tokens) arrives as a data: line containing JSON, letting your app render text incrementally.
Do I need a special SDK to stream from Qubax?
No. Qubax exposes a standard OpenAI-compatible /v1/chat/completions endpoint. Any HTTP client that can read a streaming response works — requests, httpx, aiohttp, curl, even fetch in the browser.
How do I get token usage in streaming mode?
Set stream_options: {"include_usage": true} in your request body. The server then sends a final chunk (with an empty choices array) containing the usage object with prompt_tokens and completion_tokens.
How do I calculate the cost of a streamed request?
Multiply prompt_tokens by your model's input price and completion_tokens by its output price (both per million tokens). The tutorial includes a cost_for() helper that does this. Check live prices at qubax.ai/models.
Can I switch models without rewriting my client?
Yes — because Qubax uses the OpenAI-compatible format across every model, changing model in the request body is the only change needed. GLM, DeepSeek, Claude, GPT, Gemini, and Qwen all share the same request/response shape.
What happens if the connection drops mid-stream?
You lose the unsent remainder. The robust pattern is to either retry the full request (if you haven't persisted any output yet) or resume by feeding the partial output back to the model and asking it to continue. The tutorial includes retry-with-backoff logic for transient failures.
Why is requests.iter_lines preferred over reading the whole response?
iter_lines(decode_unicode=True) yields each line as soon as it arrives, which is what makes streaming work — you process tokens the moment the server emits them rather than buffering the entire response first.
