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 constaiaGetting started
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
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:
| Input | Example |
|---|---|
Path (str or pathlib.Path) | client.analyze("scans/id.jpg") |
| Bytes | client.analyze(data, filename="id.jpg") |
| Open binary file | client.analyze(open("id.pdf", "rb")), Django UploadedFile, Flask FileStorage.stream |
(filename, bytes) tuple | client.analyze(("id.jpg", data)) |
| Remote URL | client.analyze(file_url="https://example.com/id.jpg") |
| Base64 | client.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.
| Class | When |
|---|---|
InvalidRequestError | 400, 409, 413, 415, 422 (code, param). |
AuthenticationError | 401. |
InsufficientCreditsError | 402. |
PermissionDeniedError | 403. |
NotFoundError | 404. |
RateLimitError | 429 after retries (retry_after). |
APIError | 5xx (for example 503 live_mode_unavailable). |
APIConnectionError / APITimeoutError | Network problems. |
WebhookVerificationError | Invalid 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 tomax_retriestimes, waitingRetry-After(up to 60 s) or exponential backoff with jitter. - Sends the same
Idempotency-Keyon every retry: a retried analysis is never charged twice. - Keeps at most
max_concurrencyrequests 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 supportWith 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
Django and DRF
Upload view, APIView and a csrf_exempt webhook.
Flask
request.files and a webhook with get_data().
FastAPI
UploadFile with the async client.
Celery
Tasks with retries, idempotency and concurrency.
Next steps
PHP SDK
Reference for constaia/constaia-php on PHP 8.1+: Composer install, inputs, options, methods, pagination, exceptions, retries and webhooks.
Upload widget
Reference for @constaia/widget, the <constaia-upload> web component that captures documents in the browser without exposing your key. React, Vue and any framework.