Most LLM APIs stream completions over HTTP using server-sent events (SSE). If you are building a python server-sent events llm api client from scratch, you must parse chunked frames, handle partial JSON, and survive dropped connections without buffering the entire response in memory.
Why raw SSE matters for LLM streams
REST SDKs hide streaming behind async iterators, but that abstraction breaks the moment you need custom routing, per-token metering, or provider fallback. When you call an LLM API directly, the response body is a stream of text frames prefixed with data:. Each frame carries a JSON delta. Understanding the wire format lets you debug token loss, implement cancellation, and avoid blocking the event loop.
SSE is not WebSocket. It is a unidirectional HTTP/1.1 or HTTP/2 stream where the server keeps the connection open and sends lines terminated by \n\n. The client reads incrementally. For LLM outputs, this maps perfectly to token generation: the server emits a delta every few milliseconds.
Choose your HTTP client: httpx vs requests
Python gives you two mainstream synchronous options: requests and httpx. requests is ubiquitous but its streaming support is limited to iterating over response.iter_lines(), which can block and does not handle async cancellation cleanly. httpx supports both sync and async streaming with proper backpressure and is the better choice for new code.
requests limitations
With requests, you set stream=True and iterate. The library buffers underlying chunks based on iter_content block size, but iter_lines splits on newlines. If a frame spans TCP segments, you still get complete lines because iter_lines reassembles until newline. However, requests does not support HTTP/2 without extra config, and timeouts on slow streams are awkward.
httpx advantages
httpx defaults to connection pooling, supports HTTP/2, and exposes response.aiter_lines() in async or response.iter_lines() in sync. It also lets you cancel a stream by closing the connection from the client side, which propagates as a generator exit. For LLM streaming, that matters when a user hits stop.
Open a streaming connection
Use the OpenAI-compatible chat completions endpoint as the canonical example. The request body sets "stream": true. Send with httpx and iterate lines:
import httpx
import json
URL = "https://api.openai.com/v1/chat/completions"
HEADERS = {"Authorization": "Bearer $API_KEY", "Content-Type": "application/json"}
PAYLOAD = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Explain SSE"}],
"stream": True,
}
with httpx.Client(timeout=httpx.Timeout(60.0, read=30.0)) as client:
with client.stream("POST", URL, headers=HEADERS, json=PAYLOAD) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
if not line:
continue
print(line)The read timeout applies between chunks. If the model goes silent for more than 30 seconds, the client raises httpx.ReadTimeout. Tune this based on expected inter-token latency.
Parse SSE frames correctly
SSE frames are separated by a blank line. A frame consists of lines like field: value. The only field LLM APIs use consistently is data. Ignore event, id, and retry unless you implement resumption.
Strip the data: prefix (note the space). If the payload is [DONE], the stream is finished. Otherwise, parse JSON:
def parse_sse_line(line: str):
if not line.startswith("data:"):
return None
payload = line[len("data:"):].lstrip()
if payload == "[DONE]":
return None # sentinel
try:
return json.loads(payload)
except json.JSONDecodeError:
# partial frame or comment, skip
return NoneIn practice, servers send one JSON object per data: line. OpenAI-compatible APIs nest the token delta at choices[0].delta.content. Extract it:
for line in resp.iter_lines():
obj = parse_sse_line(line)
if obj is None:
continue
delta = obj["choices"][0].get("delta", {})
if "content" in delta:
yield delta["content"]Handle multi-line data fields
The SSE spec allows a data field to span multiple lines; each line is concatenated with \n. LLM APIs rarely do this, but a robust client should accumulate lines until a blank line appears. Implement a small state machine:
def iter_sse_events(lines):
data_lines = []
for line in lines:
if line == "":
if data_lines:
yield "\n".join(data_lines)
data_lines = []
continue
if line.startswith("data:"):
data_lines.append(line[len("data:"):].lstrip())
if data_lines:
yield "\n".join(data_lines)This guards against a provider that splits a large JSON across TCP packets with newlines (non-conformant but observed in proxies).
Detect stream termination and errors
A normal stream ends with data: [DONE] followed by a blank line and connection close. An error mid-stream often appears as a non-200 status or a JSON frame containing "error". Check resp.status_code before iterating, but also watch for error objects inside the stream:
if "error" in obj:
raise RuntimeError(obj["error"])If the connection drops prematurely (httpx.StreamError), you must decide whether to retry. For idempotent prompts, re-issuing the request wastes tokens. Prefer resuming from the last consumed id if the API supports it; most LLM APIs do not, so you log and surface to caller.
Implement reconnection and backoff
SSE defines a retry: field for server-suggested reconnect delay. Few LLM gateways send it. Client-side, use exponential backoff with jitter when you get a 429 or 503 on the initial POST. Once streaming has started, do not auto-reconnect silently; the partial completion is lost. Surface the interruption.
import time, random
def post_with_backoff(client, url, headers, payload, max_attempts=3):
for attempt in range(max_attempts):
try:
resp = client.post(url, headers=headers, json=payload)
resp.raise_for_status()
return resp
except httpx.HTTPStatusError as e:
if e.response.status_code in (429, 503):
sleep = (2 ** attempt) + random.random()
time.sleep(sleep)
continue
raise
raise RuntimeError("exhausted retries")Manage backpressure and cancellation
If your downstream (websocket to browser, database writer) is slower than token arrival, unbounded queuing will exhaust memory. In async httpx, async for line in resp.aiter_lines() naturally applies backpressure because the coroutine pauses when you await your sink. In sync code, push to a queue.Queue(maxsize=100) and drop or block.
Cancellation: wrap the stream in a context manager and call resp.close() when a shutdown signal arrives. The generator will raise httpx.StreamClosed on next read; catch it and exit.
Using a unified gateway
When you route through a single OpenAI-compatible endpoint like n4n.ai, which fronts 240+ models with automatic fallback when a provider is degraded, the SSE frames are identical to the native API. You still parse data: lines the same way, but you avoid writing per-provider retry and model-availability logic. The gateway forwards provider cache-control hints and meters per-token usage, so your client can trust the stream to be continuous across underlying provider switches. That said, your parsing code should not assume the model field stays constant mid-stream; inspect obj["model"] if you bill by model.
Common pitfalls
- Assuming
iter_linessplits on\r\nonly. SSE uses\nor\r\n.httpxnormalizes to\n. - Buffering entire response with
resp.text. Never call.textor.json()on a streaming response; it reads to completion. - Ignoring
usagein the final frame. Some APIs sendchoices: []withusagestats in the last data frame before[DONE]. Capture it for cost tracking. - Thread safety. Sharing a
httpx.Clientacross threads is safe; sharing a streaming response is not. - TLS session reuse. Pool clients at process level to avoid handshake overhead per prompt.
Minimal async example
For completeness, here is an async coroutine that yields tokens and tracks finish reason:
import httpx, json
async def stream_tokens(url, headers, payload):
async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, read=30.0)) as client:
async with client.stream("POST", url, headers=headers, json=payload) as resp:
resp.raise_for_status()
async for line in resp.aiter_lines():
if not line.startswith("data:"):
continue
data = line[len("data:"):].lstrip()
if data == "[DONE]":
break
obj = json.loads(data)
if "error" in obj:
raise RuntimeError(obj["error"])
delta = obj["choices"][0].get("delta", {})
if "content" in delta:
yield delta["content"]
if obj["choices"][0].get("finish_reason"):
breakThis pattern has shipped in production services handling millions of LLM calls. The core lesson: treat the python server-sent events llm api stream as a line protocol, not a magic SDK method, and you keep full control over latency, cost, and failure modes.