3.0.0

AI Inbx Python SDK

Website · Documentation

The official typed Python client for AI Inbx API v2. It supports synchronous and asynchronous applications, bounded retries, per-request timeouts, raw responses, idempotency keys, and signed webhook verification.

pip install aiinbx

Send an email

from aiinbx import AIInbx

with AIInbx() as client:  # reads AI_INBX_API_KEY
    email = client.emails.send(
        {
            "from_": {"name": "AI Inbx", "address": "hello@example.com"},
            "to": ["ada@example.com"],
            "subject": "Hello",
            "text": "Sent with AI Inbx",
        },
        idempotency_key="welcome-ada-v1",
    )
    print(email["id"])

The default base URL is https://api.aiinbx.com/api/v2. Use base_url= to point the client at a local or self-hosted API. The client retries network failures and HTTP 408, 409, and 5xx responses up to two times when a retry cannot act twice — a GET, PUT or DELETE, or a request with an idempotency_key — and a 429 always, honoring Retry-After.

Async

from aiinbx import AsyncAIInbx

async with AsyncAIInbx() as client:
    page = await client.threads.list(limit=20)

Every sync resource has the same methods on AsyncAIInbx:

• spaces: list, iter, create, retrieve, update, delete
• api_keys: list, iter, create, delete
• emails: send, list, iter, retrieve, update, send_draft, cancel
• threads: list, iter, retrieve, iter_messages, reply, forward
• domains: list, iter, create, retrieve, update, diagnostics, delete, verify
• mailboxes: list, iter, retrieve, connect, disconnect, sync
• oauth_apps: list, iter, create, retrieve, update, delete
• webhook_endpoints: list, iter, create, retrieve, update, delete, rotate_secret, test, list_deliveries, iter_deliveries, retry_deliveries
• events: list, iter, retrieve
• suppressions: list, iter, add, retrieve, remove
• pacing_rules: list, iter, create, retrieve, update, delete, retrieve_spread, update_spread
• pacing: retrieve, release
• attachments: download, content

Every operation takes its request body as one dictionary — a TypedDict from aiinbx.models, named after the API's schema (SendEmailRequest, UpdateDomainRequest, …) — and its query filters as keyword arguments. The one spelling difference from the wire is from_, since from is a Python keyword; the SDK sends it as from. Resources, models and these method signatures are generated from the API contract, so a change to the API is a change here on the next release.

Reply and pagination

Replies infer the sender, recipients, subject, and RFC reply headers from the thread. Supply only the content for the common case:

reply = client.threads.reply(
    "thr_123",
    {"text": "Sounds good — see you Thursday."},
    idempotency_key="reply-thr-123-v1",
)

# `iter()` follows cursors lazily. Async resources support `async for`.
for email in client.emails.iter(direction="inbound"):
    print(email["subject"])

async for message in async_client.threads.iter_messages("thr_123"):
    print(message["snippet"])

Sends, replies and forwards take an idempotency_key; every call takes a timeout and works under with_raw_response.

Spaces

A space groups a customer's domains, mailboxes and mail. Pass space_id when creating a domain, mailbox, suppression or pacing rule; emails and threads inherit the space of the domain or mailbox they went through. Lists of those take a space= filter: pass a space ID for that space, "none" for only the workspace's own resources (space_id of None), or omit it for both the workspace and all spaces. API keys and webhook endpoints belong to the workspace: one key reaches every space, and one endpoint receives every space's events, each stamped with its space_id. Spaces are optional — a workspace that never creates one sees space_id of None everywhere.

external_id is your own id for the customer — unique per workspace, so a second create for the same customer fails with 409 external_id_taken — and spaces.list(external_id=...) finds the space by it.

space = client.spaces.create({"name": "Acme", "external_id": "cus_8812"})
# Later, from your side of the map:
same = client.spaces.list(external_id="cus_8812")["data"][0]
# One label under a wildcard you own (`*.mail.example.com`) is a subdomain: no DNS to publish.
client.domains.create({"name": "acme.mail.example.com", "space_id": space["id"]})
client.emails.send(
    {
        "from_": "Acme <hello@acme.mail.example.com>",
        "to": ["ada@example.com"],
        "subject": "Welcome",
        "text": "Sent from Acme's own subdomain.",
    }
)

Raw responses and request IDs

response = client.with_raw_response.emails.retrieve("eml_123")
print(response.status_code, response.request_id)
email = response.json()

# The most recently completed response is also exposed on the client.
print(client.last_request_id)

Non-success responses raise typed subclasses of APIStatusError. These include status_code, request_id, response headers, and the decoded body.

Verify webhooks

Always pass the unmodified request body. The verifier signs <unix_timestamp>.<payload> with HMAC-SHA256 and rejects payloads older than five minutes by default.

event = client.webhooks.verify(
    request_body,
    request.headers["AIInbx-Signature"],
    webhook_secret,
)

# `event` is the `aiinbx.WebhookEvent` union, generated from the API contract.
# Type checkers narrow `data` from the event type.
if event["type"] == "email.bounced":
    for recipient in event["data"]["recipients"]:
        print(recipient)

Both t=...,v1=... signatures and a bare digest plus the separate timestamp= argument are supported. Override the replay window with tolerance= only when necessary.