Examples
Complete scripts for the common jobs: run an agent in CI, fan out across repositories, watch a session live, and audit last week's work.
Each script below runs as written, given ELLIPSIS_API_TOKEN in the environment.
Run an agent in CI and fail the job if it fails
import os
import sys
from ellipsis import Ellipsis
client = Ellipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"])
handle = client.sessions.run(
prompt="Update the OpenAPI docs to match the routes on this branch",
repository=os.environ["GITHUB_REPOSITORY"],
)
print(f"session {handle.id}", flush=True)
session = handle.wait(timeout=1800)
print(f"finished {session.status}")
if session.status != "completed":
sys.exit(1)wait() raises TimeoutError after 30 minutes here, which leaves the session running. Exit non-zero on any status other than completed so a failed agent fails the build.
Fan out across repositories, concurrently
import asyncio
import os
from ellipsis import AsyncEllipsis
REPOSITORIES = ["acme/api", "acme/web", "acme/jobs"]
PROMPT = "Upgrade the pinned Python version to 3.13 and fix what breaks"
async def run_one(client: AsyncEllipsis, repository: str) -> tuple[str, str]:
handle = await client.sessions.run(prompt=PROMPT, repository=repository)
session = await handle.wait()
return repository, session.status
async def main() -> None:
async with AsyncEllipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"]) as client:
results = await asyncio.gather(
*(run_one(client, repository) for repository in REPOSITORIES)
)
for repository, status in results:
print(f"{repository:12} {status}")
asyncio.run(main())Three sessions run in parallel in three sandboxes. Each has its own budget.
Watch a session live
import asyncio
import os
from ellipsis import AsyncEllipsis, StreamUnavailableError, frames
async def main() -> None:
async with AsyncEllipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"]) as client:
handle = await client.sessions.run(prompt="Fix the flaky test in ci/")
def on_frame(frame: frames.StreamFrame) -> None:
if isinstance(frame, frames.DeltaFrame):
print(frame.text or "", end="", flush=True)
elif isinstance(frame, frames.SessionFrame):
print(f"\n[{frame.session.status}]")
try:
outcome = await handle.stream(on_frame=on_frame)
print(f"\n{outcome.type}")
except StreamUnavailableError:
session = await handle.wait()
print(f"\npolled to {session.status}")
asyncio.run(main())Needs the stream extra. Deltas print as the model produces them; the poll fallback covers an environment where the WebSocket cannot be opened.
Audit last week's sessions
import os
from collections import Counter
from ellipsis import Ellipsis
client = Ellipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"])
by_status: Counter[str] = Counter()
for session in client.sessions.list():
by_status[session.status] += 1
for status, count in by_status.most_common():
print(f"{status:12} {count}")
print(f"{sum(by_status.values())} sessions")Iterating walks every page, so this reads the full history rather than the first page. Add limit to bound the page size, not the total.
Send a follow-up to a session already running
import os
from ellipsis import APIError, Ellipsis
client = Ellipsis(api_key=os.environ["ELLIPSIS_API_TOKEN"])
handle = client.sessions.handle(os.environ["SESSION_ID"])
try:
handle.send("Also add a regression test", idempotency_key="regression-ask")
except APIError as error:
if error.code == "session_finished":
print("session already done")
else:
raiseThe idempotency key makes a retry of this script safe: the same key delivers the message once.