Streaming

SseEvent

SseEvent is a thin dict subclass with two typed properties:

class SseEvent(dict):
    @property
    def event(self) -> str: ...
    @property
    def data(self) -> Any: ...

It always carries two keys:

SseEvent subclasses dict for backward compatibility — code written against pre-0.3.0 SDKs that did event["data"]["..."] continues to work. The typed event.event and event.data accessors are equivalent.

from recall import SseEvent

event["data"]["stage"]
event.data["stage"]

The seven server event types

The Recall server emits seven distinct event types. Pattern-match on event.event — the payload shape varies per type.

Iteration patterns

for event in recall.pipelines.write_stream(messages=msgs):
    match event.event:
        case 'pipeline_init':
            print('trace:', event.data['trace_id'])
        case 'stage':
            print(event.data['stage'], event.data['duration_ms'], 'ms')
        case 'llm_call':
            print('LLM:', event.data['model'], event.data['prompt_tokens'])
        case 'result':
            print('Memories:', event.data['memoryIds'])
        case 'error':
            raise RuntimeError(event.data['message'])
        case 'done':
            print('clean shutdown')

async for event in recall.pipelines.write_stream(messages=msgs):
    if event.event == 'pipeline_init':
        print('trace:', event.data['trace_id'])
    elif event.event == 'stage':
        print(event.data['stage'], event.data['duration_ms'], 'ms')
    elif event.event == 'llm_call':
        print('LLM:', event.data['model'])
    elif event.event == 'result':
        print('Memories:', event.data['memoryIds'])
    elif event.event == 'error':
        raise RuntimeError(event.data['message'])

The [DONE] sentinel

OpenAI-style SSE streams end with a sentinel line:

data: [DONE]

The Recall SDK's parser recognises this and ends iteration cleanly. You do not need to special-case it; the for/async for loop simply exits. The synthetic done event is emitted just before the loop terminates so your code can run cleanup (close progress bars, flush metrics, etc.).

Multi-line data: joining

Per the SSE spec, multiple consecutive data: lines belong to the same event and are joined with '\n'. The parser handles this transparently:

event: stage
data: {"stage":"extract",
data:  "duration_ms":124}

This produces a single SseEvent with event.data == {'stage': 'extract', 'duration_ms': 124}. The newline is preserved between the data fragments during accumulation, then JSON-parsed once at the end.

CRLF normalization

httpx normalises both \r\n and \n line endings, but the parser also strips any stray \r for safety. You will not see CR characters in event.data regardless of how the upstream proxy buffers.

Comments and unknown fields

Per the SSE spec, lines starting with : are comments — the parser silently skips them. The id: and retry: fields are parsed but ignored; they have no semantic meaning in Recall's event model.

Mid-stream errors

A non-2xx response when opening the stream raises immediately — see the "Error handling" section in Pipelines. Errors after the connection is open arrive as event.event == 'error' events:

from recall.errors import RecallApiError

for event in stream:
    if event.event == 'error':
        # Server emitted an error mid-pipeline
        raise RecallApiError(
            event.data.get('message', 'pipeline error'),
            500,
            event.data.get('code', 'STREAM_ERROR'),
        )

The transport handles the upgrade-from-error-response case automatically: when an SSE call returns a non-2xx status, the transport reads the error body, infers the right error class (RecallAuthError/NotFoundError/ValidationError/RateLimitError/ ServerError), and raises before the first event is yielded.

Low-level parser

For testing or alternative transports, the parser is exposed:

from recall.stream import _parse_sse_lines, SseEvent

raw_lines = [
    'event: stage',
    'data: {"stage":"extract"}',
    '',
    'data: [DONE]',
    '',
]
events = list(_parse_sse_lines(iter(raw_lines)))
assert events[0].event == 'stage'
assert events[0].data == {'stage': 'extract'}
# [DONE] terminates the stream — events list ends here.

_parse_sse_lines accepts any iterable of strings and yields SseEvent instances. The leading underscore signals "internal" — it's stable in practice (used by both the sync and async transports) but reserves the right to change shape across SDK versions.

Auth and headers

Streaming requests carry the same headers as regular requests:

• Authorization: Bearer <api-key>
• Accept: text/event-stream
• Recall-Api-Version: 2026-04-30
• Recall-Sdk: py/0.3.0

The transport invokes the api_key resolver once per stream open, so a callable resolver gets a fresh token on every retry. There is no Idempotency-Key on streaming endpoints — re-opening a failed stream is a fresh logical operation.

Connection lifecycle

The sync write_stream opens the HTTP connection lazily — the first for iteration triggers the request. The connection is held open for the duration of the stream and closed when the loop exits (whether through done, error, or an unhandled exception in the body of the loop).

The async path uses async with self._client.stream(...) internally, which guarantees the connection is closed even if the consumer breaks out of the loop early.

# Both safe — connection cleaned up automatically
for event in stream:
    if event.event == 'result':
        break  # connection closes on next GC pass

async for event in stream:
    if event.event == 'result':
        break  # connection closes when async with exits

Backpressure

The SSE consumer drives the connection — if you stop iterating, the underlying socket fills its buffer and the server slows down. There is no explicit backpressure signal in SSE, but the TCP flow control gives you backpressure for free. In practice, write streams complete within ~1-3 seconds; if you're stalling for longer, you have a downstream bottleneck.