Python SDK
Start sessions, stream records, and invoke agents from Python.
The Python SDK is a client for the /v1 API with session handles, streaming, pagination, and typed errors.
Python 3.10 or newer. The stream extra enables live session streaming.
1
pip install 'ellipsis-dev[stream]'
1
import os
2
3
from ellipsis import Ellipsis
4
5
client = Ellipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"])
6
handle = client.sessions.run(
7
claude_code={
8
"model": "claude-opus-5-5",
9
"prompt": "Run the tests and report failures.",
10
},
11
environment="api-environment",
12
conversation={"interactive": False},
13
budget={"session": 3},
14
)
15
session = handle.wait(timeout=900)
16
print(session.turn.status, session.turn.reason, session.turn.detail)
run returns a handle while work continues. wait polls until the turn that answers the prompt reaches a final status (completed, failed, stopped, or cancelled) and returns the session; session.turn is that turn. A timeout stops waiting and leaves the session running.
Use sessions.handle(session_id) to attach to an existing session.
Leave conversation.interactive enabled to accept messages:
1
handle = client.sessions.run(
2
claude_code={
3
"model": "claude-opus-5-5",
4
"prompt": "Investigate the failing validation test.",
5
},
6
environment="api-environment",
7
budget={"session": 5},
8
)
9
handle.wait(timeout=900)
10
message = handle.send("Add a regression test.", idempotency_key="add-test")
11
session = handle.wait(timeout=900)
12
print(message.turn_id, session.turn.status)
send returns the message with the turn_id of the turn that answers it, and wait then waits on that turn. The same message key is accepted once per session. A message sent while a turn is running is answered after it. handle.stop() ends the running turn with stopped; it fails with 409 when no turn is pending or running.
Start an agent session as written:
1
response = client.agents.start("test-repair")
2
session = client.sessions.handle(response.session.id).wait()
3
print(session.turn.status, session.turn.detail)
Send any field of a session start to replace the agent's value for that session:
1
response = client.agents.start("test-repair", budget={"session": 5})
For an agent with an input schema, pass input:
1
response = client.agents.start(
2
"classify-change",
3
input={"description": "Reject expired reset tokens"},
4
)
See Session events for every modeled event and complete JSON examples.
1
import asyncio
2
import os
3
4
from ellipsis import AsyncEllipsis
5
6
7
async def main():
8
async with AsyncEllipsis(
9
api_key=os.environ["ELLIPSIS_API_TOKEN"]
10
) as client:
11
handle = await client.sessions.run(
12
claude_code={
13
"model": "claude-opus-5-5",
14
"prompt": "Explain how to test a request validator.",
15
},
16
conversation={"interactive": False},
17
)
18
outcome = await handle.stream(lambda frame: print(frame.type))
19
print(outcome.type)
20
21
22
asyncio.run(main())
Use AsyncEllipsis for asynchronous requests and streaming. Complete records persist; partial text deltas are live-only.
1
turns = client.sessions.turns.list(handle.id)
2
for turn in turns.turns:
3
print(turn.index, turn.status, turn.reason, turn.cost.total)
4
for message in turns.messages:
5
print(message.turn_id, message.status, message.body)
6
7
for session in client.sessions.list(days=7):
8
print(session.id, session.conversation.state)
turns.list returns every turn and message of a session, oldest first; turns.get(session_id, turn_id) returns one turn. SDK page iterators fetch every page. For manual pagination, use items, has_more, and next_cursor.
1
from ellipsis import APIError
2
3
try:
4
client.sessions.get("session_missing")
5
except APIError as error:
6
print(error.status, error.code)
The client defaults to a 60-second request timeout and 2 retries for transport errors, 429, 502, 503, and 504. Configure timeout and max_retries when creating it.