Python SDK

The Python SDK is a typed client for the Ellipsis API. Every /v1 operation is generated from the same OpenAPI spec the API reference is built from, so the two cannot disagree. On top of the generated methods it adds a session handle, live streaming, transparent pagination, and typed errors.

Install and authenticate

The package is ellipsis-dev on PyPI; the import name is ellipsis. Python 3.10 or newer. Live streaming needs the stream extra.

pip install 'ellipsis-dev[stream]'

Create an API key on the API keys page of the dashboard and keep it in an environment variable. api_key is the client's only required argument; a CLI user token from agent login works in the same argument. base_url defaults to https://api.ellipsis.dev, timeout to 60 seconds per request, and max_retries to 2.

import os

from ellipsis import Ellipsis

client = Ellipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"])

Ellipsis is synchronous and works as a context manager. AsyncEllipsis has the identical method surface with await, works as async with, and is the only client that can stream.

Start a session and wait

sessions.run() takes exactly the keyword arguments of sessions.start(): prompt, environment (a saved environment's name or env_ id, or an inline object), claude or codex, permissions, skills, output, budget in dollars, interactive, force_rebuild, and metadata. Anything omitted falls back to your organization settings, so prompt alone is a complete request. It returns a SessionHandle immediately; the agent works in a cloud sandbox while your code decides what to do.

handle = client.sessions.run(
    prompt="Fix the flaky test in ci/",
    environment="cloud_agent_environment",
    budget=5.00,
)
handle.send("Also add a regression test", idempotency_key="regression-ask")
session = handle.wait(timeout=900)
print(session.status, session.exit_status)

wait(timeout=None, poll_interval=3.0) polls until the session settles, and raises TimeoutError after timeout seconds, leaving the session running. send(message, idempotency_key=None) posts into the session's inbox: delivered at the next turn boundary, it wakes a parked session, and the same key delivers once. stop() stops the agent now; history stays readable. records(cursor=None, limit=None) returns the session's record log as a page, and refresh() re-fetches handle.session, the newest snapshot. sessions.handle(session_id) builds a handle over a session another process started.

Settled means status is completed, error, cancelled, or stopped, or a durable conversation parked with session_state of idle or closed. Parking matters: a durable conversation ends each turn with a terminal status while the conversation stays alive, so waiting on status alone would return after the first turn.

Invoke an automation

An automation is a saved agent definition. automations.run() runs it exactly as defined and returns a SessionResponse; the body carries only the typed input its schema demands, an optional budget that may only lower the automation's own, and metadata. Wrap the result in a handle to wait on it. automations.list, get, create, update, delete, link, and unlink manage the definitions; templates.list and templates.get return the built-in starters.

response = client.automations.run(
    "linear-issue-implementer", input={"issue": "WEB-142"}
)
session = client.sessions.handle(response.session.id).wait()

Stream a session live

Async only. handle.stream(on_frame) opens the session's WebSocket, calls on_frame (sync or async) for every frame, and returns a StreamOutcome when the session finishes: type is done (with the final status and exit_status), error (with a message), or aborted if you cancelled the task. stream_session(session_id=..., api_key=..., on_frame=...) does the same for a session you did not start here. Inside an async function:

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(lambda frame: print(frame.type))
    print(outcome.type, outcome.status)

Seven frame types arrive, each a model in ellipsis.frames. SnapshotFrame opens with the session and the earliest available record sequence. RecordsAppendFrame carries new records in feed_seq order, the log of what the agent did. SessionFrame resends the whole session whenever it changes. DeltaFrame is partial text as the model produces it. HeartbeatFrame is liveness, roughly every 20 seconds. DoneFrame ends the conversation. ErrorFrame reports a server-side failure with a message. Unknown frame types are dropped before they reach you, so new ones are not a break.

A dropped socket reconnects with capped backoff, up to max_reconnects consecutive failures (5 by default; any delivered frame resets the count). Resume is cursored on records_append alone: the client asks for everything after the highest feed_seq it saw, so no record is lost or repeated, and snapshot and session frames are resent whole. StreamUnavailableError means streaming cannot be used here (extra missing, protocol unsupported, reconnects exhausted): poll with handle.wait() instead. StreamAuthError means the credential was rejected; polling would fail the same way.

List and paginate

A list method returns a page. Iterating it walks every page, one request per page, holding only the current page in memory: for session in client.sessions.list(days=7): reads a week of history. sessions.list, sessions.records, reviews.list, alerts.list, and files.list paginate; each accepts cursor and limit, and limit bounds the page, not the total. For one page at a time read page.items, page.has_more, and page.next_cursor (pass it back as cursor); page.response is the raw envelope. On the async client, await the method for the first page and async for over it for the rest.

Errors

Every failure raises an EllipsisError. TransportError means the request never got a response (DNS, TLS, timeout). APIError is any non-2xx answer, carrying status, code, message, request_id, and body; its subclasses are chosen by status: AuthenticationError 401, ForbiddenError 403, NotFoundError 404, ConflictError 409, UnprocessableError 422, RateLimitError 429, ServerError 5xx. Switch on code, never on message; codes are an open vocabulary, so treat an unrecognized one by its status.

from ellipsis import APIError

try:
    handle.send("keep going")
except APIError as error:
    if error.code != "session_finished":
        raise

Transport failures and 429, 502, 503, and 504 are retried max_retries times with exponential backoff and jitter (a 429 honors Retry-After) before anything reaches you. A RateLimitError in your code means the retries are already spent. No other status is retried.

Types

Requests and responses are Pydantic v2 models in ellipsis.models. Responses are envelopes: sessions.get() returns a SessionResponse whose session is the Session, and the handle unwraps it for you. A field the API may omit is Union[T, None] and defaults to None. Every model allows extra fields, so an additive server change parses cleanly; open vocabularies such as a record's source or an error code are typed str. Timestamps are datetime. Money is a float in dollars: budget=5.00 is five dollars.

Example: run in CI and fail the job on failure

import os
import sys

from ellipsis import Ellipsis

client = Ellipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"])
owner, name = os.environ["GITHUB_REPOSITORY"].split("/", 1)
handle = client.sessions.run(
    prompt="Fix the flaky test in ci/",
    environment={"repositories": [{"owner": owner, "name": name}]},
    budget=5.00,
)
session = handle.wait(timeout=1800)
print(f"{handle.id} {session.status} ({session.exit_status})")
if session.status != "completed":
    sys.exit(1)

wait() raises TimeoutError after 30 minutes, which fails the job and leaves the session running. Any status other than completed exits non-zero.

Example: fan out across repositories concurrently

import asyncio
import os

from ellipsis import AsyncEllipsis


async def run_one(client: AsyncEllipsis, repository: str) -> None:
    owner, name = repository.split("/", 1)
    handle = await client.sessions.run(
        prompt="Fix the flaky test in ci/",
        environment={"repositories": [{"owner": owner, "name": name}]},
    )
    session = await handle.wait()
    print(f"{repository:20} {session.status}")


async def main() -> None:
    api_key = os.environ["ELLIPSIS_API_TOKEN"]
    async with AsyncEllipsis(api_key=api_key) as client:
        await asyncio.gather(
            run_one(client, "your-org/web-repo"),
            run_one(client, "your-org/api-repo"),
        )


asyncio.run(main())