ergon-platform-sdk 0.1.0

Official Python client SDK for the Ergon Platform public API

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

Mode	Constructor args	Notes
API key (M2M)	client_id, client_secret	Primary. Token exchanged + refreshed automatically. Tenant is implicit.
Bearer token	token=	Pre-obtained JWT (e.g. a user access token). Pass company_id= if the token isn't pinned to a company.

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