Python SDK

The Python SDK supports Python 3.10+ and has no runtime dependencies. It includes synchronous, thread-safe API calls, streamed uploads and downloads, safe retries, webhook verification, type hints for every method and an in-memory test client.

Choose a moderation flow

Use the SDK's moderate method to wait for a decision in the request, or its create method to queue moderation and let your app continue. Text and images support both flows. Audio and video always run asynchronously, even when using moderate.

For queued content, save the returned asset ID and set up a webhook for asset.classified (decision ready) and asset.failed (processing failed). Keep content unpublished until approved. You can retrieve the asset periodically instead of receiving events. The quick start demonstrates both flows.

Install

Moderate text

Keep credentials on your server. Without arguments, the client reads CALMLENS_API_KEY and CALMLENS_PROJECT_ID from the environment.

from calmlens import CalmLensClient

client = CalmLensClient()
asset = client.text.moderate(text="Great photo!", external_id="comment-42")
if asset["status"] == "approved":
    print("Approved", (asset.get("report") or {}).get("flagged"))

Methods take snake_case keyword arguments, or one options dict with the API's camelCase keys, such as client.assets.retrieve({"assetId": asset_id}). Responses are dicts with the API's camelCase field names.

Only publish content after an explicit approved result. An approved, flagged asset can still need review. Pending, rejected and failed requests must not be treated as approval. See Report format and Policies and actions.

Upload and deliver images

with open("photo.jpg", "rb") as file:
    queued = client.assets.create(
        name="photo.jpg",
        content={"data": file, "type": "image/jpeg"},
        visibility="private",
        processing_options={"keep_content_after_processing": True},
    )

# Retrieve later, or respond to an asset.classified webhook.
asset = client.assets.retrieve(asset_id=queued["id"])
if asset["status"] == "approved":
    signed = client.assets.create_signed_url(asset_id=asset["id"], resource="content", expires_in_seconds=60)
    print(signed["url"])

assets.create() queues processing. assets.moderate() waits for a decision for supported content; audio and video remain asynchronous.

data accepts bytes or a binary file object. Files are streamed in chunks rather than read into memory, and their size is measured automatically. For a stream that can't seek, such as a pipe, pass its sizeInBytes. The timeout applies to a silent connection, not to the whole transfer, so large uploads and downloads are not cut off.

Remote downloads are limited to 10 MB for images and 20 MB for other files. For multipart limits and sequential uploads, see Multipart uploads. Retain content only when you need previews, transforms or CDN delivery.

Operations

List pagination starts at zero. Use None to clear nullable fields or move assets to the project root; omit a field to leave it unchanged.

Concurrency and async code

One client is safe to share between threads, and reusing it keeps connections open between calls. In a forked worker process, such as a gunicorn or Celery worker, the client opens new connections rather than sharing the parent's. Calls are synchronous; from async code, run them in a thread:

asset = await asyncio.to_thread(client.text.moderate, text=body, external_id=comment_id)

Errors and retries

Every failure raises CalmLensError or one of its subclasses: CalmLensAPIError for API responses, CalmLensTimeoutError and CalmLensTransportError for network failures, and CalmLensValidationError for invalid arguments or configuration, detected before anything is sent. Each error has a stable code (constants in calmlens.error_codes), the HTTP status, the response's request_id to quote to support, retry_after, headers, and the is_not_found and is_retryable helpers.

from calmlens import CalmLensAPIError, error_codes

try:
    asset = client.assets.retrieve(asset_id="asset-id")
except CalmLensAPIError as error:
    if error.code != error_codes.ASSET_NOT_FOUND:
        print(error.code, error.status, error.request_id)
        raise
    asset = None

The client retries reads, and uploads, after network failures, timeouts and 429, 502, 503 or 504 responses, up to max_retries times (default 2) with a growing random delay, or after the Retry-After delay. Uploads send an Idempotency-Key, generated per call and reused for its retries, so a retried upload never creates a second asset; pass idempotency_key= to retry from another process. Other writes are not retried. Configure timeout (default 60 seconds of inactivity), connect_timeout (default 10 seconds) and max_retries on the client, or per call with client.with_options(timeout=300).

Moderation results

assets.moderate and text.moderate return the asset as a dict with the API's camelCase keys. id and status are always present. report is missing or None until classification finishes, so read it with asset.get("report"). Other optional fields, such as externalId, folderId and tags, appear only when they apply. Status and other enum values are plain strings, so a value added in a newer API version doesn't break your code; treat any status other than approved as not approved. See Report format.

Events feed

events.list returns {"events": [...], "cursor": ..., "hasMore": ...}. Without after, it returns no events and a cursor for the current time. events.iter_pages follows the cursor from after until hasMore is False; save the last page's cursor to resume.

cursor = client.events.list()["cursor"]  # no events; starts from now
for page in client.events.iter_pages(after=cursor, limit=50):
    for event in page["events"]:
        print(event["type"], event["facts"])
    cursor = page["cursor"]

Webhooks

Verify the exact raw request body, then parse it, with parse_webhook_event. It accepts only the raw str or bytes body: a body that was parsed and serialised again rarely has the same bytes, so the signature would not match.

import os
from calmlens import WebhookVerificationError, parse_webhook_event

def handle_webhook(raw_body: bytes, headers) -> int:
    try:
        event = parse_webhook_event(raw_body, headers=headers, secret=os.environ["CALMLENS_WEBHOOK_SECRET"])
    except WebhookVerificationError:
        return 401
    # Queue the work, keyed by event["id"], then respond quickly.
    return 204

Pass the raw body your framework received:

verify_webhook_signature takes the same arguments and returns True or False. Both read the x-calmlens-signature and x-calmlens-timestamp headers from headers, or take them as signature= and timestamp=. Timestamps more than 5 minutes old are rejected. Deliveries also carry x-calmlens-delivery, a unique ID for deduplicating retries. See Webhooks & Connectors.

Install the standalone CalmLens CLI to forward events locally:

calmlens listen --forward-to http://localhost:3000/webhook --env-file .env.local
calmlens trigger clean

Test without API calls

See Python testing for offline mock setup, request inspection and simulated failures.