Parsing SSE in Python When UTF-8 Characters Cross Chunk Boundaries

Define the boundaries before writing the parser

Name the stages explicitly: transport chunks are byte sequences returned by a socket or test fixture; decoded text is Unicode produced by an incremental decoder; lines are terminated according to the SSE grammar; an SSE event is assembled from fields until an empty line; and a provider payload is whatever the application chooses to parse from the event data. None of these boundaries should be inferred from a read size. A read can end inside a UTF-8 code point, between the carriage return and line feed of CRLF, or in the middle of a JSON value.

Decode bytes incrementally

Create one UTF-8 incremental decoder for the lifetime of the stream. Pass each transport chunk with final=False, and call it with final=True only when the byte stream is known to have ended. Decode errors should be reported as transport or text-decoding failures, not converted into partial application data. The incremental decoder preserves incomplete multibyte sequences between calls.

import codecs


def decode_chunks(chunks):
    decoder = codecs.getincrementaldecoder("utf-8")(
        errors="strict"
    )
    for chunk in chunks:
        if not isinstance(chunk, (bytes, bytearray, memoryview)):
            raise TypeError("transport chunks must be bytes-like")
        text = decoder.decode(bytes(chunk), final=False)
        if text:
            yield text

    tail = decoder.decode(b"", final=True)
    if tail:
        yield tail

A parser that keeps state across reads

CR is itself a valid line ending: consume it immediately, then skip one following LF, even when it arrives in the next chunk. This avoids delaying a complete CR-terminated event while waiting for more bytes. The initial UTF-8 BOM is ignored once. This deliberately limited parser handles data, event and comments; it is not an EventSource replacement and does not implement id, retry or reconnect behavior. Line and event limits count decoded characters, not incoming bytes: the caller must also cap the size of each byte chunk and consume returned events promptly.

import codecs


class ParseLimitError(ValueError):
    pass


class IncompleteStream(ValueError):
    pass


class SSEParser:
    def __init__(self, *, max_line_chars=8192, max_event_chars=1_048_576):
        self._decoder = codecs.getincrementaldecoder("utf-8")()
        self._line = []
        self._first_char = True
        self._pending_cr = False
        self._data = []
        self._event = None
        self._max_line_chars = max_line_chars
        self._max_event_chars = max_event_chars
        self._event_chars = 0

    def feed(self, chunk):
        text = self._decoder.decode(chunk, final=False)
        events = []
        self._consume_text(text, events)
        return events

    def close(self):
        events = []
        text = self._decoder.decode(b"", final=True)
        self._consume_text(text, events)

        if self._line:
            raise IncompleteStream("stream ended within a line")
        if self._data or self._event is not None:
            raise IncompleteStream("stream ended before the blank delimiter")
        return events

    def _consume_text(self, text, events):
        for char in text:
            if self._first_char:
                self._first_char = False
                if char == "\ufeff":
                    continue
            if self._pending_cr:
                self._pending_cr = False
                if char == "\n":
                    continue

            if char == "\r":
                self._emit_line(events)
                self._pending_cr = True
            elif char == "\n":
                self._emit_line(events)
            else:
                self._line.append(char)
                if len(self._line) > self._max_line_chars:
                    raise ParseLimitError("line size limit exceeded")

    def _emit_line(self, events):
        line = "".join(self._line)
        self._line.clear()

        if line == "":
            if self._data:
                events.append({
                    "event": self._event or "message",
                    "data": "\n".join(self._data),
                })
            self._data.clear()
            self._event = None
            self._event_chars = 0
            return

        if line.startswith(":"):
            return

        field, separator, value = line.partition(":")
        if not separator:
            value = ""
        elif value.startswith(" "):
            value = value[1:]

        if field == "data":
            projected = self._event_chars + len(value)
            if self._data:
                projected += 1
            if projected > self._max_event_chars:
                raise ParseLimitError("event size limit exceeded")
            self._data.append(value)
            self._event_chars = projected
        elif field == "event":
            self._event = value

Empty data is not the same as no data

SSE dispatch checks whether the data buffer is empty before removing the final appended LF. Therefore data: followed by a blank line produces a message with an empty string, whereas a comment-only block produces no message. Multiple data fields join with LF, one space after the first colon is removed, and an empty event field falls back to message. The parser above resets per-event state at each blank delimiter.

parser = SSEParser()
events = parser.feed(b"data:\n\n")
events += parser.close()
# Expected by this fixture: [{"event": "message", "data": ""}]

Keep event state separate from provider payload state

After an SSE event is framed, the application may attempt to parse its data as JSON. A JSON decode failure is different from a UTF-8 failure and different again from a valid JSON object that lacks the fields expected by a provider-specific protocol. Do not treat an arbitrary sentinel, an empty data field, or a provider-specific finish marker as a universal SSE completion rule. The parser should expose framed events; a separate adapter can interpret the payload contract when that contract is documented.

import json


def parse_json_data(event):
    try:
        return json.loads(event["data"])
    except json.JSONDecodeError as exc:
        raise ValueError("SSE event data is not valid JSON") from exc

Choose and document a stricter failure profile

The parser intentionally uses strict UTF-8 decoding and raises IncompleteStream for a final unterminated line or event. That is a local diagnostic policy, not the browser EventSource algorithm: the SSE standard discards an unfinished event at EOF. ParseLimitError stops oversized lines or data fields without truncating them. After an exception or close, discard the parser instance rather than trying to resume it. These limits do not bound an arbitrarily large input chunk or a caller that retains every returned event.

# Line and event limits apply independently.
parser = SSEParser(max_line_chars=256, max_event_chars=4096)
# Feed bounded chunks, consume events immediately, then call close once.

Build deterministic offline fixtures

Construct byte fixtures directly and split them at deliberate positions. Include a multibyte UTF-8 character split across reads, a CRLF split between reads, multiline data fields, comment lines, two complete events in one read, an empty event, invalid UTF-8, invalid JSON, an oversized line or data field, and a stream ending before its blank event separator. Run the same fixtures with different chunk partitions where applicable. These cases validate parser state transitions only; they do not establish network behavior, provider compatibility, model support, latency, throughput, or production readiness.

raw = "\ufeff: keepalive\r\ndata: café\r\n\r\n".encode("utf-8")
expected = [{"event": "message", "data": "café"}]
for split in range(len(raw) + 1):
    parser = SSEParser()
    actual = parser.feed(raw[:split])
    actual += parser.feed(raw[split:])
    actual += parser.close()
    assert actual == expected
# Offline assertion only, not a live-provider check.

State the reference boundary and limits

Use the HTML Standard's Server-Sent Events section as the primary reference for line, field, comment, data-buffer, and dispatch behavior, and Python's codecs documentation as the primary reference for incremental decoder interfaces. The implementation should document any deliberate deviations, such as a chosen policy for incomplete final input or application-specific size limits. A local parser harness can demonstrate deterministic handling of supplied bytes, but it cannot verify an upstream service's framing, payload schema, retry behavior, proxy behavior, or model compatibility.

Prompt template

Write an original English engineering guide based on the supplied brief. Explain the distinction between transport chunks, incremental UTF-8 decoding, decoded lines, SSE events, and provider-specific payloads. Cover data fields, comments, empty-line dispatch, CRLF split across reads, multiline data, two events in one read, incomplete final input, decode errors, JSON errors, and bounded buffers. Include concise Python examples that are illustrative only and must not be described as tested. Propose an offline deterministic fixture harness with network access disabled and no API key or model request. Do not invent prices, timings, benchmarks, promotions, supported models, compatibility claims, or production test results. Refer to the HTML Server-Sent Events standard and Python incremental codec documentation as primary public references without adding unsupported citations. Make uncertainty and parser-policy choices explicit.