campai-client

A Python SDK for campai.com, generated from campai's OpenAPI spec, with pydantic v2 models, sync + async clients, and a durable patch layer for spec-vs-reality fixes.

Status: early / alpha. See docs/DESIGN.md for the full architecture and docs/adr/ for key decisions.

Install

pip install campai-client

Requires Python ≥ 3.13. Runtime deps: httpx, pydantic>=2, pydantic-settings.

Quickstart

from campai_client import CampaiClient
from campai_client import models

client = CampaiClient(
    api_key="...",  # or env CAMPAI_API_KEY
    organization_id="...",  # or env CAMPAI_ORGANIZATION_ID
    mandate_id="...",  # optional default; overridable per call
    # base_url defaults to https://cloud.campai.com/api
)

# Ergonomic namespaces mirror campai's tag hierarchy. List bodies (pagination +
# filters) are typed via the generated request model:
page = client.crm.applications.forms.list(
    body=models.CrmApplicationsFormsListFormsRequest(limit=50)
)
print(page.count, len(page.items))

form = client.crm.applications.forms.get(application_form_id="...")

# Walk every page transparently:
for form in client.crm.applications.forms.iterate():
    ...

client.close()

Async mirror:

from campai_client import AsyncCampaiClient

async with AsyncCampaiClient(api_key="...", organization_id="...") as client:
    form = await client.crm.applications.forms.get(application_form_id="...")
    async for f in client.crm.applications.forms.aiterate():
        ...

Configuration is resolved from constructor args or CAMPAI_* env vars (CAMPAI_API_KEY, CAMPAI_ORGANIZATION_ID, CAMPAI_MANDATE_ID, CAMPAI_BASE_URL).

Errors

HTTP status + campai's error envelope map to a typed hierarchy: CampaiAPIError (BadRequestError/AuthenticationError/PermissionError/NotFoundError/ServerError), plus CampaiValidationError (response didn't match the models) and CampaiConfigError.

How it's built (three layers)

1. Spec pipeline (spec/, scripts/): download → normalize/overlay → committed spec.
2. Generated layer (src/campai_client/_generated/, never hand-edited): 2 200+ pydantic v2 models via openapi-generator (models-only), plus the operations manifest (operations.py) built from the spec.
3. Facade + patch layer (src/campai_client/, durable): clients, resource namespaces, the patch pipeline, pagination, errors.

See docs/adr/0002 for how the runtime call path and resource tree are driven by the manifest.

The agent maintenance loop

When campai's spec changes, regenerate and let the tests triage the drift:

pixi run regen          # fetch + normalize + generate + manifest + format
pixi run test           # replay (VCR) tests

Triage failures:

• CampaiValidationError / import error → schema drift. Fix spec/overlay.yaml (schema-level, JSON-Pointer keyed), then pixi run regen.
• Behavioural mismatch (id-only response, aliasing, follow-up needed) → add or adjust a patch in src/campai_client/patches/ plus a test. Patches without tests are not allowed.
• New/removed operations → the manifest + resource namespaces regenerate automatically; add smoke tests for important new resources.

Cassettes are re-recorded (CAMPAI_RECORD=1 pixi run test-record) only when the real API behaviour changed.

Development

pixi manages the dev environment (the shipped package uses standard pyproject.toml deps).

pixi run check       # lint + typecheck + test  (CI aggregate)
pixi run lint        # ruff check
pixi run format      # ruff format
pixi run typecheck   # ty check
pixi run test        # pytest (VCR replay)
pixi run regen       # full regeneration (openapi-generator via pixi's JVM)

lefthook runs format/lint/typecheck on staged files and guards against hand-edits to _generated/**.

Regeneration (pixi run regen/generate) runs the openapi-generator JAR with a JVM supplied by pixi (openjdk) — no Docker required (it runs fine inside containers). The JAR is pinned by version + SHA-256 and cached under .cache/. Using or testing the shipped client needs neither Java nor the JAR.

Integration tests (real campai org)

Most tests use a mock transport. The integration tests (tests/resources/test_live.py) hit the real API and are recorded with VCR, then replayed from committed cassettes (no credentials on replay). Provide credentials via env vars to record:

CAMPAI_RECORD=1 CAMPAI_API_KEY=... CAMPAI_ORGANIZATION_ID=... \
  CAMPAI_MANDATE_ID=... pixi run test-record

Credentials are scrubbed from cassettes. See tests/README.md for the full flow and scrubbing tradeoffs.

License

Apache-2.0.