Pipelines resource

The non-streaming pipelines.write(...) is functionally identical to memories.write(...) — both POST to /v1/remember. The pipelines resource adds two streaming methods that return generators of SseEvent instances, exposing the seven stages of the write pipeline and the read pipeline as they run.

In multi-tenant servers, bind a per-user context with for_user() before calling any pipelines method:

# All pipeline methods are available on the bound view.
result = recall.for_user(user_id).pipelines.write(messages=[
    {'role': 'user', 'content': 'I prefer dark mode.'},
])

The bound view sets X-Recall-User-ID on every request and is otherwise identical to the root client. It does not construct a new HTTP connection.

write()

def write(
    self,
    messages: list[dict[str, str]],
    *,
    idempotency_key: Optional[str] = None,
) -> Any: ...

Non-streaming write — alias for memories.write(). Returns the final result dict in one shot.

result = recall.pipelines.write(messages=[
    {'role': 'user', 'content': 'I work at Arc Labs.'},
])

result = await recall.pipelines.write(messages=[
    {'role': 'user', 'content': 'I work at Arc Labs.'},
])

write_stream()

def write_stream(
    self,
    messages: list[dict[str, str]],
) -> Generator[SseEvent, None, None]: ...

async def write_stream(
    self,
    messages: list[dict[str, str]],
) -> AsyncGenerator[SseEvent, None]: ...

Stream the write pipeline. Yields SseEvent instances for each stage (pipeline_init, stage_started, stage, llm_call, result, error, done). The connection closes when the server emits the OpenAI-style [DONE] sentinel.

Examples

for event in recall.pipelines.write_stream(messages=[
    {'role': 'user', 'content': 'I work at Arc Labs.'},
    {'role': 'assistant', 'content': 'Got it.'},
]):
    if event.event == 'stage':
        print('Stage:', event.data.get('stage'))
    elif event.event == 'result':
        print('Memories:', event.data['memoryIds'])
    elif event.event == 'error':
        print('Error:', event.data)

async for event in recall.pipelines.write_stream(messages=[
    {'role': 'user', 'content': 'I work at Arc Labs.'},
    {'role': 'assistant', 'content': 'Got it.'},
]):
    if event.event == 'stage':
        print('Stage:', event.data.get('stage'))
    elif event.event == 'result':
        print('Memories:', event.data['memoryIds'])

When to use

• Surface pipeline progress in a UI ("extracting…", "deduping…", "indexing…").
• Build interactive debugging tools that show what the LLM extracted before the dedupe stage filtered candidates.
• Monitor latency stage-by-stage in production.

For straight-line "give me the IDs of the memories you wrote", use pipelines.write() or memories.write() — they're cheaper because they don't allocate the SSE buffer.

search_stream()

def search_stream(
    self,
    query: str,
    limit: Optional[int] = None,
) -> Generator[SseEvent, None, None]: ...

# AsyncRecall:
async def search_stream(
    self,
    query: str,
    limit: Optional[int] = None,
) -> AsyncGenerator[SseEvent, None]: ...

Stream the read pipeline. Yields events for each retriever as it completes, then the fused result, then done.

Examples

for event in recall.pipelines.search_stream(
    query='UI preferences',
    limit=5,
):
    if event.event == 'stage':
        print('Stage:', event.data.get('stage'))
    elif event.event == 'result':
        for hit in event.data['results']:
            print(hit['content'])

async for event in recall.pipelines.search_stream(
    query='UI preferences',
    limit=5,
):
    if event.event == 'result':
        for hit in event.data['results']:
            print(hit['content'])

When to use

• Render a "Recall is thinking…" indicator with stage detail in a chat UI.
• Debug retrieval — see which retriever fired first and what scores it produced before fusion.
• Build a streaming dashboard that shows search latency broken down by retriever.

SSE event types

event.event is one of:

See Streaming for the full SSE parser internals.

Iteration patterns

The sync stream is a Generator — iterate with for:

for event in recall.pipelines.write_stream(messages=msgs):
    ...

The async stream is an AsyncGenerator — iterate with async for:

async for event in recall.pipelines.write_stream(messages=msgs):
    ...

Note that async streams must be iterated inside an async function. You cannot list(...) an AsyncGenerator — use a comprehension:

events = [event async for event in recall.pipelines.write_stream(messages=msgs)]

Error handling

A non-2xx response when opening the stream raises a typed error immediately — RecallAuthError on 401/403, RecallNotFoundError on 404, RecallValidationError on 422, RecallRateLimitError on 429, RecallServerError on 5xx. The transport reads the error envelope from the streaming response body and constructs the right error class.

from recall.errors import RecallAuthError

try:
    for event in recall.pipelines.write_stream(messages=msgs):
        handle(event)
except RecallAuthError as exc:
    print('auth failed:', exc.code, exc.request_id)

Errors emitted mid-stream (after the connection is open) arrive as event.event == 'error' events. Inspect event.data and break out of the loop:

for event in stream:
    if event.event == 'error':
        raise RuntimeError(f"pipeline failed: {event.data}")

Idempotency

pipelines.write() carries a Recall-Idempotency-Key header (UUID v4 auto-generated unless you override). The streaming variants do not — SSE streams are not retriable in the same way. If a streaming write fails mid-stream, you must retry the whole call.