Python

Installation¶

pip install sema-sdk

API Client¶

Initialize¶

In production, prefer environment variables (SEMA_API_KEY, SEMA_BASE_URL) and omit the key (and optionally base URL) below.

from sema_sdk import SemaClient

client = SemaClient(
    api_key="sk_live_...",  # optional if SEMA_API_KEY is set
    base_url="https://dev-api.withsema.com",  # optional, falls back to SEMA_BASE_URL
    timeout=30.0,  # optional, in seconds
    max_retries=3,  # optional, set to 0 to disable retries
)

You can omit api_key when SEMA_API_KEY is set; omit base_url when SEMA_BASE_URL is set.

The client requires HTTPS in production: it rejects live keys (sk_live_...) with a non-HTTPS base URL and warns for other HTTP base URLs (e.g. localhost).

Use as a context manager for automatic cleanup:

with SemaClient(api_key="sk_live_...") as client:
    # ... use client

The client automatically retries requests on transient failures (5xx errors, network errors, rate limits) with exponential backoff. Rate-limited requests (429) respect the Retry-After header when present.

Create an Inbox¶

inbox = client.create_inbox(
    name="Support Inbox",
    webhook_url="https://example.com/webhooks/sema",
    webhook_secret="whsec_...",  # optional, auto-generated if omitted
)
print(inbox.id)

Upload Content¶

# From bytes
item = client.upload_item(
    inbox_id=inbox.id,
    file=b"Hello, World!",
    sender_address="sender@example.com",
    subject="Test Upload",
    filename="hello.txt",
)

# From file
with open("document.pdf", "rb") as f:
    item = client.upload_item(
        inbox_id=inbox.id,
        file=f,
        sender_address="sender@example.com",
    )

print(item.id, item.status, item.is_duplicate)

Get Item Details¶

item = client.get_item(item_id="...")
print(item.sender_address, item.subject, item.auth_decision)

List Items¶

result = client.list_items(inbox_id="...", limit=10)
for item in result.items:
    print(item.id, item.status)

List Inboxes¶

result = client.list_inboxes(limit=10, offset=0)
print(result.total)
for inbox in result.inboxes:
    print(inbox.id, inbox.name)

list_inboxes(limit=..., offset=...) returns an InboxList with inboxes and total.

Pagination Iterators¶

For large result sets, use iterators to automatically paginate:

# Iterate through all items in an inbox
for item in client.iter_items(inbox_id):
    print(item.id, item.status)

# Iterate through all inboxes
for inbox in client.iter_inboxes():
    print(inbox.id, inbox.name)

# Custom page size
for item in client.iter_items(inbox_id, page_size=50):
    print(".", end="")

Check Deliveries¶

result = client.get_item_deliveries(item_id="...")
for delivery in result.deliveries:
    print(delivery.status, delivery.attempt_count)
    for attempt in delivery.attempts:
        print(f"  Attempt {attempt.attempt_number}: {attempt.status_code}")

Async Client¶

For async applications, use AsyncSemaClient:

from sema_sdk import AsyncSemaClient

async with AsyncSemaClient(api_key="sk_live_...") as client:
    inbox = await client.create_inbox(name="Support Inbox")
    item = await client.get_item(item_id="...")

    # Async pagination
    async for item in client.iter_items(inbox_id):
        print(item.id)

AsyncSemaClient has the same methods as SemaClient but all are async and require await.

Webhook Verification¶

Initialize¶

from sema_sdk import WebhookVerifier

verifier = WebhookVerifier(
    secret="whsec_...",
    tolerance_seconds=300,  # optional, default 5 minutes
)

Verify and Parse¶

from sema_sdk import WebhookVerificationError

def handle_webhook(request):
    try:
        event = verifier.verify(
            payload=request.body,
            headers=request.headers,
        )
    except WebhookVerificationError as e:
        return Response(status=400, body=str(e))

    # Use webhook_id as idempotency key
    if already_processed(event.webhook_id):
        return Response(status=200)

    # Process the event
    print(event.payload.event_type)   # "ITEM_READY"
    print(event.payload.item_id)
    print(event.payload.inbox_id)
    print(event.payload.payload_mode) # "full" or "thin"

    # Access deliverable data
    print(event.payload.deliverable.raw_ref)
    if event.payload.deliverable.sender:
        print(event.payload.deliverable.sender.address)

    mark_processed(event.webhook_id)
    return Response(status=200)

Error Handling¶

from sema_sdk import (
    SemaError,           # Base exception
    SemaAPIError,        # API errors (has status_code, response_body)
    AuthenticationError, # 401 - Invalid API key
    NotFoundError,       # 404 - Resource not found
    RateLimitError,      # 429 - Rate limit exceeded
    WebhookVerificationError,  # Signature/timestamp invalid
)

try:
    item = client.get_item("nonexistent")
except NotFoundError:
    print("Item not found")
except SemaAPIError as e:
    print(f"API error {e.status_code}: {e.message}")

Observability Hooks¶

Add optional hooks for logging, tracing, or metrics:

def before_request(ctx):
    print(f"→ {ctx['method']} {ctx['url']} (attempt {ctx['attempt']})")

def after_response(ctx):
    print(f"← {ctx['status']} in {ctx['duration_ms']:.1f}ms")

def on_error(ctx, error):
    print(f"✗ {ctx['method']} {ctx['url']}: {error}")

client = SemaClient(
    api_key="sk_live_...",
    on_before_request=before_request,
    on_after_response=after_response,
    on_error=on_error,
)

Hook context includes:

Field	Description	Available In
`method`	HTTP method (GET, POST, etc.)	All hooks
`url`	Request URL	All hooks
`attempt`	Attempt number (1-based)	All hooks
`status`	HTTP status code	`on_after_response`
`duration_ms`	Request duration in milliseconds	`on_after_response`

Hooks can be sync or async. Errors raised in hooks are silently ignored to prevent breaking SDK functionality.

Inline Images¶

Emails with pasted screenshots or inline images include cid: references in the HTML body. To render these images:

# Get attachments with presigned URLs
attachments = client.get_item_attachments(item_id)

# Build lookup from content_id to download URL
cid_to_url = {
    att.content_id: att.download_url
    for att in attachments.attachments
    if att.content_id and att.download_url
}

# Replace cid: references in HTML body
html = event.payload.deliverable.content_summary.body_html
if html:
    for cid, url in cid_to_url.items():
        html = html.replace(f"cid:{cid}", url)

Type Reference¶

All responses are Pydantic models with full type hints:

Inbox - Inbox configuration and status
InboxList - List response with inboxes and total
Item - Item details including sender, status, metadata
Delivery - Delivery status and attempts
WebhookEvent - Verified webhook with webhook_id, timestamp, payload
WebhookPayload - Parsed webhook body with deliverable data