Sema Python SDK

Official Python SDK for Sema - Universal ingestion infrastructure for AI Agents.

Installation

pip install sema-sdk

Quick Start

API Client

from sema_sdk import SemaClient

with SemaClient(api_key="sk_live_...") as client:
    # Create an inbox
    inbox = client.create_inbox(
        name="My Inbox",
        webhook_url="https://example.com/webhooks/sema",
    )
    print(f"Created inbox: {inbox.id}")

    # Upload content
    item = client.upload_item(
        inbox_id=inbox.id,
        file=b"Hello, World!",
        sender_address="sender@example.com",
        subject="Test Upload",
    )
    print(f"Created item: {item.id}")

    # Check deliveries
    deliveries = client.get_item_deliveries(item.id)
    for d in deliveries.deliveries:
        print(f"Delivery {d.id}: {d.status}")

Async API Client

import asyncio

from sema_sdk import AsyncSemaClient

async def main() -> None:
    async with AsyncSemaClient(api_key="sk_live_...") as client:
        # Create an inbox
        inbox = await client.create_inbox(
            name="My Inbox",
            webhook_url="https://example.com/webhooks/sema",
        )
        print(f"Created inbox: {inbox.id}")

        # Upload content
        item = await client.upload_item(
            inbox_id=inbox.id,
            file=b"Hello, World!",
            sender_address="sender@example.com",
            subject="Test Upload",
        )
        print(f"Created item: {item.id}")

        # Check deliveries
        deliveries = await client.get_item_deliveries(item.id)
        for d in deliveries.deliveries:
            print(f"Delivery {d.id}: {d.status}")

asyncio.run(main())

Webhook Verification

from sema_sdk import WebhookVerifier, WebhookVerificationError

verifier = WebhookVerifier(secret="whsec_...")

def handle_webhook(request):
    try:
        event = verifier.verify(
            payload=request.body,
            headers=request.headers,
        )
        
        # Use event.webhook_id as idempotency key
        print(f"Received: {event.webhook_id}")
        print(f"Item ID: {event.payload.item_id}")
        print(f"Event type: {event.payload.event_type}")
        
        return Response(status=200)
        
    except WebhookVerificationError as e:
        return Response(status=400, body=str(e))

API Reference

SemaClient

Synchronous client for the Sema API.

client = SemaClient(
    api_key="sk_live_...",
    base_url="https://api.withsema.com",  # optional
    timeout=30.0,  # optional, in seconds
    max_retries=3,  # optional, set to 0 to disable retries
)

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.

AsyncSemaClient

Async client for the Sema API.

from sema_sdk import AsyncSemaClient

async with AsyncSemaClient(
    api_key="sk_live_...",
    base_url="https://api.withsema.com",  # optional
    timeout=30.0,  # optional, in seconds
    max_retries=3,  # optional, set to 0 to disable retries
) as client:
    inboxes = await client.list_inboxes(limit=10)

Inbox Methods

• create_inbox(name, *, description=None, webhook_url=None, ...) - Create a new inbox
• get_inbox(inbox_id) - Get an inbox by ID
• update_inbox(inbox_id, *, name=None, webhook_url=None, ...) - Update an inbox
• list_inboxes(*, limit=None, offset=None) - List inboxes with optional pagination; returns InboxList (inboxes, total)

Item Methods

• upload_item(inbox_id, file, *, sender_address, ...) - Upload content to an inbox
• get_item(item_id) - Get an item by ID
• list_items(inbox_id, *, limit=None, offset=None) - List items in an inbox

Delivery Methods

• get_item_deliveries(item_id) - Get deliveries for an item

Attachment Methods

• get_item_attachments(item_id) - Get attachments for an item with presigned download URLs

Pagination Iterators

For large result sets, use async iterators to automatically paginate:

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

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

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

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:

• method - HTTP method (GET, POST, etc.)
• url - Request URL
• attempt - Attempt number (1-based, increases on retries)
• status - HTTP status code (after_response only)
• duration_ms - Request duration in milliseconds (after_response only)

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

WebhookVerifier

Verifies Standard Webhooks signatures.

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

event = verifier.verify(payload, headers)
# Returns WebhookEvent with:
#   - webhook_id: str (use as idempotency key)
#   - timestamp: int
#   - payload: WebhookPayload

Email Utilities

resolve_email_inline_images

HTML emails embed inline images using cid: references. This utility replaces them with download URLs:

from sema_sdk import SemaClient, WebhookVerifier, resolve_email_inline_images

client = SemaClient(api_key="sk_live_...")
verifier = WebhookVerifier(secret="whsec_...")

def handle_webhook(request):
    event = verifier.verify(request.body, request.headers)
    
    # Get the HTML body with cid: references
    content_summary = event.payload.deliverable.content_summary
    body_html = (content_summary.body_html or "") if content_summary else ""
    
    # Fetch attachments with presigned download URLs
    attachments = client.get_item_attachments(event.payload.item_id).attachments
    
    # Replace cid: references with actual URLs
    resolved_html = resolve_email_inline_images(body_html, attachments)
    
    # resolved_html now has working image URLs
    return resolved_html

partition_email_attachments

Email clients like Gmail set content_id on ALL attachments, not just inline ones. This utility correctly identifies which attachments are truly inline (embedded in HTML) vs which should be listed separately:

from sema_sdk import SemaClient, partition_email_attachments, resolve_email_inline_images

client = SemaClient(api_key="sk_live_...")

def handle_webhook(request):
    # ... verify webhook ...
    
    content_summary = event.payload.deliverable.content_summary
    body_html = (content_summary.body_html or "") if content_summary else ""
    attachments = client.get_item_attachments(event.payload.item_id).attachments
    
    # Resolve inline images in HTML
    resolved_html = resolve_email_inline_images(body_html, attachments)
    
    # Partition: inline images vs files to list separately
    inline, non_inline = partition_email_attachments(body_html, attachments)
    
    # inline attachments are already displayed in resolved_html
    # non_inline attachments should be listed separately
    for att in non_inline:
        print(f"Attachment: {att.filename} ({att.content_type})")

Exceptions

• SemaError - Base exception for all SDK errors
• SemaAPIError - Error returned by the API (has status_code and response_body)
• AuthenticationError - Invalid or missing API key (401)
• NotFoundError - Resource not found (404)
• RateLimitError - Rate limit exceeded (429)
• WebhookVerificationError - Signature or timestamp validation failed

Examples

See the examples/ directory for runnable examples:

• send_content.py - Create inbox and upload content
• receive_webhook.py - Flask webhook receiver with signature verification

Requirements

• Python 3.10+
• httpx
• pydantic

License

MIT