Python

Install, configure, and use the aiinbx package — sync and async clients, pagination, raw responses, typed errors, and webhook verification.

The official Python client. Sync and async on httpx, fully typed, Python 3.10+.

pip install aiinbx

Getting started

from aiinbx import AIInbx

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

print(email["id"], email["thread_id"])

Use the client as a context manager so its connection pool is closed. A long-lived client constructed once at startup is fine too — just close it on shutdown.

Async

from aiinbx import AsyncAIInbx

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

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

AsyncAIInbx mirrors AIInbx method for method. Everything below applies to both, awaiting where appropriate.

Configuration

import os

client = AIInbx(
    api_key=os.environ["AI_INBX_API_KEY"],
    timeout=30.0,
    max_retries=2,
    # base_url="http://localhost:3000/api/v2",
)

Resources

Replying

Recipients, subject, and reply headers are inferred from the thread — pass the content as keyword arguments:

reply = client.threads.reply("thr_123", text="Sounds good — see you Thursday.")

For a reply assembled dynamically, pass a ThreadReplyParams dict instead. Idempotency keys, timeouts, and raw responses work the same either way.

Pagination

list returns one page; iter follows cursors lazily, keeping your filters:

page = client.emails.list(limit=100, status="scheduled")
print(page["data"], page["next_cursor"])

for email in client.emails.iter(direction="inbound"):
    print(email["subject"])

On AsyncAIInbx, iter is an async iterator:

async for email in client.emails.iter(direction="inbound"):
    print(email["subject"])

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 on the client.
print(client.last_request_id)

Errors

Non-2xx responses raise typed subclasses of APIStatusError, each carrying status_code, request_id, response headers, and the decoded body:

from aiinbx import APIStatusError, NotFoundError, RateLimitError

try:
    client.domains.retrieve("dom_missing")
except NotFoundError:
    domain = None
except RateLimitError as error:
    schedule_retry(error.headers.get("retry-after"))
except APIStatusError as error:
    print(error.status_code, error.body["code"], error.request_id)
    print(error.body.get("issues"))

All derive from AIInbxError, so one except AIInbxError catches everything the client raises. Codes and what to do about each are in Errors.

Webhooks

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,
)

if event["type"] == "email.bounced":
    for recipient in event["data"]["recipients"]:
        print(recipient)

event is a discriminated WebhookEvent union — a type checker narrows data from type.

Both t=...,v1=... signatures and a bare digest with a separate timestamp= argument are supported. Widen the replay window with tolerance= only as a diagnostic.

A module-level verify_webhook is exported too, for handlers that don’t have a client to hand:

from aiinbx import verify_webhook

event = verify_webhook(body, signature, secret)

FastAPI

from fastapi import FastAPI, Request, Response
from aiinbx import AIInbx, WebhookVerificationError

app = FastAPI()
client = AIInbx()

@app.post("/webhooks/aiinbx")
async def webhook(request: Request):
    body = await request.body()  # raw bytes, before any parsing

    try:
        event = client.webhooks.verify(
            body,
            request.headers["aiinbx-signature"],
            WEBHOOK_SECRET,
        )
    except WebhookVerificationError:
        return Response(status_code=400)

    if event["type"] == "email.received" and event["data"]["category"] == "human":
        await queue_reply(event["data"]["thread_id"])

    return Response(status_code=204)

Take Request rather than a Pydantic model — a declared body model parses the request before your handler runs, and the signature covers the exact bytes. More in Verifying.

Typing

The package ships py.typed, so mypy and Pyright see everything. Resources return TypedDicts that match the API’s JSON, and models are importable:

from aiinbx.models import WebhookEvent, ThreadReplyParams

def summarize(event: WebhookEvent) -> str:
    if event["type"] == "email.received":
        return event["data"]["snippet"]  # narrowed
    return event["type"]