Python SDK
Start sessions, stream records, and invoke automations from Python.
Install
Python 3.10 or newer. The stream extra enables live session streaming.
pip install 'ellipsis-dev[stream]'Start a session
import os
from ellipsis import Ellipsis
from ellipsis.models import ClaudeConfig
client = Ellipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"])
handle = client.sessions.run(
harness=ClaudeConfig(type="claude_code"),
environment="api-environment",
prompt="Run the tests and report failures.",
interactive=False,
budget=3,
)
session = handle.wait(timeout=900)
print(session.status, session.exit_status)run returns a handle while work continues. wait polls until the session settles. A timeout stops waiting and leaves the session running.
Use sessions.handle(session_id) to attach to an existing session.
Continue a conversation
Leave interactive enabled to accept messages:
handle = client.sessions.run(
harness=ClaudeConfig(type="claude_code"),
environment="api-environment",
prompt="Investigate the failing validation test.",
budget=5,
)
handle.wait(timeout=900)
handle.send("Add a regression test.", idempotency_key="add-test")
handle.wait(timeout=900)
handle.stop()The same message key is accepted once per session. A message during an active turn waits for the next turn.
Invoke an automation
response = client.automations.run(
"test-repair",
prompt="Investigate the request validation tests.",
)
session = client.sessions.handle(response.session.id).wait()
print(session.exit_status)For an automation with an input schema, pass input instead:
response = client.automations.run(
"classify-change",
input={"description": "Reject expired reset tokens"},
)Stream a session
See Session events for every modeled event and complete JSON examples.
import asyncio
import os
from ellipsis import AsyncEllipsis
from ellipsis.models import ClaudeConfig
async def main():
async with AsyncEllipsis(
api_key=os.environ["ELLIPSIS_API_TOKEN"]
) as client:
handle = await client.sessions.run(
harness=ClaudeConfig(type="claude_code"),
prompt="Explain how to test a request validator.",
interactive=False,
)
outcome = await handle.stream(lambda frame: print(frame.type))
print(outcome.type)
asyncio.run(main())Use AsyncEllipsis for asynchronous requests and streaming. Complete records persist; partial text deltas are live-only.
Read results
for execution in client.sessions.executions(handle.id).executions:
if execution.result is not None:
print(execution.result.text)
for session in client.sessions.list(days=7):
print(session.id, session.exit_status)SDK page iterators fetch every page. For manual pagination, use items, has_more, and next_cursor.
Errors
from ellipsis import APIError
try:
client.sessions.get("session_missing")
except APIError as error:
print(error.status, error.code)The client defaults to a 60-second request timeout and two retries for transport errors, 429, 502, 503, and 504. Configure timeout and max_retries when creating it.