Ergon Platform SDK (Python)

Typed Python client for the Ergon Platform public HTTP API. Async and sync, with automatic API-key authentication, retries, pagination, and typed errors.

• Distribution: ergon-platform-sdk
• Import root: ergon_platform

This is the client SDK (for consuming the platform API). It is unrelated to ergon-framework-python (import root ergon), which is for building services.

Install

pip install ergon-platform-sdk
# or, in this workspace:
uv sync

Quick start

Synchronous

from ergon_platform import ErgonClient

# Machine-to-machine: the API key's company is baked into the issued token,
# so you never pass company_id.
client = ErgonClient(
    client_id="ek_...",
    client_secret="eks_...",
    base_url="http://localhost",  # gateway origin
)

# IAM — `companies` is a uniform collection
company = client.iam.companies.create(name="Acme")
for key in client.iam.api_keys.iter(company_id=company.id):
    print(key.name)

# Workflows — the service node *is* the workflow resource (list/create are
# company-scoped); drill into one workflow's subtree via the scope accessor.
page = client.workflows.list(limit=50)  # company_id is inferred from the JWT
for wf in page.items:
    print(wf.id, wf.name)

scope = client.workflows.workflow(page.items[0].id)
scope.create_item(title="New ticket")          # POST workflows/{id}/items
client.workflows.items.claim(item_id="...")     # id-addressed action

Asynchronous

import asyncio
from ergon_platform import AsyncErgonClient


async def main() -> None:
    async with AsyncErgonClient(client_id="ek_...", client_secret="eks_...") as client:
        # `iter` transparently follows pagination across pages
        async for bucket in client.buckets.buckets.iter():
            print(bucket.id, bucket.name)


asyncio.run(main())

Errors

from ergon_platform import NotFoundError, PermissionDeniedError

try:
    client.iam.companies.get("does-not-exist")
except NotFoundError as exc:
    print(exc.status_code, exc.detail, exc.request_id)

Authentication

Services

Ten services hang off the client:

client.iam, client.workflows, client.channels, client.agents, client.worksheets, client.buckets, client.event_streams, client.conversations, client.automations, client.apps.

Conventions

The facade follows the same shape across every service, so once you learn one you know them all:

• Collections — a uniform REST resource exposes list / iter / get / create / update / delete. For example client.buckets.buckets or client.iam.companies.
• Pages — list(...) returns a Page with .items and .total; iter(...) is a generator (or async generator) that transparently walks every page for you.
• Scopes — resources nested under a parent id are reached through an accessor that returns a sub-client, e.g. client.workflows.workflow(workflow_id) exposes workflows/{id}/*, and client.buckets.buckets.access(bucket_id).grants exposes the per-bucket grant subtree.
• Actions — non-CRUD endpoints are plain methods named after the verb, e.g. client.workflows.items.claim(item_id=...) or .route(item_id=..., target_phase_id=...).
• Typing — responses are parsed into generated Pydantic models when the schema is unambiguous; otherwise a permissive model is used that preserves every field the server returns (so new fields never break you).
• Escape hatch — the full, fully-typed generated client and every schema for every endpoint live under ergon_platform._generated.<service> if you need an operation the facade doesn't surface ergonomically.

Both ErgonClient and AsyncErgonClient expose the identical surface; the async variant returns awaitables / async iterators.

Coverage

The facade wires every public operation of all ten services (see the generated matrix at docs/coverage.md). Only each service's /health liveness probe is intentionally left out. A CI guard (scripts/coverage.py --check) fails the build if the platform grows a new endpoint that isn't wired, so the SDK can't silently drift behind the API.

Development

uv run python scripts/export_openapi.py   # refresh specs from a running platform
uv run python scripts/generate.py          # regenerate the _generated client layer
uv run python scripts/coverage.py          # regenerate docs/coverage.md
uv run python scripts/coverage.py --check  # CI guard: fail on unwired endpoints
uv run --package ergon-platform-sdk pytest # unit tests
ERGON_INTEGRATION=1 uv run --package ergon-platform-sdk pytest -m integration