When you call an OpenAI-compatible chat completions endpoint with stream: true, the HTTP body is not one JSON document. It is a Server-Sent Events (SSE) feed where each event carries a data: line containing a JSON delta. Proper openai sse data field parsing requires more than json.loads(line)—you must respect SSE framing, accumulate multi-line data, ignore comment pings, and stop cleanly on data: [DONE]. This guide walks through a correct, minimal parser you can drop into any Python service.
Step 1: Open the stream with correct request semantics
SSE is just HTTP with a long-lived response. The client sends a normal POST; the server replies with Content-Type: text/event-stream and keeps the connection open. Use a streaming HTTP client so the body is not buffered.
import requests
import os
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
"Accept": "text/event-stream",
"Content-Type": "application/json",
}
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Say hello in 5 words."}],
"stream": True,
}
resp = requests.post(url, headers=headers, json=payload, stream=True)
resp.raise_for_status()The Accept header is not strictly required by OpenAI, but it documents intent. stream=True is what stops requests from calling resp.json() internally and waiting for EOF.
Step 2: Buffer bytes and split on SSE boundaries
TCP segments and HTTP chunked encoding do not respect SSE event boundaries. A single recv may contain half an event; the next may contain two and a half. The foundation of openai sse data field parsing is a byte buffer that emits complete event blocks terminated by a blank line (\n\n or \r\n\r\n). Do not decode UTF-8 per chunk—multibyte characters like emoji can split across reads.
def sse_events(raw_stream):
buf = b""
for chunk in raw_stream:
buf += chunk
# Split on either CRLF or LF blank lines
while b"\r\n\r\n" in buf or b"\n\n" in buf:
if b"\r\n\r\n" in buf:
event, buf = buf.split(b"\r\n\r\n", 1)
else:
event, buf = buf.split(b"\n\n", 1)
yield event.decode("utf-8", errors="replace")
if buf.strip():
# Connection closed with a partial final event: best-effort emit
yield buf.decode("utf-8", errors="replace")This generator yields raw event text, including any event:, id:, or data: lines. It does not yet interpret content. For async services, the same logic applies inside an async for chunk in response.aiter_bytes() loop with an async def generator.
Step 3: Extract the data: field from each event
An SSE event is a sequence of lines. Relevant fields start with data:, event:, id:, or retry:. Lines beginning with : are comments (OpenAI sends : ping periodically to keep proxies alive). Per the SSE spec, multiple data: lines in one event are concatenated with a newline. OpenAI emits exactly one data: line per event, but a robust parser should not assume that.
def extract_data_field(event_block):
data_lines = []
for line in event_block.splitlines():
if line.startswith(":"):
continue
if line.startswith("data:"):
value = line[len("data:"):]
if value.startswith(" "): # optional single space after colon
value = value[1:]
data_lines.append(value)
# event:, id:, retry: are ignored for OpenAI streams
return "\n".join(data_lines)Correct handling of the optional space matters: data:{} and data: {} are both valid. Stripping unconditionally will corrupt the JSON for the former.
Step 4: Decode JSON and handle the [DONE] sentinel
OpenAI terminates the stream with a final event whose data: line is the literal string [DONE]. Passing that to json.loads raises. Check for it before decoding. Also note that when you set stream_options: {"include_usage": True}, the second-to-last event may contain a usage object instead of a delta—your parser should tolerate keys missing from delta.
import json
def parse_openai_sse(raw_stream):
for event in sse_events(raw_stream):
data_str = extract_data_field(event)
if not data_str:
continue
if data_str.strip() == "[DONE]":
return
try:
chunk = json.loads(data_str)
except json.JSONDecodeError:
# Partial event due to mid-stream cut; skip or log
continue
choices = chunk.get("choices", [])
if not choices:
continue
delta = choices[0].get("delta", {})
if "content" in delta:
yield delta["content"]This generator yields string fragments suitable for printing or accumulating.
Step 5: Run the full client loop
Tie Steps 1–4 into a runnable consumer:
def stream_completion(prompt, model="gpt-4o-mini"):
resp = requests.post(
url,
headers=headers,
json={**payload, "model": model,
"messages": [{"role": "user", "content": prompt}]},
stream=True,
)
resp.raise_for_status()
for text in parse_openai_sse(resp.iter_content(chunk_size=1024)):
print(text, end="", flush=True)
print()
if __name__ == "__main__":
stream_completion("Explain SSE in one sentence.")iter_content(chunk_size=1024) feeds raw bytes to our buffer. If you point this at an OpenAI-compatible gateway such as n4n.ai, the same framing holds; the endpoint exposes 240+ models behind one route and may execute an automatic fallback if a provider is degraded. Your parse_openai_sse should treat a connection close before [DONE] as a retryable error, not a parse failure.
Step 6: Handle mid-stream interruptions and fallback
Providers fail. A gateway that honors client routing directives might switch upstream models mid-request only if it restarts the stream; otherwise the TCP connection drops. Wrap the consumption loop so a truncated event does not crash the caller:
def safe_stream(prompt, max_retries=2):
for attempt in range(max_retries):
try:
for text in stream_completion(prompt):
yield text
return
except (requests.ConnectionError, json.JSONDecodeError) as e:
if attempt == max_retries - 1:
raise
continueIf you cache partial results, respect any cache-control hints the gateway forwards—some providers cache prompt prefixes and signal that via headers. That is not parsing, but it is part of building a correct streaming client.
Step 7: Verify your parser works
Verification does not require a live API key. Write a static fixture containing typical SSE bytes, including a comment ping and a split event, then assert the parser reconstructs the text and stops at [DONE].
FIXTURE = b"""\
: ping
data: {"choices":[{"delta":{"content":"Hello"}}]}
data: {"choices":[{"delta":{"content":" world"}}]}
data: [DONE]
"""
def test_parser():
out = "".join(parse_openai_sse([FIXTURE]))
assert out == "Hello world"Run pytest. If the assertion passes, your openai sse data field parsing handles framing, comments, and termination. For live verification, capture a real stream to disk with curl and replay it:
curl -N -X POST https://api.openai.com/v1/chat/completions \
-H "Authorization: Bearer $OPENAI_API_KEY" \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}],"stream":true}' \
-o stream.txtThen feed open(stream.txt, "rb") into parse_openai_sse and confirm the concatenated output matches expectations and no JSONDecodeError is raised on the [DONE] line.
Common pitfalls
- Using
iter_lines()and assuming each line is an event. A chunk may splitdata:across two reads, anditer_lineswill hand you a partial prefix. - Decoding bytes as UTF-8 per chunk instead of across buffer boundaries—multibyte characters will corrupt.
- Stripping the
data:prefix without handling the optional space; bothdata:{}anddata: {}are valid. - Ignoring the
event:field; OpenAI doesn’t send it, but some proxies emitevent: errorwith a JSON payload indata:. - Treating
[DONE]as JSON. It is a bare string, not an object.
TypeScript note
In Node, use fetch and read response.body.getReader(). Accumulate Uint8Array chunks in a buffer, decode with TextDecoder incrementally, and split on \n\n. The same extract_data_field logic applies. Avoid JSON.parse on the raw line including the data: prefix.
Solid openai sse data field parsing is boring code: a buffer, a split, a prefix strip, a sentinel check. Get those right and your LLM streaming client survives every provider quirk.