Streaming completions from an OpenAI-compatible endpoint keeps a single HTTP connection open and pushes incremental tokens as Server-Sent Events. The openai done marker streaming convention terminates that stream with a non-JSON data: [DONE] line, and your client must recognize it precisely. Treat the marker as a control signal, not as a data record, or you will either hang waiting for a closed socket or crash on json.loads('[DONE]').
Step 1: Inspect the raw SSE frame format
OpenAI’s streaming response is plain text over HTTP, not WebSocket. Each event is one or more data: lines followed by a blank line. A minimal successful stream looks like this:
data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"Hello"}}]}
data: {"id":"chatcmpl-1","choices":[{"delta":{"content":" world"}}]}
data: [DONE]The [DONE] token is not wrapped in quotes and is not valid JSON. It is a sentinel. Any robust client must separate the control plane (the sentinel) from the data plane (the JSON delta objects). The SSE specification also allows event: and id: lines, but OpenAI-compatible APIs omit them; you only need to handle data:.
Step 2: Open a streaming request
Use a client that exposes the raw byte stream. In Python, requests with stream=True works; in Node, fetch with response.body.getReader() is standard.
import requests
def open_stream():
resp = requests.post(
"https://api.openai.com/v1/chat/completions",
json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}], "stream": True},
headers={"Authorization": "Bearer $OPENAI_API_KEY"},
stream=True,
)
resp.raise_for_status()
return respconst resp = await fetch("https://api.openai.com/v1/chat/completions", {
method: "POST",
headers: { "Content-Type": "application/json", Authorization: `Bearer ${process.env.OPENAI_API_KEY}` },
body: JSON.stringify({ model: "gpt-4o-mini", messages: [{ role: "user", content: "hi" }], stream: true }),
});
const reader = resp.body!.getReader();Step 3: Buffer and split events correctly
TCP chunks do not respect line or event boundaries. Accumulate bytes and split on the double newline that separates SSE events. In Python, iter_lines handles line splitting but not event grouping; do that yourself.
def parse_sse(resp):
event_lines = []
for line in resp.iter_lines(decode_unicode=True):
if line is None:
continue
if line.startswith("data: "):
event_lines.append(line[6:])
elif line == "":
if event_lines:
yield "".join(event_lines)
event_lines = []
if event_lines:
yield "".join(event_lines)This yields the payload after the data: prefix for each event, whether it is JSON or the [DONE] string. In TypeScript the same logic operates on a string buffer:
let buf = "";
const decoder = new TextDecoder();
function* eventsFromBuffer() {
let idx;
while ((idx = buf.indexOf("\n\n")) !== -1) {
const event = buf.slice(0, idx);
buf = buf.slice(idx + 2);
const dataLine = event.split("\n").find(l => l.startsWith("data: "));
if (dataLine) yield dataLine.slice(6).trim();
}
}Step 4: Detect the openai done marker streaming sentinel
Consume the generator and break on the exact sentinel. Strip whitespace because some proxies append \r or trailing spaces. The openai done marker streaming protocol is unforgiving: a single missed comparison leaves the loop waiting on a dead socket.
import json
def collect(resp):
content = ""
for data_str in parse_sse(resp):
if data_str.strip() == "[DONE]":
# clean termination
return content
chunk = json.loads(data_str)
delta = chunk["choices"][0].get("delta", {})
if "content" in delta:
content += delta["content"]
raise RuntimeError("Stream ended without [DONE]")The TypeScript reader loop mirrors this:
let buf = "";
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) throw new Error("Stream ended without [DONE]");
buf += decoder.decode(value, { stream: true });
let idx;
while ((idx = buf.indexOf("\n\n")) !== -1) {
const event = buf.slice(0, idx);
buf = buf.slice(idx + 2);
const dataLine = event.split("\n").find(l => l.startsWith("data: "));
if (!dataLine) continue;
const data = dataLine.slice(6).trim();
if (data === "[DONE]") { await reader.cancel(); break; }
const chunk = JSON.parse(data);
process.stdout.write(chunk.choices[0]?.delta?.content ?? "");
}
}Step 5: Extract deltas, finish reason, and usage
Production code needs more than content. Capture finish_reason and usage if present. Some providers send usage only on the last JSON event before [DONE]; others omit it unless you pass stream_options: { include_usage: true }.
def collect_full(resp):
content = ""
finish_reason = None
usage = None
for data_str in parse_sse(resp):
if data_str.strip() == "[DONE]":
break
chunk = json.loads(data_str)
choice = chunk["choices"][0]
delta = choice.get("delta", {})
if "content" in delta:
content += delta["content"]
if choice.get("finish_reason"):
finish_reason = choice["finish_reason"]
if "usage" in chunk:
usage = chunk["usage"]
return content, finish_reason, usageStep 6: Stay robust against backend switches
Network middleboxes and gateways may change providers mid-request. If you route through an OpenAI-compatible gateway such as n4n.ai, the same openai done marker streaming contract is emitted even when the backend model is swapped due to rate limits or degradation. Your parser should not inspect the model field to decide when to stop; it should trust the sentinel.
Also guard against premature socket closure. In Python, attach an else clause to the for loop (it runs if break never executes):
else:
raise RuntimeError("Connection closed before [DONE]")In TypeScript, the reader.read() resolving done: true before you saw [DONE] is the equivalent failure; throw as shown above.
Step 7: Verify with a deterministic test
Mock the SSE source so the test never depends on network or quota. Below is a pytest snippet that validates both parsing and sentinel handling.
def test_parse_sse_done():
class FakeResp:
def iter_lines(self, decode_unicode=True):
yield "data: {\"choices\":[{\"delta\":{\"content\":\"Hi\"}}]}"
yield ""
yield "data: [DONE]"
yield ""
events = list(parse_sse(FakeResp()))
assert events[0].startswith("{")
assert events[-1].strip() == "[DONE]"
def test_collect():
class FakeResp:
def iter_lines(self, decode_unicode=True):
yield "data: {\"choices\":[{\"delta\":{\"content\":\"Hi\"}}]}"
yield ""
yield "data: [DONE]"
yield ""
assert collect(FakeResp()) == "Hi"Run pytest -q and confirm both pass. For TypeScript, feed a ReadableStream with the same string chunks to your reader loop and assert the accumulated string equals "Hi" and that reader.cancel was called.
Integration check
Against a live endpoint, set max_tokens=5 and stream a trivial prompt. You should see the connection close within a second or two and your process exit cleanly. If it hangs, you missed the sentinel.
Common pitfalls
- JSON parsing the sentinel:
json.loads("[DONE]")throws. Always compare as string first. - Line-ending assumptions: Use
splitlines()or strip\r. Windows-style\r\nwill break naiveline == ""checks if not stripped. - Multiple events per chunk: A single
read()may contain twodata:blocks. Your buffer must loop on\n\n, not just split once. - Trailing
data:with no blank line: Some proxies flush the[DONE]without a final newline. Handle residual buffer after the socket closes.
The openai done marker streaming spec is simple, but the surrounding buffering is where most client bugs live. Write the parser once, cover it with the mock test above, and never think about it again.