Python SDK

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 TypedDicts.

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

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

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

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 = 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; 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 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 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.

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: 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).

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

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 for Flask and FastAPI examples.