Disclosure: this tutorial is published by the FlowBarAI team. AI assisted with research, drafting, and code preparation. The examples were tested locally with synthetic streams, without API credentials or paid model calls. They are not a provider benchmark or a production compatibility certification.
A streaming integration can look healthy while silently losing the last part of an answer. The HTTP request succeeded. Tokens appeared. The socket closed. None of those observations, individually, establishes that the application received a complete response.
This tutorial builds a small, testable boundary between network bytes and application success. It targets engineers diagnosing an OpenAI-compatible Chat Completions stream. For ordinary application integration, prefer the provider's maintained SDK; raw parsing is useful for gateway tests, transport adapters, and reproducing compatibility defects.
1. Separate four boundaries
Treat this as a pipeline:
HTTP body bytes -> SSE data events -> chat chunks -> application result
| Layer | Boundary | Common mistake |
|---|---|---|
| Transport | An arbitrary group of bytes | Assuming one read contains one event |
| SSE | A blank-line-terminated event | Parsing each data: line as independent JSON |
| Chat protocol | Choice deltas and termination markers | Assuming every chunk contains text |
| Application | An explicitly accepted completion state | Treating EOF or a partial answer as success |
A network read can stop in the middle of a JSON string, a CRLF pair, or a multibyte UTF-8 character. Conversely, one read can contain several complete events. Your parser should produce the same result for every partition of the same valid byte sequence.
The WHATWG SSE specification defines UTF-8 decoding; LF, CRLF, and CR line endings; comment lines; and joining multiple data fields with newlines. A final incomplete event is discarded at EOF. Removing the blank-line boundary changes the protocol.
2. Define a narrow completion contract
The example below accepts one choice (index == 0), text deltas, a successful finish_reason of stop, and a subsequent [DONE] event. Other finish reasons become explicit failures under this application's policy. Empty text is permitted; the caller can impose a separate nonempty-output rule.
The Chat Completions reference documents that requesting streamed usage can produce a final chunk with an empty choices array, before [DONE]. An interrupted request might never deliver that usage chunk. Consequently, choices == [] is not itself an error, and absent usage remains unknown rather than zero.
This is a strict application profile, not a universal model API protocol:
- It rejects tool calls, audio, refusal deltas, multiple choices, and text arriving after a stop.
- It does not implement the Responses API's typed event lifecycle.
- Usage is retained as unvalidated provider metadata, not a reconciled bill.
- It buffers text and returns it only after its completion contract passes. An interactive UI would expose provisional deltas separately and mark interrupted output as incomplete.
Do not require this exact contract from a provider that documents a different termination sequence; create and test a separate adapter instead of silently relaxing checks.
3. Parse SSE with explicit budgets
Save this complete file as stream_guard.py (Python 3.11+; standard library only):
"""Data-only SSE framing and a deliberately narrow chat text consumer."""
import json
class StreamFailure(RuntimeError):
pass
def sse_data(chunks, *, line_limit=65_536, event_limit=262_144):
line = bytearray()
parts = []
size = 0
skip_lf = False
first_line = True
for chunk in chunks:
for byte in chunk:
if skip_lf:
skip_lf = False
if byte == 10:
continue
if byte not in (10, 13):
line.append(byte)
if len(line) > line_limit:
raise StreamFailure("line_limit")
continue
skip_lf = byte == 13
raw = bytes(line)
line.clear()
if first_line:
raw = raw.removeprefix(b"\xef\xbb\xbf")
first_line = False
if not raw:
if parts:
yield b"\n".join(parts).decode("utf-8", "replace")
parts.clear()
size = 0
elif not raw.startswith(b":"):
field, _, value = raw.partition(b":")
if value.startswith(b" "):
value = value[1:]
if field == b"data":
size += len(value) + 1
if size > event_limit:
raise StreamFailure("event_limit")
parts.append(value)
# EOF is not a blank line: discard an unfinished event.
def collect_text(chunks, *, output_limit=100_000):
"""One choice, text-only, successful stop + DONE required; no retries."""
output = []
size = 0
usage = None
finished = False
for data in sse_data(chunks):
if data == "[DONE]":
if not finished:
raise StreamFailure("missing_stop")
return {"text": "".join(output), "usage": usage}
try:
obj = json.loads(data)
except ValueError:
raise StreamFailure("invalid_json") from None
if not isinstance(obj, dict):
raise StreamFailure("invalid_shape")
if obj.get("error") is not None:
raise StreamFailure("upstream_error")
choices = obj.get("choices")
if not isinstance(choices, list):
raise StreamFailure("invalid_choices")
if obj.get("usage") is not None:
if not isinstance(obj["usage"], dict):
raise StreamFailure("invalid_usage")
usage = obj["usage"]
if not choices:
continue
if (len(choices) != 1 or not isinstance(choices[0], dict)
or choices[0].get("index") != 0):
raise StreamFailure("unsupported_choices")
choice = choices[0]
delta = choice.get("delta")
if not isinstance(delta, dict):
raise StreamFailure("invalid_delta")
if any(delta.get(k) for k in
("tool_calls", "function_call", "refusal", "audio")):
raise StreamFailure("unsupported_output")
token = delta.get("content")
if token is not None and not isinstance(token, str):
raise StreamFailure("invalid_content")
if finished:
raise StreamFailure("choice_after_stop")
if token:
size += len(token)
if size > output_limit:
raise StreamFailure("output_limit")
output.append(token)
reason = choice.get("finish_reason")
if reason is not None:
if reason != "stop":
raise StreamFailure("non_stop_finish")
finished = True
raise StreamFailure("missing_done")
There are three different budgets here:
-
line_limit: 65,536 raw bytes per physical line, including ignored comment lines. -
event_limit: 262,144 bytes of accumulated data-field values plus separator allowance per event. -
output_limit: 100,000 Python string characters in the collected answer.
These are example limits to tune against your workload. Python objects and Unicode storage add overhead; a 100,000-character limit is not a 100,000-byte memory limit. The byte-by-byte implementation favors inspectability over throughput and is not presented as a performance-optimized streaming library.
The SSE layer deliberately emits only data payloads. It ignores event, id, and retry fields and has no reconnect machinery. That makes it unsuitable as a complete browser EventSource replacement or a consumer of named-event protocols without another adapter.
One leading BOM is removed. Invalid UTF-8 becomes replacement characters, matching the relevant decoding behavior; that is not a text-integrity guarantee. An application requiring strict encoding integrity should implement and test an explicit rejection policy.
Notice that ordinary role-only deltas are harmless, and usage-only chunks are retained. By contrast, the consumer raises on malformed JSON, an error envelope, or an unsupported completion shape. Its own error messages are short categories rather than echoed payloads.
4. Test the partition invariant
Save this second complete file beside the first as test_published.py:
import json
import unittest
from stream_guard import StreamFailure, collect_text, sse_data
def event(delta, reason=None):
obj = {"choices": [{"index": 0, "delta": delta,
"finish_reason": reason}]}
return ("data: " + json.dumps(obj, ensure_ascii=False) + "\r\n\r\n").encode()
STOP = event({}, "stop")
DONE = b"data: [DONE]\r\n\r\n"
class StreamTests(unittest.TestCase):
def test_every_possible_two_chunk_split(self):
wire = (b"\xef\xbb\xbf: heartbeat\r\n\r\n"
+ event({"role": "assistant"})
+ event({"content": "Hello \u4f60\U0001f680"}) + STOP
+ b'data: {"choices":[],"usage":{"total_tokens":7}}\r\n\r\n'
+ DONE)
for cut in range(len(wire) + 1):
with self.subTest(cut=cut):
result = collect_text([wire[:cut], wire[cut:]])
self.assertEqual(result["text"], "Hello \u4f60\U0001f680")
self.assertEqual(result["usage"], {"total_tokens": 7})
self.assertEqual(collect_text(bytes([b]) for b in wire)["text"], "Hello \u4f60\U0001f680")
def test_line_endings_and_multiline_data(self):
for nl in (b"\n", b"\r\n", b"\r"):
wire = nl.join([b": ping", b"data: one", b"data: two", b"", b""])
self.assertEqual(list(sse_data([wire])), ["one\n two"])
def test_partial_event_at_eof_is_not_dispatched(self):
self.assertEqual(list(sse_data([b"data: partial\n"])), [])
with self.assertRaisesRegex(StreamFailure, "^missing_done$"):
collect_text([STOP + b"data: [DONE]\n"])
def test_stop_and_done_are_both_required(self):
for wire in (STOP, event({"content": "partial"}) + DONE):
with self.subTest(wire=wire), self.assertRaises(StreamFailure):
collect_text([wire])
def test_length_finish_is_not_success(self):
with self.assertRaisesRegex(StreamFailure, "^non_stop_finish$"):
collect_text([event({"content": "cut off"}, "length") + DONE])
def test_budgets(self):
with self.assertRaisesRegex(StreamFailure, "^line_limit$"):
list(sse_data([b":abcdef"], line_limit=4))
with self.assertRaisesRegex(StreamFailure, "^event_limit$"):
list(sse_data([b"data: aa\ndata: bb\n\n"], event_limit=5))
with self.assertRaisesRegex(StreamFailure, "^output_limit$"):
collect_text([event({"content": "abc"}) + STOP + DONE], output_limit=2)
def test_absent_usage_stays_unknown(self):
self.assertIsNone(collect_text([STOP + DONE])["usage"])
if __name__ == "__main__":
unittest.main()
Run:
python -m unittest -v test_published
The expected result is 7 tests passing. Each split-point loop runs multiple subcases: it is not seven network requests. No SDK installation, secret, account, or model is needed.
The first fixture places a Chinese character and an emoji in the JSON payload, then tries every possible two-chunk split, including splits inside UTF-8 code points and CRLF pairs. It also feeds the same stream one byte at a time. The line-ending test checks LF, CRLF, and CR independently.
The negative tests matter just as much. A parser that produces the right text on one clean fixture can still misreport completion after a dropped connection.
| Received sequence | Result of this consumer |
|---|---|
Text, stop, optional usage, [DONE]
|
Accepted; usage may remain None
|
| Text, then EOF | missing_done |
stop, then EOF |
missing_done |
Text, then [DONE] without stop
|
missing_stop |
length, then [DONE]
|
non_stop_finish |
Unterminated [DONE] event |
Not dispatched; missing_done
|
| Over-budget line, event, or answer | Explicit budget failure |
During preparation, the two source files printed here passed those 7 tests locally on Python 3.11.9. A separate local regression suite added 24 tests covering error envelopes, invalid shapes, unsupported deltas, post-stop choices, and synthetic transport failures. That evidence establishes behavior for these fixtures; it does not prove live provider support, speed, billing, or operational reliability.
5. Put a transport contract around the parser
The code accepts an iterable of already-decompressed response-body byte chunks. Connect it to a raw streaming-body iterator, not an SDK iterator that already yields parsed chat objects. Do not pass HTTP chunk-size framing or compressed wire bytes into it.
Before consuming the body, validate the HTTP status and expected text/event-stream media type, allowing legitimate media-type parameters. Handle a non-stream JSON error response at that outer boundary.
Then make the operational limits explicit:
- Configure a read timeout and a total deadline. A tiny heartbeat can keep an idle timeout satisfied while the request never finishes.
- Cap total transferred bytes and concurrency outside the parser. Its retained-state budgets do not limit stream duration, total traffic, HTTP-library buffers, or simultaneous requests.
- Pull bounded-size chunks and process them incrementally. Converting the entire network iterator into a list defeats the memory model.
- On failure, close the response and mark the answer incomplete. Do not let provisional text become an accepted result merely because it was displayed.
- Normalize transport exceptions before recording them. This consumer's own categories omit raw JSON, but an HTTP client's exceptions can still contain sensitive request context.
An SSE id or a successful HTTP connection does not, by itself, establish resumable generation. Reissuing a generation request can be a new operation. Define replay and reconciliation behavior from the specific service contract rather than inheriting browser reconnect assumptions.
6. Turn captures into compatibility tests
For a real integration, prepare a small acceptance matrix for each exact model route and client version:
- Can the route stream text with the parameters you actually send?
- Which termination sequence and usage behavior does it document?
- How does your application classify timeouts, malformed events, refusals, and truncated output?
- Can a deliberately interrupted request be distinguished from an accepted completion in both the UI and stored job state?
Use approved, synthetic prompts for captures and review fixtures before storing them. Record protocol and client versions, timestamps, and outcome categories without publishing keys, private prompts, or response bodies.
For FlowBarAI specifically, consult the current integration documentation and verify the selected route's behavior separately; this article does not assert that every route supports every field or streaming mode. Our earlier multi-model migration checklist covers the surrounding configuration checks.
A useful review question for any streaming adapter is: if the same bytes arrive in a different partition, or the connection ends one event early, does the application still tell the truth about completion?
Top comments (0)