Python Streaming with httpx

The pattern

Three things every Python Ollama streaming client gets right:

timeout=None — first-token latency on a cold model can be tens of seconds. The default httpx timeout will kill your stream mid-warm-up. Set it to None and rely on the server.
iter_lines() — let the HTTP client buffer until newlines, then yield complete lines. Don't read raw bytes and reassemble lines yourself.
flush=True when printing tokens — otherwise stdio buffers swallow the streaming feel.

Where this fits in production

This is the building block for streaming endpoints in your own backend. cwkPippa's Ollama vessel uses this exact pattern — stream NDJSON from Ollama, transform each chunk into a universal StreamChunk, write it to JSONL ground truth, then yield it to the frontend.

Common mistakes

Forgetting timeout=None. Default timeouts kill you on cold starts.
Buffering by character instead of by line. iter_text() chunks at network boundaries; iter_lines() handles the boundary problem for you.
Not handling the empty final chunk. done: true arrives with empty content. Break before printing.

Code

Production-grade streamer·python

import httpx
import json
from typing import Iterator

def stream_chat(model: str, messages: list[dict],
                base_url: str = "http://localhost:11434",
                **options) -> Iterator[dict]:
    """Stream NDJSON chunks from /api/chat. Yields parsed dicts."""
    payload = {"model": model, "messages": messages, "stream": True}
    if options:
        payload["options"] = options

    with httpx.stream("POST", f"{base_url}/api/chat",
                      json=payload, timeout=None) as r:
        r.raise_for_status()
        for line in r.iter_lines():
            if not line:
                continue
            yield json.loads(line)

# Usage — print tokens as they arrive, capture metrics
full = ""
for chunk in stream_chat("qwen2.5:7b",
                          [{"role": "user", "content": "Explain MLX in 3 sentences."}]):
    if chunk.get("done"):
        eval_count = chunk.get("eval_count", 0)
        eval_dur_ns = chunk.get("eval_duration", 1)
        tps = eval_count / (eval_dur_ns / 1e9)
        print(f"\n[{eval_count} tokens @ {tps:.1f} tok/s]")
        break
    delta = chunk.get("message", {}).get("content", "")
    full += delta
    print(delta, end="", flush=True)

Async version with httpx.AsyncClient·python

import httpx, json
from typing import AsyncIterator

async def stream_chat_async(model: str, messages: list[dict]) -> AsyncIterator[dict]:
    """Async streaming — preferred for FastAPI / asyncio servers."""
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream(
            "POST",
            "http://localhost:11434/api/chat",
            json={"model": model, "messages": messages, "stream": True},
        ) as r:
            r.raise_for_status()
            async for line in r.aiter_lines():
                if not line:
                    continue
                yield json.loads(line)

# Usage in async context
async def main():
    async for chunk in stream_chat_async(
        "qwen2.5:7b",
        [{"role": "user", "content": "What is GGUF?"}],
    ):
        if chunk.get("done"):
            break
        print(chunk["message"]["content"], end="", flush=True)

Python Streaming with httpx

The pattern

Where this fits in production

Common mistakes

Code

External links

Exercise

Progress

Comments 0