swarm_analytics

A typed, auto-generated Python client for the OpenSwarm product-analytics ingest API. It is the single network egress the desktop FastAPI backend imports to send analytics — identity comes from a bearer token, payloads are validated against the exact pydantic models the server enforces, and delivery is fire-and-forget with background retry.

Why it's hard to call wrong

• Identity is impossible to pass. No method takes install_id/user_id; the server resolves them from the token.
• Per-request meta is auto-filled. ts and submission_id never appear in a signature — the transport stamps them (and reuses submission_id on every retry for idempotency).
• Enums stay enums. status, action, role, etc. are Literals. A bad value raises pydantic.ValidationError synchronously, in your stack, before any network I/O.
• Models are vendored verbatim from the service, so the client validates with the same schema the server uses.

Install

pip install ./sdk          # from the repo root

Usage

from swarm_analytics import AnalyticsClient, AgentMessage

# One-time bootstrap on first launch (unauthenticated, blocking)
token = AnalyticsClient.register(base_url="https://analytics.openswarm.ai", install_id=install_id)
# persist `token` in settings; reuse forever

client = AnalyticsClient(base_url="https://analytics.openswarm.ai", token=token)

client.events.app_lifecycle.opened(os="darwin", os_version="25.3.0", app_version="1.2.0")
client.events.agent.create(id="sess_123", name="Refactor auth", dashboard_id="dash_1")
client.events.agent.message(agent_id="sess_123", seq=0,
                            message=AgentMessage(id="m1", role="user", content="hello"))
client.events.onboarding.step(step_id="connect_provider", status="completed")
client.events.dashboard.event(dashboard_id="dash_1", action="create")
client.logs.write(tag="agent", subtag="tool", data={"name": "shell"})
client.identify.link_email(email="user@example.com")

# On shutdown
client.events.app_lifecycle.closed()
client.flush(timeout=2.0)
client.close()

Durability (optional)

By default events live in an in-memory queue and are lost if the process dies with deliveries pending. Pass a spool for crash/offline durability:

from swarm_analytics import SqliteSpool
client = AnalyticsClient(base_url=..., token=..., spool=SqliteSpool("service_spool.db"))

Opt-out

client = AnalyticsClient(base_url=..., token=..., mode="minimal")  # mutes product events; diagnostics still flow

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

The models, route table, and namespaces under src/swarm_analytics/_generated/ are produced from the live service. Regenerate whenever the backend's ingest models or routes change:

PYTHONPATH=<repo_root> python sdk/generate.py

ROUTE_SPECS in generate.py (nice method name + category per endpoint) is the only human input; it is cross-checked against the live app, so a new or removed endpoint fails generation rather than drifting silently.

Drift check (CI)

PYTHONPATH=<repo_root> python sdk/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 && PYTHONPATH=<repo_root> python -m pytest tests -q

Covers synchronous validation, meta auto-fill, identity-from-token, idempotent submission_id reuse across retries, opt-out gating, the drift check, and an end-to-end pass through the real FastAPI app.