Streaming a Response Without Blocking Your App

A streamed model response is a sequence of small JSON documents delivered as server-sent events. The Python side of it is fifteen lines. The part that takes an afternoon is that four separate buffers, in four separate pieces of software, will each happily hold your tokens until the response is complete.

What is actually on the wire

Set "stream": true and the response comes back with Content-Type: text/event-stream and a body that arrives in pieces. Each piece is one or more lines. A frame is a line beginning data:, a blank line separates frames, and a line beginning : is a comment, used by many servers as a keep-alive ping.

data: {"id":"chatcmpl-1","choices":[{"index":0,"delta":{"role":"assistant"}}]}

data: {"id":"chatcmpl-1","choices":[{"index":0,"delta":{"content":"The"}}]}

data: {"id":"chatcmpl-1","choices":[{"index":0,"delta":{"content":" sky"}}]}

data: {"id":"chatcmpl-1","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

Three things about that are worth internalising. The first frame often carries only {"role": "assistant"} and no text, so code that assumes delta["content"] exists crashes on the first frame. The last content-bearing frame is followed by one carrying finish_reason and an empty delta. And [DONE] is not JSON — feeding it to json.loads raises JSONDecodeError: Expecting value, which is the single most common error message from a first streaming implementation.

A generator, not a callback

Make the stream a generator function. The caller then decides what to do with each fragment — print it, push it down a socket, accumulate it — and the network code has no idea which. This is the difference between a function you can reuse and one you copy and edit each time.

# stream.py
import json
import os
from collections.abc import Iterator

import httpx

BASE_URL = os.environ["LLM_BASE_URL"].rstrip("/")
API_KEY = os.environ["LLM_API_KEY"]

TIMEOUT = httpx.Timeout(connect=5.0, read=90.0, write=10.0, pool=5.0)


def stream_chat(messages: list[dict], model: str) -> Iterator[str]:
    """Yield text fragments as they arrive. Raises on a non-2xx status."""
    payload = {"model": model, "messages": messages, "stream": True}
    with httpx.Client(timeout=TIMEOUT) as client:
        with client.stream(
            "POST",
            f"{BASE_URL}/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json=payload,
        ) as response:
            if response.status_code >= 400:
                response.read()            # body is lazy on a streamed response
                response.raise_for_status()

            for line in response.iter_lines():
                line = line.strip()
                if not line or line.startswith(":"):
                    continue
                if not line.startswith("data:"):
                    continue
                data = line[5:].strip()
                if data == "[DONE]":
                    return
                try:
                    frame = json.loads(data)
                except json.JSONDecodeError:
                    continue               # a truncated frame is not fatal
                for choice in frame.get("choices", []):
                    text = choice.get("delta", {}).get("content")
                    if text:
                        yield text

The response.read() before raise_for_status() is the non-obvious line. On a streamed response httpx has not fetched the body yet, so an error raised without reading it first gives you a status code and no message — and the message is where the provider tells you which of your parameters it rejected.

for fragment in stream_chat([{"role": "user", "content": "Count to five."}],
                            model="openai/gpt-4o-mini"):
    print(fragment, end="", flush=True)
print()

The async version

The async client has the same method names with a prefixes on the iterators. If you are serving concurrent requests, this is the one you want, because the synchronous client holds a thread for the entire minute a long answer takes.

from collections.abc import AsyncIterator

async def astream_chat(messages: list[dict], model: str) -> AsyncIterator[str]:
    payload = {"model": model, "messages": messages, "stream": True}
    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        async with client.stream(
            "POST",
            f"{BASE_URL}/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json=payload,
        ) as response:
            if response.status_code >= 400:
                await response.aread()
                response.raise_for_status()
            async for line in response.aiter_lines():
                line = line.strip()
                if not line.startswith("data:"):
                    continue
                data = line[5:].strip()
                if data == "[DONE]":
                    return
                try:
                    frame = json.loads(data)
                except json.JSONDecodeError:
                    continue
                for choice in frame.get("choices", []):
                    text = choice.get("delta", {}).get("content")
                    if text:
                        yield text

Creating a client per call, as both of these do, costs a TCP and TLS handshake every time. For anything that runs in a loop, build one client at startup and reuse it — forty requests at once with asyncio shows the connection-pool settings that go with it.

The delta fields other than content

delta.content is the one every example handles. A real stream carries several others, and ignoring them is the difference between a demo and something that works when the model calls a tool.

Tool calls are the case that needs real code, because the accumulator is not obvious. Each frame contributes a fragment of one call, identified by its index in the list:

from collections import defaultdict

def accumulate_tool_calls(frames) -> list[dict]:
    """Assemble streamed tool-call fragments into complete calls."""
    partial: dict[int, dict] = defaultdict(
        lambda: {"id": "", "name": "", "arguments": ""}
    )
    for frame in frames:
        for choice in frame.get("choices", []):
            for call in choice.get("delta", {}).get("tool_calls", []) or []:
                slot = partial[call.get("index", 0)]
                if call.get("id"):
                    slot["id"] = call["id"]
                function = call.get("function") or {}
                if function.get("name"):
                    slot["name"] += function["name"]
                if function.get("arguments"):
                    slot["arguments"] += function["arguments"]
    return [partial[i] for i in sorted(partial)]

Two details that cost an afternoon each if you guess them. The name is concatenated rather than assigned, because it too can arrive in pieces on some endpoints. And arguments is a JSON string built from fragments, so it is only valid once finish_reason == "tool_calls" has been seen — parsing it early gives you a JSONDecodeError on a partial object every time.

Four places your tokens get stuck

“It works in the terminal but the browser shows nothing until the end” is always one of these four, and they are worth checking in this order because that is roughly how often each is the culprit.

A quick way to tell your code from the infrastructure: run curl -N against your endpoint. -N disables curl’s own buffering, so if curl trickles and the browser does not, the problem is between the two and not in your Python.

Getting the token counts out of a stream

A non-streamed response ends with a usage object. A streamed one historically did not, which is why so much streaming code has no cost telemetry. Most OpenAI-compatible endpoints now support asking for it:

payload = {
    "model": model,
    "messages": messages,
    "stream": True,
    "stream_options": {"include_usage": True},
}

When it is honoured, a final frame arrives after the last content frame, with an empty choices list and a populated usage object. Handle it by looking for usage on every frame rather than by position:

usage = frame.get("usage")
if usage:
    total_in = usage.get("prompt_tokens")
    total_out = usage.get("completion_tokens")

How a stream fails, and how to notice

The dangerous property of a stream is that it can fail after it has succeeded. You have already sent a 200 and forty tokens to the client when the upstream connection drops, so there is no status code left to change and the user sees a plausible answer that stops early.

• Track termination explicitly. Set a flag when you see [DONE] or a finish_reason. If the loop exits without either, the stream was cut — log it and tell the caller, rather than returning the partial text as if it were an answer.
• The read timeout applies between chunks, not to the whole response. httpx’s read timeout resets on every chunk received, so a ninety-second read timeout does not cap a five-minute answer. If you need a wall-clock cap, measure it yourself with time.monotonic() in the loop and break.
• Retrying a stream is not free. If it dies at token 300 you have already paid for 300 output tokens and a restart pays for them again. Retry the whole call only if nothing has been delivered to the user yet; after that, degrade instead — graceful degradation and what retries cost both cover this trade.
• Do not parse JSON output incrementally by hand. A half-arrived JSON object is not repairable in the general case. If you are streaming structured output, either buffer to the end and parse once, or use a parser built for prefixes — streaming JSON parsing explains why the naive version bites.