Streaming with httpx

~22 min · httpx-stream, sse-parse

Level 0Tokenizer

0 XP0/54 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

Use client.stream() with iter_lines() / aiter_lines() to process SSE events in real-time.

Async Streaming (Chat Completions)

The .stream() context manager keeps the connection open. The response body is NOT read into memory — it's processed line by line. Always use timeout=None for the read timeout when streaming.

The async streaming pattern

async with client.stream("POST", url, headers=h, json=body) as response: opens the stream. Inside the block, async for line in response.aiter_lines(): yields SSE frames as they arrive. The framing (data: prefix, empty-line separator, [DONE] terminator) is yours to parse.

The single biggest mistake is reaching for response.aread() or response.json() inside the streaming block — both collect the whole body, defeating the streaming. Either iterate or accept that you're not really streaming.

Code

client.stream('POST', url, json=...)·python

import os, json, httpx, asyncio
from typing import AsyncIterator

async def stream_chat_async(
    messages: list[dict], model: str = "gpt-4o-mini",
) -> AsyncIterator[str]:
    """Yield text delta strings from a streaming response."""
    url = "https://api.openai.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
        "Content-Type": "application/json",
    }
    body = {"model": model, "messages": messages, "stream": True}
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", url, headers=headers, json=body) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line or line == "data: [DONE]":
                    continue
                if line.startswith("data: "):
                    try:
                        chunk = json.loads(line[len("data: "):])
                        content = chunk["choices"][0]["delta"].get("content", "")
                        if content:
                            yield content
                    except (json.JSONDecodeError, KeyError, IndexError):
                        pass

async def main():
    async for token in stream_chat_async([{"role": "user", "content": "Count to 5."}]):
        print(token, end="", flush=True)
    print()

asyncio.run(main())

External links

httpx streaming responses

Exercise

Stream a Responses call via raw httpx. Parse the SSE frames yourself. Print: count of response.output_text.delta events, the assembled text, and any non-text events you saw.