# Python SDK

Source: https://agentboxd.com/docs/python-sdk

> Sync and async clients for Python 3.9+, fully typed, with httpx as the only dependency.

shell:

```bash
pip install agentboxd
```

The package has a sync client (`Agentboxd`) and an async one (`AsyncAgentboxd`). Both read `AGENTBOXD_API_KEY` and `AGENTBOXD_BASE_URL` (default `https://api.agentboxd.com`) and work as context managers. Responses are plain dicts described by `TypedDict`s.

agent.py:

```python
from agentboxd import Agentboxd

with Agentboxd() as mr:  # reads AGENTBOXD_API_KEY and AGENTBOXD_BASE_URL
    inbox = mr.inboxes.create(
        username="support-bot",
        display_name="Support Bot",
        client_id="support-bot",  # idempotent
    )
    print(inbox["address"])  # support-bot@homingbox.net

    # Wait for the next email (long-poll, up to 60 s). None on timeout.
    msg = mr.messages.wait(inbox["id"], timeout=60)
    if msg:
        print(msg["extracted_text"])  # the new part of the reply only
        mr.messages.reply(inbox["id"], msg["id"], text="Thanks, on it.")

```

Temporary inboxes: `mr.inboxes.create_temporary(ttl_seconds=900)`. Custom domains: `mr.domains.create/list/get/verify/update/delete/rotate_dkim/activate_dkim`, and `mr.inboxes.create(domain=...)`. Kill switch: `mr.inboxes.pause(inbox_id, reason=...)` and `mr.inboxes.resume(inbox_id)`. Deliverability summary: `mr.deliverability()`.

## Async

async_agent.py:

```python
import asyncio
from agentboxd import AsyncAgentboxd


async def main() -> None:
    async with AsyncAgentboxd() as mr:
        inbox = await mr.inboxes.create()
        msg = await mr.messages.wait(inbox["id"], timeout=30)


asyncio.run(main())
```

## Contacts, knowledge and drafts

The same resources as the TypeScript client, with snake_case names: `mr.contacts.list/get/by_address/update`, `mr.inboxes.update`, `mr.threads.update`, `mr.knowledge.list/create/get/update/delete/search`, `mr.drafts.create/list/list_all/get/update/delete/send/schedule/cancel`, `mr.messages.draft_reply` (with `save=True`), and for [document extraction](https://agentboxd.com/docs/document-extraction) `mr.messages.attachment_text(message_id, attachment_id, offset=None, max_chars=None)` and `mr.messages.extract_attachment(message_id, attachment_id, schema=..., instructions=None)`.

contacts.py:

```python
c = mr.contacts.by_address("priya.n@gmail.com")
print(c["notes"], c["metadata"].get("crm_id"))

mr.contacts.update(
    c["id"],
    notes=(c["notes"] or "") + "
Asked about order 1042 on 25 Sep.",
    metadata={"crm_id": "hs_48213", "plan": "wholesale"},
    add_labels=["customer"],
)

```

knowledge.py:

```python
mr.knowledge.create(
    title="Shipping times",
    body="- Germany: 1–2 working days (DHL)
- EU: 3–5 working days",
    inbox_id=inbox["id"],
)

hits = mr.knowledge.search("how long does shipping to France take", inbox_id=inbox["id"])
for h in hits["data"]:
    print(h["title"], h["rank"], h["snippet"])

```

draft.py:

```python
draft = mr.messages.draft_reply(msg["id"], instructions="Keep it short.")
print(draft["text"], [c["title"] for c in draft["citations"]])
# review it, then:
mr.messages.reply(msg["inbox_id"], msg["id"], text=draft["text"])
```

## Agent-to-agent messaging

`mr.messages.send(..., data={...}, type="task")` and `reply(..., data=..., type=...)` send an [agent message](https://agentboxd.com/docs/agent-messaging); `list`, `wait` and `search` take `channel=` and `type=`. `agentboxd.identity.verify_agent_message(message, recipient=..., replay_cache=...)` (and `averify_agent_message`) checks a message’s signature outside Agentboxd; it needs the `identity` extra (`pip install 'agentboxd[identity]'`).

## Sign in with Agentboxd

`mr.identity.token(inbox_id, audience=..., nonce=None, scope=None, expires_in=None)` mints an [identity token](https://agentboxd.com/docs/agent-identity) for one app (needs `identity:sign`); pass `identity_id=` instead of `inbox_id` for an identity without a mailbox, which `mr.identities.create/list/get/update/delete/pause/resume` manage. Apps are managed with `mr.identity.clients.list/create/get/update/delete/rotate_secret`, and an inbox’s sign-in switch and history with `mr.identity.inbox.get/update/sign_ins`.

## Realtime events

`AsyncAgentboxd.stream(inbox_ids=, event_types=, payload=, since=)` is an async iterator over the [event stream](https://agentboxd.com/docs/realtime-events) that reconnects and resumes on its own. It needs the `stream` extra, `pip install 'agentboxd[stream]'` (adds `websockets`). Both clients have `stream_token()` for other WebSocket clients.

listen.py:

```python
async with AsyncAgentboxd() as mr:
    async for event in mr.stream(event_types=["message.received"]):
        print(event["data"]["message"]["subject"])
```

## Work queue

The [claim/ack queue](https://agentboxd.com/docs/api#claim-and-ack-work-queue): `mr.messages.claim(inbox_id, limit=, lease_seconds=, consumer=, wait=, enriched=, analysis=, since=)`, `mr.messages.ack(message_id, lease_id, mark_read=None)`, `nack(message_id, lease_id, delay_seconds=None)` and `extend(message_id, lease_id, lease_seconds=None)`. `mr.inboxes.consume(inbox_id, handler, concurrency=1, lease_seconds=300, stop=None)` runs the loop: it extends the lease while the handler runs, acks when it returns and nacks with backoff when it raises. The sync client runs handlers in threads and stops when `stop` (a `threading.Event`) is set; `AsyncAgentboxd` takes an async handler and an `asyncio.Event`. A handler with two parameters also gets a `ConsumeContext` (`lease_id`, `delivery_count`, `lost`, `extend`).

worker.py:

```python
import threading
from agentboxd import Agentboxd

mr = Agentboxd()
stop = threading.Event()


def handle(message):
    process_ticket(message)  # acked when this returns, nacked (retried later) if it raises


mr.inboxes.consume(inbox_id, handle, concurrency=4, enriched=True, stop=stop)
```

## Pagination

paginate.py:

```python
from agentboxd import iter_all

for msg in iter_all(mr.messages.list, inbox["id"], is_read=False):
    print(msg["subject"])
```

## Errors

Non-2xx responses raise `MailroomError` (alias `AgentboxdError`; `status`, `code`, `message`), or a subclass: `BadRequestError` (400), `AuthenticationError` (401), `PermissionDeniedError` (403), `NotFoundError` (404), `UnprocessableEntityError` (422), `RateLimitError` (429), `InternalServerError` (5xx). Network failures raise `APIConnectionError` with `status` 0.

## Webhooks

`verify_webhook(signature, timestamp, raw_body, secret)` returns `False` on bad input and never raises. See [Webhooks](https://agentboxd.com/docs/webhooks) for Flask and FastAPI examples.