Python SDK

creb-sdk wraps the public /v1 REST surface with a sync httpx-based client and Pydantic v2 models.

Availability. The SDK is provided to customers and pilot partners on request. It is not published to PyPI. The REST API and its OpenAPI specification are public.

Install

Install the wheel your InsuredAudit contact provides:

pip install ./creb_sdk-<version>-py3-none-any.whl
# or with dev extras (pytest, mypy, ruff, respx)
pip install "./creb_sdk-<version>-py3-none-any.whl[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 are provided with the SDK on request:

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

Getting started

API

SDKs

Concepts

Operations

Python SDK

Install

Construct

Method map

Error handling

Streamed downloads

Examples