swarm_analytics_private

A typed, auto-generated Python client for the OpenSwarm product-analytics private API — the internal read/ops surface under /private (the admin "Tables" explorer and the classify jobs/runs + manual triggers).

This is the sibling of swarm_analytics (the public ingest SDK), but a different kind of client: where the public SDK is fire-and-forget ingest, this one is blocking request/response — every method issues one HTTP call and returns a validated pydantic model.

Internal tooling only. Authentication is a static admin key sent in the X-Admin-Key header, not an install bearer token. Don't distribute this to end users alongside the public SDK.

Why it's hard to call wrong

• One credential, sent everywhere. The client only ever sends the admin key; generation fails unless every /private route is gated by require_admin.
• Responses are vendored verbatim from the service, so the client validates with the exact pydantic schema the server returns. status/trigger stay Literals.
• Params are typed. Path and query params become typed keyword arguments; None query params are dropped so an omitted optional means "server default".

Install

pip install ./sdk-private    # from the repo root

Usage

from swarm_analytics_private import AdminClient

with AdminClient(base_url="https://analytics.openswarm.ai", admin_key=ADMIN_KEY) as client:
    # Reader — generic DB introspection
    tables = client.reader.tables()                       # -> TablesResponse
    rows = client.reader.table(name="agent", limit=50,    # -> TableRowsResponse
                               sort="created_at", dir="desc",
                               filters={"dashboard_id": "dash_1"})
    # Keyset streaming of a large table (constant per-page cost, complete):
    page = client.reader.table(name="agent_message", limit=200,
                               after_rowid=0, columns="agent_id,seq")
    # page.rows each carry "_rowid"; page.next_cursor feeds the next after_rowid;
    # page.total is None on keyset pages.
    agent_ids = client.reader.distinct_values(name="agent_message",  # -> DistinctValuesResponse
                                              column="agent_id")
    shape = client.reader.json_shape(name="session_message")  # -> JsonShapeResponse

    # Classify — jobs/runs + manual triggers
    jobs = client.classify.list_jobs()                    # -> JobsResponse
    runs = client.classify.list_runs(job_id="classify_agents", limit=20)  # -> RunsResponse
    result = client.classify.run_clio()                   # -> ClassifyAgentsRecord (POST)
    cluster = client.classify.run_cluster()               # -> ClassificationClusterRecord (POST)

The filters keyword on reader.table is the typed surface over the endpoint's arbitrary equality-filter feature (any column name → column = value).

Regenerating (auto-generated — do not hand-edit _generated/)

The models and namespaces under src/swarm_analytics_private/_generated/ are produced from the live service. Regenerate whenever the backend's private routes or response models change:

PYTHONPATH=<repo_root> python sdk-private/generate.py

ROUTE_SPECS in generate.py (namespace + method name per endpoint) is the only human input; it is cross-checked against the live app, and every route must be admin-gated and declare a pydantic response_model, so a new/removed endpoint — or a route that returns an untyped dict — fails generation rather than drifting silently.

Drift check (CI)

PYTHONPATH=<repo_root> python sdk-private/generate.py --check

Exits non-zero if the committed _generated/ output is stale. The same guard runs as tests/test_drift.py.

Tests

cd sdk-private && PYTHONPATH=<repo_root> python -m pytest tests -q

Covers the admin-key header, None-param dropping, path interpolation, response validation, and the drift check.