If you need to python parse streaming json llm output from a raw REST endpoint, you quickly discover that response.json() is useless—the body never terminates until the generation finishes, and you want tokens now. This tutorial builds a minimal, correct line-buffered reader for Server-Sent Events (SSE) style JSON streams using Python’s httpx, then shows the same pattern with requests.
Prerequisites
- Python 3.9+ (uses
httpxand standard libraryjson) httpxinstalled:pip install httpx- An OpenAI-compatible LLM endpoint and API key. The code below uses a placeholder
https://api.example.com/v1/chat/completions. Swap in your own, or an OpenAI-compatible gateway. - Familiarity with
dictaccess andforloops. No async required; we use the synchronous API.
How LLM streaming actually works
Most LLM APIs do not send a single JSON document. They send text/event-stream (SSE): a sequence of lines prefixed with data: , each containing a complete JSON object, terminated by a blank line. A typical chunk looks like:
data: {"id":"chatcmpl-123","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}The stream ends with data: [DONE]. Because TCP packets can split anywhere, your iter_bytes loop may yield half a line. The core problem when you python parse streaming json llm feeds is therefore line reassembly, not JSON parsing.
Step 1: Open a streaming request
Use httpx.Client with stream=True and set Accept: text/event-stream.
import httpx, json, os
URL = "https://api.example.com/v1/chat/completions"
HEADERS = {
"Authorization": f"Bearer {os.environ['API_KEY']}",
"Content-Type": "application/json",
"Accept": "text/event-stream",
}
PAYLOAD = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Say hello in 5 words."}],
"stream": True,
}
with httpx.Client(timeout=30.0) as client:
with client.stream("POST", URL, headers=HEADERS, json=PAYLOAD) as resp:
resp.raise_for_status()
# we will read from resp.iter_bytes() nextStep 2: Buffer bytes and split into lines
Maintain a buffer bytearray. On each chunk, append and split on \n. Keep the trailing partial line in the buffer.
buffer = bytearray()
for chunk in resp.iter_bytes():
buffer.extend(chunk)
while (nl := buffer.find(b"\n")) >= 0:
line = bytes(buffer[:nl]).decode("utf-8").rstrip("\r")
del buffer[:nl+1]
if line.startswith("data: "):
yield line[6:]This generator yields only the payload strings (without data: ). Blank lines and other SSE fields (event:, id:) are ignored, which is fine for LLM streams.
Step 3: Parse each JSON chunk
Now wrap the generator and parse. Handle [DONE] and JSONDecodeError defensively.
def stream_lines(resp):
buffer = bytearray()
for chunk in resp.iter_bytes():
buffer.extend(chunk)
while (nl := buffer.find(b"\n")) >= 0:
line = bytes(buffer[:nl]).decode("utf-8").rstrip("\r")
del buffer[:nl+1]
if line.startswith("data: "):
yield line[6:]
for payload in stream_lines(resp):
if payload.strip() == "[DONE]":
break
try:
obj = json.loads(payload)
except json.JSONDecodeError:
continue # skip malformed partials; real SSE shouldn't trigger this
print(obj)Expected output at this checkpoint
For a short completion you’ll see objects like:
{"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}
{"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}
{"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":" there"},"finish_reason":null}]}
{"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}Step 4: Extract deltas and usage
The useful field is choices[0].delta. Content tokens live under delta.content. Some providers also send usage in the final chunk when stream_options={"include_usage": True}.
full_text = ""
for payload in stream_lines(resp):
if payload.strip() == "[DONE]":
break
obj = json.loads(payload)
delta = obj["choices"][0].get("delta", {})
if "content" in delta:
full_text += delta["content"]
print(delta["content"], end="", flush=True)
if obj["choices"][0].get("finish_reason") == "stop":
usage = obj.get("usage")
if usage:
print(f"\nTokens: {usage}")
print("\nFULL:", full_text)Running this prints streamed tokens with no newline between them, then the aggregated string.
Step 5: Handle incomplete lines and connection drops
A robust reader must tolerate:
- Chunks that split
data:across reads (handled by line buffer) - Non-UTF8 bytes (rare;
errors="replace"is safer) - Connection closing early (catch
httpx.StreamError)
Modify the decode:
line = bytes(buffer[:nl]).decode("utf-8", errors="replace").rstrip("\r")And wrap the outer loop:
try:
for payload in stream_lines(resp):
...
except httpx.StreamError as e:
print("Stream interrupted:", e)Using requests instead of httpx
If you are stuck on requests, the pattern is identical. requests iterates response.iter_content(chunk_size=None).
import requests, json, os
with requests.post(URL, headers=HEADERS, json=PAYLOAD, stream=True) as resp:
resp.raise_for_status()
buffer = bytearray()
for chunk in resp.iter_content(chunk_size=1024):
if not chunk:
continue
buffer.extend(chunk)
while (nl := buffer.find(b"\n")) >= 0:
line = bytes(buffer[:nl]).decode("utf-8", errors="replace").rstrip("\r")
del buffer[:nl+1]
if line.startswith("data: "):
payload = line[6:]
if payload.strip() == "[DONE]":
break
obj = json.loads(payload)
print(obj["choices"][0].get("delta", {}).get("content", ""), end="")A note on gateways
When you point the same parsing code at a gateway such as n4n.ai, the SSE shape is unchanged—it exposes one OpenAI-compatible endpoint across 240+ models and forwards provider cache-control hints, so the stream_lines helper above works without modification while the gateway handles provider fallback behind the scenes.
Full runnable script
import httpx, json, os
URL = "https://api.example.com/v1/chat/completions"
HEADERS = {
"Authorization": f"Bearer {os.environ['API_KEY']}",
"Content-Type": "application/json",
"Accept": "text/event-stream",
}
PAYLOAD = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Count to 3."}],
"stream": True,
"stream_options": {"include_usage": True},
}
def stream_lines(resp):
buffer = bytearray()
for chunk in resp.iter_bytes():
buffer.extend(chunk)
while (nl := buffer.find(b"\n")) >= 0:
line = bytes(buffer[:nl]).decode("utf-8", errors="replace").rstrip("\r")
del buffer[:nl+1]
if line.startswith("data: "):
yield line[6:]
with httpx.Client(timeout=30.0) as client:
with client.stream("POST", URL, headers=HEADERS, json=PAYLOAD) as resp:
resp.raise_for_status()
full = ""
for payload in stream_lines(resp):
if payload.strip() == "[DONE]":
break
try:
obj = json.loads(payload)
except json.JSONDecodeError:
continue
delta = obj.get("choices", [{}])[0].get("delta", {})
if "content" in delta:
full += delta["content"]
print(delta["content"], end="", flush=True)
if obj.get("choices", [{}])[0].get("finish_reason") == "stop":
if "usage" in obj:
print(f"\nUsage: {obj['usage']}")
print("\nAGGREGATE:", full)Common pitfalls
- Forgetting
stream=True:httpx.Client.post(..., stream=True)is required; without it the client buffers the whole body. - Assuming one JSON per
iter_byteschunk: Neverjson.loads(chunk)directly. Always line-buffer. - Ignoring
data:prefix: Raw line includes it; parse after stripping exactly 6 characters. - Not handling
[DONE]: Some SDKs hide it; raw streams emit it explicitly. - Encoding errors crash the loop: Use
errors="replace"if you suspect non-UTF8.
The pattern above is all you need to python parse streaming json llm responses from any OpenAI-compatible endpoint, whether you self-host, call a vendor directly, or route through a gateway.