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

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

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

Install the pinned package in a virtual environment:

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:

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.

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 to choose an operation and its fields. The Python client uses these keyword arguments:

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:

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:

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 in its framework.