Source: https://docs.cumuluslabs.io/sdk-python
Content version: 0d306667

# Build with the Python SDK

Use `cumulus-sdk` to call the Cumulus HTTP API from your Python application. Its synchronous `Cumulus` client exposes domain methods such as `client.agents.list` and returns Pydantic response models. This guide uses version **0.2.11** and Python **3.10 or newer**.

## Install and connect {#install}

Obtain a read-only package username and token from Cumulus. Add them to `~/.netrc` and restrict its permissions with `chmod 600 ~/.netrc`:

```text
machine gitlab.cumuluslabs.io
  login <package-username>
  password <package-token>
```

Install the pinned package in a virtual environment:

```sh
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --index-url https://gitlab.cumuluslabs.io/api/v4/projects/97/packages/pypi/simple cumulus-sdk==0.2.11
```

Package credentials grant download access. Your workspace's `CUMULUS_API_KEY` authenticates API requests; keep it on your server.

Set `CUMULUS_API_URL` to your workspace's API origin and supply its API key through your environment. Save this as `list_agents.py`:

```python
import os
from cumulus import Cumulus

with Cumulus(
    base_url=os.environ["CUMULUS_API_URL"],
    api_key=os.environ["CUMULUS_API_KEY"],
    timeout=30.0,
) as client:
    page = client.agents.list(query={"limit": 5})
    for agent in page.agents:
        print(agent.ref.id, agent.name)
    print("Next page:", page.next_cursor)
```

Run `python list_agents.py`. This lists agents visible to your key. The context manager closes the HTTP client; a long-lived application can instead create one client and call `client.close()` during shutdown. Timeouts are in seconds.

The example passes the URL explicitly. If omitted, `base_url` reads `CUMULUS_BASE_URL`, then defaults to `https://api.cumuluslabs.io`. The Python constructor does not automatically read `CUMULUS_API_URL`. If `api_key` is omitted, it reads `CUMULUS_API_KEY`.

## Read an existing run

Set `RUN_ID` to a run your key can read. This script prints its transcript, tool effects and workflow result as JSON so you can inspect what was said, which actions occurred and their outcome. It does not start or change a run.

```python
import json
import os
from cumulus import Cumulus

with Cumulus(
    base_url=os.environ["CUMULUS_API_URL"],
    api_key=os.environ["CUMULUS_API_KEY"],
) as client:
    path = {"run_id": os.environ["RUN_ID"]}
    transcript = client.runs.transcript(path=path)
    effects = client.runs.effects(path=path)
    result = client.runs.result(path=path)
    print(json.dumps({
        "transcript": transcript.model_dump(mode="json", by_alias=True),
        "effects": effects.model_dump(mode="json", by_alias=True)["items"],
        "result": result.model_dump(mode="json", by_alias=True),
    }, indent=2))
```

`model_dump(mode="json", by_alias=True)` produces JSON-compatible values, including strings for UUIDs and enum values. Inspect each effect's state when deciding whether a business action succeeded. Transcript, recording and analysis have separate states; a run ending does not make every artifact ready.

## Pass operation arguments

Use the [customer HTTP API reference](/docs/api-reference) to choose an operation and its fields. The Python client uses these keyword arguments:

| Argument | Example and purpose |
| --- | --- |
| `path` | `{"run_id": run_id}` fills named URL parameters. |
| `query` | `{"limit": 20, "cursor": page.next_cursor}` supplies query parameters; `None` values are omitted. |
| `body` | A dictionary or a generated Pydantic request model. Model bodies use JSON aliases and omit unset fields. |
| `idempotency_key` | A saved string identifying one intended command. Operations requiring an idempotency header generate a UUID when you omit it. |

For example, `client.call("runs.transcript", path={"run_id": run_id})` calls the same operation as `client.runs.transcript(path={"run_id": run_id})`. Python keywords use a trailing underscore in domain methods, such as `client.agents.import_`.

Responses are validated against the installed package's generated models. Dictionary request bodies are sent as supplied; the client does not validate them against a request schema before sending. Construct a generated request model yourself when you want local Pydantic validation.

## Handle failures and retries

Catch `PlatformError` and inspect `code`, `status`, `retryable`, `outcome`, `details` and `idempotency_key`. This bounded retry is suitable for a read:

```python
from time import sleep
from cumulus import Cumulus, PlatformError

def list_agents_with_retry(client: Cumulus):
    for attempt in range(3):
        try:
            return client.agents.list(query={"limit": 5})
        except PlatformError as error:
            if not error.retryable or attempt == 2:
                raise
            sleep(0.25 * (2 ** attempt))
```

Commands need a different decision. If `error.outcome == "unknown"`, the operation may have happened. Preserve the original body and `error.idempotency_key`, inspect the affected resource, and use that same key if replaying the same intent. A fresh key can create another operation. The client does not automatically retry requests.

A connection that was never established reports `outcome="none"`. A lost command response or an incompatible successful command response reports an unknown outcome. An incompatible successful read response reports `owner_invalid_response` and is not retryable. Authentication and policy refusals should be handled through their structured error codes.

## Follow run events

`client.events(run_id, after_sequence=..., last_event_id=...)` reads server-sent events and yields validated Pydantic `SemanticEvent` models. For an existing run:

```python
import os
from cumulus import Cumulus

with Cumulus(
    base_url=os.environ["CUMULUS_API_URL"],
    api_key=os.environ["CUMULUS_API_KEY"],
) as client:
    for event in client.events(os.environ["RUN_ID"], after_sequence=0):
        data = event.model_dump(mode="json", by_alias=True)
        print(data["sequence"], data["type"])
        # In an application, durably handle and save this sequence before continuing.
        if data["type"] == "run.ended":
            break
```

Resume from your last durably handled sequence using `after_sequence`, and deduplicate event IDs. `last_event_id` sends the `Last-Event-ID` header when you have a saved cursor. This iterator has no HTTP timeout and does not reconnect automatically. Streaming transport, JSON or Pydantic validation errors can propagate directly; handle them separately from `PlatformError` when implementing reconnection.

## Troubleshoot version 0.2.11

Six methods have a known successful-response parsing defect: `numbers.release`, `secrets.delete`, `team.invite_revoke`, `team.remove`, `test_suites.delete` and `trunks.delete`. Their generated response target is the `Deleted` enum, but the client calls `model_validate` on it. A successful HTTP response can therefore raise `AttributeError: model_validate` instead of returning a model or `PlatformError`.

That exception does not prove the deletion failed. Inspect the resource's current state before deciding what happened. For these affected operations, use the documented HTTP endpoint directly with your HTTP client and inspect its response, or obtain a corrected SDK release. Changing to `client.call` does not bypass this defect because it uses the same response parser.

The Python package provides HTTP access; the TypeScript SDK's graph-authoring, customer-server adapters and runtime helpers are separate features. A Python server can receive tool calls, initial context and lifecycle events by implementing the [customer HTTP callback contracts](/docs/sdk-http-contracts) in its framework.