Python SDK

creb-sdk wraps the public /v1 REST surface with a sync httpx-based client and Pydantic v2 models. Source lives at packages/sdk-python/ in the platform repo.

Install

pip install creb-sdk
# or with dev extras (pytest, mypy, ruff, respx)
pip install "creb-sdk[dev]"

Requires Python 3.10 or newer.

Construct

from creb_sdk import CrebClient

with CrebClient(
    base_url="https://api.insuredaudit.com",
    api_key=os.environ["CREB_API_KEY"],
    tenant_id="acme",          # optional default
    timeout_s=30.0,             # optional, default 30s
    max_attempts=3,             # optional retry budget
) as client:
    ...

Use the client as a context manager so the underlying httpx.Client is closed cleanly. If your app has its own lifecycle, call client.close() manually.

Method map

Method	Underlying route
`healthz()`	GET /v1/healthz
`ingest_event()`	POST /v1/events
`ingest_batch()`	POST /v1/events (batch shape)
`validate_event()`	POST /v1/events/validate
`member_timeline()`	GET /v1/members/{id}/timeline
`search_events()`	POST /v1/search
`upload_document()`	POST /v1/documents
`download_document()`	GET /v1/documents/{id} (streamed)
`request_bundle()`	POST /v1/bundles
`get_bundle()`	GET /v1/bundles/{id}
`list_bundles()`	GET /v1/bundles
`download_bundle()`	GET /v1/bundles/{id}/download (streamed)
`request_bundle_and_wait()`	POST then poll until sealed
`create_webhook()`	POST /v1/webhooks
`list_webhooks()`	GET /v1/webhooks
`rotate_webhook_secret()`	POST /v1/webhooks/{id}/rotate-secret
`delete_webhook()`	DELETE /v1/webhooks/{id}
`list_rules()`	GET /v1/rules

Error handling

from creb_sdk import CrebApiError

try:
    client.ingest_event(envelope)
except CrebApiError as e:
    if e.code == "RATE_LIMITED":
        ...
    elif e.status == 400:
        print("validation:", e.body)
    else:
        raise

Streamed downloads

Both download_document() and download_bundle() return an open httpx.Response. Iterate .iter_bytes() and close the response when done:

with client.download_bundle(bundle_id) as resp:
    with open("bundle.zip", "wb") as f:
        for chunk in resp.iter_bytes():
            f.write(chunk)

Examples

Five runnable examples ship in packages/sdk-python/examples/:

01_ingest_event.py
02_request_bundle.py
03_search_timeline.py
04_manage_webhooks.py
05_list_rules.py