Wiring up python httpx sse streaming llm calls by hand gives you full control over connection lifecycle, timeouts, and retries without adopting a provider-specific SDK. This tutorial builds a minimal client that speaks Server-Sent Events to an OpenAI-compatible /v1/chat/completions endpoint and renders tokens as they land.
Prerequisites
- Python 3.9 or newer
httpxinstalled (pip install httpx==0.27.*)- An API key for any OpenAI-compatible inference service (OpenAI, a self-hosted vLLM server, or a gateway like n4n.ai)
- Familiarity with basic
requests-style calls; async knowledge helps but isn’t required
Set your key in the environment:
export LLM_API_KEY="sk-..."
export LLM_BASE_URL="https://api.openai.com/v1" # or your gateway URLWhy httpx instead of requests
requests blocks a thread and has no native async streaming. LLM tokens often trickle over 10–60 seconds; you want either cooperative concurrency or a clean sync iterator. httpx gives both Client and AsyncClient with the same streaming mental model and no extra dependencies.
What the wire format actually looks like
A streaming chat completion response is not JSON lines. It is SSE: every event is prefixed with data: and terminated by a blank line. The payload is a JSON object matching the non-streaming shape, but choices[0].delta carries incremental content instead of message.
data: {"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}
data: {"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"delta":{"content":" world"},"finish_reason":null}]}
data: {"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"delta":{},"finish_reason":"stop"}]}
data: [DONE]The client must buffer lines, ignore non-data: lines (such as event: or id:), and stop when it sees [DONE].
Sync streaming with httpx
The synchronous path is the fastest to debug. Use httpx.Client with stream=True and iterate response.iter_lines().
import os
import json
import httpx
BASE_URL = os.environ["LLM_BASE_URL"]
API_KEY = os.environ["LLM_API_KEY"]
def stream_sync(prompt: str) -> None:
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
}
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
with httpx.Client(timeout=httpx.Timeout(30.0, read=60.0)) as client:
with client.stream("POST", f"{BASE_URL}/chat/completions",
json=payload, headers=headers) as resp:
if resp.status_code != 200:
raise RuntimeError(f"HTTP {resp.status_code}: {resp.read().decode()}")
for line in resp.iter_lines():
if not line or not line.startswith("data: "):
continue
data = line[len("data: "):]
if data == "[DONE]":
break
chunk = json.loads(data)
delta = chunk["choices"][0]["delta"]
if "content" in delta:
print(delta["content"], end="", flush=True)
print() # newline after stream
if __name__ == "__main__":
stream_sync("Explain SSE in one sentence.")Expected output (truncated):
Server-Sent Events stream incremental text from a server over a single HTTP connection.Key points: client.stream opens the connection lazily; iter_lines handles line buffering. We set a read timeout longer than the total timeout because the connection stays open while tokens trickle.
Async streaming with httpx.AsyncClient
Production services usually need concurrency. The async API mirrors the sync one but uses async for.
import os
import json
import asyncio
import httpx
BASE_URL = os.environ["LLM_BASE_URL"]
API_KEY = os.environ["LLM_API_KEY"]
async def stream_async(prompt: str) -> None:
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
}
headers = {"Authorization": f"Bearer {API_KEY}"}
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0, read=60.0)) as client:
async with client.stream("POST", f"{BASE_URL}/chat/completions",
json=payload, headers=headers) as resp:
if resp.status_code != 200:
body = await resp.aread()
raise RuntimeError(f"HTTP {resp.status_code}: {body.decode()}")
async for line in resp.aiter_lines():
if not line or not line.startswith("data: "):
continue
data = line[len("data: "):]
if data == "[DONE]":
break
chunk = json.loads(data)
delta = chunk["choices"][0]["delta"]
if "content" in delta:
print(delta["content"], end="", flush=True)
print()
if __name__ == "__main__":
asyncio.run(stream_async("List three SSE pitfalls."))Expected output:
1. Line buffering errors. 2. Ignoring keep-alive timeouts. 3. Assuming order is preserved without IDs.Robustness: partial frames and errors
Real networks truncate. iter_lines hides most of this, but a provider may emit an event: error frame or a JSON object with error. Capture it:
if line.startswith("event: "):
event_type = line[len("event: "):]
continue
if line.startswith("data: ") and event_type == "error":
raise RuntimeError(f"SSE error frame: {data}")Also watch for chunk.get("error"). If the stream dies mid-token, you must decide whether to retry the whole request or resume. Stateless LLM calls are idempotent only if you don’t mind regenerating; add a client-side request_id and check the gateway’s support for resumption.
Timeouts and cancellation
httpx timeouts are per-phase. A streaming call can have a short connect timeout but a long read timeout. If the user hits Ctrl-C, AsyncClient cancels the request and closes the socket; the server may still finish generating upstream, so meter on client receipt, not assumption.
timeout = httpx.Timeout(connect=5.0, read=120.0, write=5.0, pool=5.0)Parsing raw bytes (advanced)
Sometimes iter_lines is too high-level (e.g., when providers send \r\n inconsistently). Use aiter_raw and split on \n\n:
buffer = b""
async for raw in resp.aiter_raw():
buffer += raw
while b"\n\n" in buffer:
frame, buffer = buffer.split(b"\n\n", 1)
for line in frame.splitlines():
if line.startswith(b"data: "):
data = line[len(b"data: "):]
# parse JSON, handle [DONE]This gives full control but you must handle partial frames across chunk boundaries.
Pointing at a gateway
The code above is agnostic to the backend. If you target n4n.ai, which exposes one OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is rate-limited, you only swap LLM_BASE_URL and LLM_API_KEY. The SSE frames are byte-identical, so the parser needs zero changes. You can also pass x-routing-directive headers to pin a provider; the gateway forwards cache-control hints and returns per-token usage in the final chunk.
export LLM_BASE_URL="https://api.n4n.ai/v1"
export LLM_API_KEY="your-gateway-key"No code edits required. That’s the advantage of sticking to the raw python httpx sse streaming llm contract instead of a bespoke SDK.
Testing with a local SSE mock
Before hitting a paid endpoint, stub the stream with httpx.MockTransport:
def mock_handler(request):
def body():
yield b"data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n"
yield b"data: [DONE]\n\n"
return httpx.Response(200, stream=body())
# mount transport on client for testsThis catches parser regressions offline and validates your [DONE] handling.
Production checklist
- Connection pooling: reuse a single
httpx.Client/AsyncClientacross requests. Creating a client per call exhausts file descriptors. - Backpressure: if your downstream (websocket, TUI) is slower than token generation, buffer with a bounded queue.
asyncio.Queue(maxsize=1024)prevents unbounded memory. - Usage metering: the last non-
[DONE]chunk often includesusage. Capture it for cost tracking:
if "usage" in chunk:
total_tokens = chunk["usage"]["total_tokens"]- Retries: wrap the stream in a tenacity retry limited to non-200 or connection errors. Don’t blindly retry after receiving the first token unless you can dedupe.
- Logging: redact
Authorizationheaders; logmodelandfinish_reason.
Wrapping up the client
A 30-line function is enough for most internal tools. When you need multi-turn streaming, append assistant deltas to a rolling message list and resend. The python httpx sse streaming llm pattern stays the same: open stream, iterate lines, parse data:, print delta, break on [DONE].
If you outgrow hand-rolled parsing, lift the SSE loop into a small class with an async def __aiter__ that yields str tokens. Until then, httpx and the standard library are all you need.