Constaia
SDKs

Python SDK

Reference for the official constaia SDK for Python 3.9+: sync and async clients, inputs, options, pagination, errors, concurrency and webhooks.

constaia is the official Python SDK. It ships a synchronous client (Constaia) and an asynchronous one (AsyncConstaia) on top of httpx, responses typed with TypedDict, typed errors, retries, a client-side concurrency limit and webhook verification. Requires Python ≥ 3.9. Current version: 0.2.0.

pip install constaia

Getting started

verify.py
from constaia import Constaia

client = Constaia()  # reads CONSTAIA_API_KEY

analysis = client.analyze(
    "dni_valid.jpg",
    expect="es_dni",
    checks={"not_expired": True, "holder": {"full_name": "María García López"}},
    language="en",
)

verdict = analysis["verdict"]
if verdict and verdict["status"] == "valid":
    print(analysis["fields"]["document_number"]["value"])  # 12345678Z
elif verdict and verdict["status"] == "review":
    print("Manual review:", analysis["warnings"])
else:
    for reason in verdict["reasons"] if verdict else []:
        print(reason["code"], reason["message"])

A ck_test_… key spends no credits and answers by file name. See Test mode.

Async

verify_async.py
import asyncio

from constaia import AsyncConstaia


async def main() -> None:
    async with AsyncConstaia() as client:
        analysis = await client.analyze(file_url="https://example.com/id.jpg", expect="es_dni")
        print(analysis["verdict"])
        async for item in client.analyses.list(status="completed"):
            print(item["id"])


asyncio.run(main())

The async client has the same methods with await. Close it with await client.close() (or use async with).

Configuration

import httpx
from constaia import Constaia

client = Constaia(
    api_key="ck_live_...",  # default: CONSTAIA_API_KEY
    base_url="https://api.constaia.com",  # default; or CONSTAIA_BASE_URL
    timeout=60.0,  # seconds per attempt
    max_retries=2,  # 0 disables retries
    max_concurrency=4,  # requests in flight at once; the rest wait in the client
    default_headers={"X-Team": "ops"},
    http_client=httpx.Client(proxy="http://proxy:8080"),  # optional
)

Create one client per process and reuse it: it shares connections and the concurrency limit.

Accepted files

analyze() and classify() take exactly one of these inputs:

InputExample
Path (str or pathlib.Path)client.analyze("scans/id.jpg")
Bytesclient.analyze(data, filename="id.jpg")
Open binary fileclient.analyze(open("id.pdf", "rb")), Django UploadedFile, Flask FileStorage.stream
(filename, bytes) tupleclient.analyze(("id.jpg", data))
Remote URLclient.analyze(file_url="https://example.com/id.jpg")
Base64client.analyze(file_base64=b64, filename="id.jpg")

JPEG, PNG, WEBP, HEIC and PDF, up to 20 MB and 30 pages per synchronous analysis. The SDK reads the file into memory once so retries resend the same bytes.

Options

Names are the API's (snake_case). Since async is a Python keyword, the option is called async_.

analysis = client.analyze(
    "id.jpg",
    expect=["es_dni", "es_nie", "passport"],  # without expect there is no verdict
    checks={
        "not_expired": True,
        "min_age_years": 18,
        "holder": {"full_name": "María García López", "document_number": "12345678Z"},
        "require_fields": ["birth_date"],
    },
    extract=True,  # or your own JSON Schema (dict)
    storage="none",  # "none" | "temporary" | "persistent"
    ttl_hours=24,  # with "temporary"
    keep_results=True,
    async_=False,  # True → 202 and webhook
    export=["xlsx", "json"],
    metadata={"registration_id": "123"},
    language="en",
    idempotency_key="registration-123-id",  # optional: your own
    timeout=90,
)

Every option is explained in POST /v1/analyze and Checks.

processing

The SDK already accepts processing="sovereign" | "standard", but the parameter is still being rolled out in the API: until then, sending it returns 422 invalid_parameter. See Storage and privacy.

Methods

client.analyze(file, **options)
client.classify(file, expect="invoice")  # 0.2 credits

client.analyses.get("an_01J…")
page = client.analyses.list(limit=50, status="completed", type="es_dni", metadata={"event": "42"})
page.data, page.has_more  # first page
for a in client.analyses.list(type="es_dni"):  # iterating walks every page
    print(a["id"])
recent = client.analyses.list().to_list(200)
client.analyses.delete("an_01J…")
client.analyses.export("an_01J…", "xlsx", save_to="analysis.xlsx")  # returns the bytes

client.batches.create(items=[{"file_url": "https://example.com/1.pdf"}], options={"expect": "invoice"})
client.batches.create(files=["a.jpg", "b.pdf"], options={"expect": "payment_receipt"})
client.batches.get("bat_01J…")

client.document_types.list(language="en")
client.document_types.get("medical_certificate_sport")

endpoint = client.webhook_endpoints.create(
    url="https://example.com/webhooks/constaia",
    events=["analysis.completed", "analysis.review_required"],
)
endpoint["secret"]  # whsec_… — only shown now
client.webhook_endpoints.list()
client.webhook_endpoints.get("we_01J…")
client.webhook_endpoints.delete("we_01J…")

client.balance()
client.usage(from_="2026-09-01", to="2026-09-30")

analyses.list() returns the first page (.data, .has_more, .next_page()); iterating it makes the SDK fetch the next pages with starting_after. In the async client, await gives the first page and async for walks them all. See Pagination.

Types

Responses are plain dicts shaped like the API JSON, typed with TypedDict (constaia.types.Analysis, Classification, Batch, WebhookEndpoint…). mypy and pyright check field names; nothing is validated at runtime, so new API fields never break older SDK versions.

from constaia.types import Analysis


def is_ok(a: Analysis) -> bool:
    return a["verdict"] is not None and a["verdict"]["status"] == "valid"

Errors

Every exception extends ConstaiaError, with message, status, type, code, param and request_id.

ClassWhen
InvalidRequestError400, 409, 413, 415, 422 (code, param).
AuthenticationError401.
InsufficientCreditsError402.
PermissionDeniedError403.
NotFoundError404.
RateLimitError429 after retries (retry_after).
APIError5xx (for example 503 live_mode_unavailable).
APIConnectionError / APITimeoutErrorNetwork problems.
WebhookVerificationErrorInvalid webhook signature.
from constaia import (
    ConstaiaError,
    InsufficientCreditsError,
    InvalidRequestError,
    RateLimitError,
)

try:
    client.analyze("id.jpg", expect="es_dni")
except InvalidRequestError as e:
    print(e.code, e.param)  # e.g. unsupported_file_type, file
except InsufficientCreditsError:
    notify_billing()
except RateLimitError as e:
    print(f"retry in {e.retry_after} s")
except ConstaiaError as e:
    print(e.status, e.type, e.request_id)

Every code is listed in Errors.

Retries, concurrency and limits

  • Retries 429, 408, 409 idempotency_in_progress, 5xx (except 501) and network errors up to max_retries times, waiting Retry-After (up to 60 s) or exponential backoff with jitter.
  • Sends the same Idempotency-Key on every retry: a retried analysis is never charged twice.
  • Keeps at most max_concurrency requests in flight per client (threads or asyncio tasks); the rest wait.
  • Exposes the latest limit headers:
client.balance()
client.last_rate_limit  # RateLimit(limit=10, remaining=9, reset=1, policy='10;w=1', retry_after=None)
client.last_request_id  # "req_…": quote it when contacting support

With several processes or workers, the per-key limit is shared: keep max_concurrency × processes close to your requests-per-second limit. See Rate limits.

Webhooks

import constaia

event = constaia.webhooks.verify(raw_body, headers, "whsec_...")  # raises WebhookVerificationError
if event["type"] == "analysis.completed":
    save(event["data"])

Always pass the raw body (request.body in Django, request.get_data() in Flask, await request.body() in FastAPI). Default tolerance: 300 s (tolerance=). For your tests, constaia.webhooks.sign(payload, secret) builds valid headers. More in Webhooks.

Framework guides

Next steps

On this page