Streaming

Streaming delivers a session's activity as it happens, instead of polling for a status. It is async only, because live frames arrive over a WebSocket.

Install the extra

pip install 'ellipsis-dev[stream]'

Without it, streaming raises StreamUnavailableError telling you to install websockets.

Stream a session

import os

from ellipsis import AsyncEllipsis

async with AsyncEllipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"]) as client:
    handle = await client.sessions.run(prompt="Fix the flaky test in ci/")
    outcome = await handle.stream(on_frame=print)
    print(outcome.type, outcome.status)

on_frame is called for every frame and may be sync or async. stream() returns when the session finishes, and the StreamOutcome says how: done for a finished conversation, error for a server-side failure with a message, aborted if you cancelled the task.

To stream a session you did not start in this process, use stream_session() directly:

import os

from ellipsis import stream_session

outcome = await stream_session(
    session_id="session_7Hq2mX4p",
    api_key=os.environ["ELLIPSIS_API_TOKEN"],
    on_frame=print,
)

Frame types

Seven frame types arrive, each a typed model in ellipsis.frames:

Ignore frame types you do not recognize. New frame types and new values inside them are additive, not a protocol break, so a client that rejects the unfamiliar breaks on a future release.

Reconnect and resume

A dropped socket reconnects with capped backoff, up to max_reconnects consecutive failures (5 by default). Resume is cursored on records_append only: the client tracks the highest feed_seq it has seen and asks for everything after it, so no record is delivered twice or lost. snapshot and session frames are whole-state and are resent in full, which is why they never advance the cursor.

Any frame arriving resets the failure count, so a long session that drops occasionally keeps a full reconnect budget each time.

When streaming is not available

Two errors, with different responses:

• StreamUnavailableError means streaming cannot be used here: the extra is missing, the server does not support the protocol version, or reconnects are exhausted. Poll with handle.wait() instead.
• StreamAuthError means the credential was rejected for this session. Polling would fail the same way, so this is not a fallback case; fix the credential.

from ellipsis import StreamAuthError, StreamUnavailableError

try:
    await handle.stream(on_frame=print)
except StreamUnavailableError:
    session = await handle.wait()
except StreamAuthError:
    raise

The stream client speaks protocol version 3 and sends it on every handshake.