AI API Streaming in Python: SSE, AsyncIO, and Real-Time UIs

AI API streaming in Python uses Server-Sent Events (SSE) over AsyncIO to deliver tokens as the model generates them. The pattern matters: streaming moves perceived latency from full-generation time (2-10s) to first-token time (180-420ms) — a 5-10x UX improvement. This guide covers the OpenAI SDK streaming, raw SSE parsing, AsyncIO fan-out to multiple models, and real-time UI integration with backpressure handling.

Why Streaming Changes Everything

Without streaming, a 500-token response at 40ms/token means users watch a spinner for 20 seconds. With streaming, the first tokens render in under 400ms and content flows continuously. Perceived latency drops even though total generation time is identical.

Basic Streaming with the OpenAI SDK

The OpenAI Python SDK handles SSE parsing internally — iterate the stream and read deltas:

from openai import OpenAI

client = OpenAI(
    api_key="your-drai-key",
    base_url="https://api.dr-ai.top/v1"
)

stream = client.chat.completions.create(
    model="gpt-5-mini",
    messages=[{"role": "user", "content": "Explain SSE in one paragraph"}],
    stream=True,
    max_tokens=300,
)

full = []
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:               # None on keep-alive chunks
        full.append(delta)
        print(delta, end="", flush=True)

print("".join(full))

Two details everyone hits: delta can be None (skip, don't crash), and always set max_tokens — uncapped streams run for minutes.

Raw SSE Parsing (No SDK)

Under the hood, SSE is just text/event-stream with data: lines. Parsing it yourself matters for custom clients and debugging:

import httpx, json

async def raw_sse(url, headers, payload):
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", url, 
                                  headers=headers, json=payload) as r:
            async for line in aiter_lines(r):
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data == "[DONE]":
                    break
                chunk = json.loads(data)
                delta = (chunk["choices"][0]
                         .get("delta", {}).get("content"))
                if delta:
                    yield delta

async def aiter_lines(response):
    async for raw in response.aiter_lines():
        yield raw

AsyncIO: Concurrent Streams from Multiple Models

The killer AsyncIO pattern — stream multiple models simultaneously and take the first good answer (or race for latency):

import asyncio

async def stream_one(model, prompt):
    try:
        stream = await client.chat.completions.create(
            model=model, prompt=prompt, stream=True, timeout=8)
        return (model, stream)
    except Exception:
        return (model, None)

async def race_models(prompt, models):
    tasks = [stream_one(m, prompt) for m in models]
    # as_completed: yield whichever model starts producing first
    for coro in asyncio.as_completed(tasks):
        model, stream = await coro
        if stream:
            async for delta in stream:
                yield delta
            return  # first successful stream wins

This hedging pattern costs 2-3x tokens on the racing requests but cuts p99 latency dramatically — standard practice in latency-critical search and autocomplete.

Fan-Out: One Request, Many Consumers

Serve one upstream stream to N connected clients (team collaboration, live previews) with per-client queues and backpressure:

class StreamHub:
    def __init__(self):
        self.subscribers = []   # list[asyncio.Queue]

    def subscribe(self):
        q = asyncio.Queue(maxsize=100)  # backpressure bound
        self.subscribers.append(q)
        return q

    async def broadcast(self, delta):
        dead = []
        for q in self.subscribers:
            try:
                q.put_nowait(delta)      # drop for slow clients
            except asyncio.QueueFull:
                dead.append(q)           # or await q.put() to block
        for q in dead:
            self.subscribers.remove(q)

# FastAPI endpoint per subscriber
@app.get("/stream")
async def stream_endpoint(request):
    q = hub.subscribe()
    async def generate():
        while True:
            delta = await q.get()
            if delta is SENTINEL:
                break
            yield f"data: {json.dumps({'delta': delta})}\n\n"
    return StreamingResponse(generate())

The bounded queue is the key design choice: slow consumers get drops (chat) or disconnection (financial data) instead of unbounded memory growth.

Mid-Stream Error Recovery

Streams die at token 500 of 1,000. Handle it deliberately:

async def resilient_stream(messages, min_tokens=50):
    collected, attempts = [], 0
    while attempts < 3:
        try:
            stream = await client.chat.completions.create(
                model="gpt-5-mini", messages=messages, stream=True)
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    collected.append(delta)
                    yield delta
            return
        except Exception as e:
            attempts += 1
            if len(collected) >= min_tokens:
                # enough context: continue, don't restart
                messages.append({"role": "assistant",
                                 "content": "".join(collected)})
                messages.append({"role": "user",
                                 "content": "continue exactly where you stopped"})
            await asyncio.sleep(2 ** attempts * 0.5)
    yield "\n[generation failed after retries]"

Backpressure: Don't Outrun Your Consumer

If your UI renders slower than the model generates, the unconsumed buffer grows. For UIs, coalesce deltas — render at 30-60fps, not per-token:

import time

async def throttled_render(q, fps=30):
    buffer, last = [], 0.0
    while True:
        delta = await q.get()
        buffer.append(delta)
        now = time.monotonic()
        if now - last >= 1/fps:
            render("".join(buffer))     # batch to one paint
            buffer, last = [], now

Checklist: Production Streaming

• Always stream=True for interactive UIs; blocking for batch only
• Handle None deltas and [DONE] sentinels
• Set max_tokens and client timeouts (disable infinite read)
• Persist partial output for mid-stream recovery
• Bound every queue; decide drop-vs-block per use case
• Throttle UI rendering to frame rate
• Log time-to-first-token separately from total time

DrAI normalizes SSE across all 18+ providers — Anthropic, DeepSeek, and Gemini streams all arrive as standard OpenAI chunks, so this code works unchanged for every model. Get a free key at ai.dr-ai.top and see the SSE vs WebSocket deep-dive, latency optimization guide, and error handling patterns for the surrounding production stack.

Streaming doesn't have to be expensive — see per-model pricing on the DrAI pricing page.