The Compass SDK
The async Python client for Compass — configure it, call resources, and run your work.
The Compass SDK is an async Python client for the platform. It's the same wheel as the CLI, imported as a library, and it exposes every resource you work with in the app.
Install
Download the wheel from Settings → Developers → SDK / CLI and add it to your project (Python 3.13+):
pip install ./compass_cli.whl
# or: uv add ./compass_cli.whlfrom compass_core import CompassCreate a client
The client is async and works as an async context manager. Give it an API key and your Compass URL:
import asyncio
from compass_core import Compass
async def main():
async with Compass(
token="<your-api-key>",
base_url="https://your-compass-url",
) as client:
...
asyncio.run(main())The key is already scoped to a project, so you don't pass a project id. (There's an
optional project_id argument for user-token use; with an API key, leave it unset.)
The client sends your key as a bearer token and targets the v1 API.
Resources
The client groups calls by resource:
| Resource | Use it for |
|---|---|
client.runs | Start runs and read their results and traces. |
client.workflows | List, get, create, update, validate, and deploy workflows. |
client.conversations | Group runs into conversations. |
client.connections | Manage integration connections. |
client.providers | Manage provider (LLM) connections. |
client.gateways | Manage gateway connections. |
client.streams | Subscribe to a run's live events. |
(Agents are workflows of type agent, so they're managed through client.workflows
and run through client.runs.)
Run something
Start a run with either a deployment_id (the live version) or a workflow_id (a
specific workflow or agent), passing its input:
run = await client.runs.create(
deployment_id="<deployment-id>",
input={"topic": "Q3 report"},
)
print(run.id, run.status)Read the result
Fetch the run for its status and output, or pull its full trace:
run = await client.runs.get(run.id)
tree = await client.runs.tree(run.id) # hierarchical trace
steps = await client.runs.executions(run.id) # per-node executionsStream it live
To watch a run as it happens, iterate its event stream:
async for event in client.streams.events(run.id):
print(event)Handling errors
The SDK raises typed exceptions you can catch:
| Exception | When |
|---|---|
AuthError | The key is missing, invalid, or lacks access. |
NotFoundError | The resource doesn't exist in your project. |
ValidationError | The request was rejected as invalid. |
RateLimitError | You've been rate limited (the client retries with backoff). |
ServerError | Something failed server-side. |
All of them derive from a common CompassClientError, so you can catch broadly or
narrowly.