Inkbox Python SDK

API-first communication infrastructure for AI agents — email, phone, encrypted vault, and identities.

Install & Init

pip install inkbox

Always use the context manager — it manages the underlying HTTP session:

from inkbox import Inkbox

with Inkbox(api_key="ApiKey_...") as inkbox:
    ...

Constructor: Inkbox(apikey, baseurl="https://inkbox.ai";, timeout=30.0)

Core Model

Inkbox (admin-only client)
├── .create_identity(handle)  → AgentIdentity
├── .get_identity(handle)     → AgentIdentity
├── .list_identities()        → list[AgentIdentitySummary]
├── .mailboxes                → MailboxesResource
├── .phone_numbers            → PhoneNumbersResource
├── .texts                    → TextsResource
├── .imessages                → IMessagesResource
├── .imessage_contact_rules   → IMessageContactRulesResource
├── .mail_identity_contact_rules  → MailIdentityContactRulesResource   (keyed by agent_handle)
├── .phone_identity_contact_rules → PhoneIdentityContactRulesResource  (keyed by agent_handle)
├── .signing_keys             → SigningKeysResource  (per-identity: create_or_rotate/get_status)
├── .mail_contact_rules       → MailContactRulesResource   (DEPRECATED — per-mailbox)
├── .phone_contact_rules      → PhoneContactRulesResource  (DEPRECATED — per-number)
├── .sms_opt_ins              → SmsOptInsResource
├── .contacts                 → ContactsResource  (.facts, .correspondence, .access, .vcards)
├── .notes                    → NotesResource     (.access)
├── .vault                    → VaultResource
├── .whoami()                 → WhoamiResponse
└── .create_signing_key()     → SigningKey  (DEPRECATED — org-level; use .signing_keys)

AgentIdentity (identity-scoped helper)
├── .mailbox                 → IdentityMailbox | None
├── .phone_number            → IdentityPhoneNumber | None
├── .mail_filter_mode / .phone_filter_mode → FilterMode
├── .credentials             → Credentials  (requires vault unlocked)
├── .list_mail_contact_rules() / .create_mail_contact_rule(...) / .get_/.update_/.delete_
├── .list_phone_contact_rules() / .create_phone_contact_rule(...) / ...  (requires phone number)
├── .get_signing_key_status() / .create_signing_key()
├── mail methods             (requires assigned mailbox)
├── phone methods            (requires assigned phone number)
└── text methods             (requires assigned phone number)

An identity must have a channel assigned before you can use mail/phone methods. If not assigned, an InkboxError is raised with a clear message.

Agent Signup

For the full agent self-signup flow (register, verify, check status, restrictions, and direct API examples), read the shared reference:

See: skills/inkbox-agent-self-signup/SKILL.md

Python SDK methods: Inkbox.signup(...), Inkbox.verifysignup(apikey, ...), Inkbox.resendsignupverification(apikey), Inkbox.getsignupstatus(apikey).

Identities

identity = inkbox.create_identity("sales-agent")
identity = inkbox.get_identity("sales-agent")
identities = inkbox.list_identities()  # → list[AgentIdentitySummary]

identity.update(new_handle="new-name")   # rename
identity.refresh()                       # re-fetch from API, updates cached channels
identity.delete()                        # cascades: mailbox + tunnel + phone-number release

Channel Management

# Identity is created with a mailbox AND tunnel atomically — both come back on the response
print(identity.email_address)            # e.g. "[email protected]"
print(identity.tunnel.public_host)       # e.g. "sales-agent.inkboxwire.com"

# Phone numbers are still opt-in
phone = identity.provision_phone_number(type="local", state="NY")  # local only; toll_free is rejected (422)
print(phone.number)                      # e.g. "+12125551234"

# Release the phone number (vendor + local)
identity.release_phone_number()

Mailboxes and tunnels are not separately linkable — they are 1:1 with their owning identity. Use inkbox.create_identity() to provision both; use identity.delete() to remove both (cascade).

Mail

Import historical mail

from inkbox import MailImportFormat

created = inkbox.mailboxes.imports.create(
    email,
    source_format=MailImportFormat.AUTO,
    original_addresses=["[email protected]"],
)
inkbox.mailboxes.imports.upload(created.upload, "archive.mbox")
inkbox.mailboxes.imports.start(email, str(created.job.id))
job = inkbox.mailboxes.imports.wait(email, str(created.job.id), poll_interval=5)

Formats: auto, mbox, eml, zip. A ZIP may hold .eml and/or .mbox files (a Gmail Takeout ZIP imports as-is); other entries, including nested archives, are ignored. wait returns all terminal states; failure/cancellation are job results, not transport errors. A timeout does not cancel. Counters are cumulative and never go backwards, so a stalled counter is a signal, not normal churn; counters may still remain unchanged while a slow message is processed, and they must not be treated as a percentage. Jobs run one at a time per organization and share overall import capacity, so a long queued stretch is normal; do not cancel and recreate. Unsafe imported content may be rejected.

Upload targets expire after 5 minutes: refreshuploadtarget(email, jobid) and upload again, or cancel the job so it does not hold the mailbox for 24 hours. Limits: 1 GiB per upload, 50 MiB per message, 100,000 messages and 20 originaladdresses per job, 65,000 entries per ZIP, 20 jobs per organization per 24 hours (MailImportQuotaExceededError.retryafterseconds), and one in-flight import per mailbox.

Send

sent = identity.send_email(
    to=["[email protected]"],
    subject="Hello",
    body_text="Hi there!",          # plain text (optional)
    body_html="<p>Hi there!</p>",   # HTML (optional)
    cc=["[email protected]"],          # optional
    bcc=["[email protected]"],        # optional
    in_reply_to_message_id=sent.id, # for threaded replies
    attachments=[{                  # optional
        "filename": "report.pdf",
        "content_type": "application/pdf",
        "content_base64": "<base64>",
    }, {
        "filename": "chart.png",        # inline image: set content_id and
        "content_type": "image/png",    # reference it from body_html as
        "content_base64": "<base64>",   # <img src="cid:chart">. needs body_html
        "content_id": "chart",          # + image/*, unique per send; not on forwards.
    }],
    track_opens=True,               # optional; embed a tracking pixel
)
# track_opens tracks sends only when an HTML body is present. Opens
# surface on the returned Message as sent.first_opened_at / sent.open_count
# (approximate — proxy prefetch inflates it, the per-window debounce
# collapses repeats, so it can read above or below the true count; prefer
# first_opened_at. pixels can also raise spam scores).
#
# send_email / reply_all_email / forward_email all raise
# StorageLimitExceededError (402) when the mailbox is at its storage cap —
# see "Storage cap (402)" below.

Drafts

from inkbox import DraftRecipients

draft = identity.create_email_draft(
    subject="Work in progress",
    idempotency_key="draft-create-2026-08-19-1",
)
for saved in identity.iter_email_drafts():
    print(saved.id, saved.generation)
current = identity.get_email_draft(draft.id)
current = identity.update_email_draft(
    current.id,
    generation=current.generation,
    recipients=DraftRecipients(to=["[email protected]"]),
    subject=None,  # explicit null clears; omission leaves unchanged
)

current = inkbox.drafts.add_attachments(
    identity.email_address,
    current.id,
    generation=current.generation,
    attachments=[{
        "filename": "notes.txt",
        "content_type": "text/plain",
        "content_base64": "bm90ZXM=",
    }],
)
part = current.attachment_metadata[0]
content = inkbox.drafts.download_attachment(
    identity.email_address, current.id, part.part_index, generation=current.generation
)
current = inkbox.drafts.remove_attachment(
    identity.email_address, current.id, part.part_index, generation=current.generation
)

copy = identity.duplicate_email_draft(current.id, generation=current.generation)
identity.delete_email_draft(copy.id, generation=copy.generation)
sent = identity.send_email_draft(current.id, generation=current.generation)

Drafts share the mailbox's standard Drafts folder with connected mail clients. Reuse one idempotencykey and the exact same request when retrying a logical create after an ambiguous result. Use a new key after the original draft is sent or deleted. Forward-only options require forwardmessageid. Use the latest returned generation for every mutation. A partindex belongs to the generation that returned it, so refresh attachment metadata after edits.

Successful send returns a Message and removes the draft; an exact-generation retry may return the same sent message. HTTP 409 errors remain structured on InkboxAPIError.detail["error"]: refresh on draftgenerationconflict and retry the same ID and generation on draftsendinprogress. Never resend draftdelivery_uncertain; after checking sent mail, duplicate or delete it instead.

Read

# Iterate all messages — pagination handled automatically (Iterator[Message])
for msg in identity.iter_emails():
    print(msg.subject, msg.from_address, msg.is_read)

# Filter by direction
for msg in identity.iter_emails(direction="inbound"):   # or "outbound"
    ...

# Unread only (client-side filtered)
for msg in identity.iter_unread_emails():
    ...

# Mark as read
ids = [msg.id for msg in identity.iter_unread_emails()]
identity.mark_emails_read(ids)
identity.mark_emails_unread(ids)   # batch counterpart
# Note: fetching a single inbound message by id (inkbox.messages.get) with
# an API key marks it read server-side; iterating does not, so
# mark_emails_read is the way to clear unread for list-only workflows.
# is_read (agent consumed via API) is distinct from first_opened_at
# (recipient's mail client loaded the tracking pixel).

# Get full thread (oldest-first)
thread = identity.get_thread(msg.thread_id)
for m in thread.messages:
    print(f"[{m.from_address}] {m.subject}")

Thread Folders

Threads carry a folder field: inbox, spam, archive, or blocked (server-assigned, never client-set).

from inkbox import ThreadFolder
# Thread.folder / ThreadDetail.folder is always one of the four values above.

Low-level folder listing / per-thread updates (list(folder=…), list_folders(email), update(..., folder=…)) live on ThreadsResource. Passing folder="blocked" to update raises ValueError before the HTTP call.

Storage cap (402)

Every mailbox has a plan storage cap. All three send paths — sendemail, replyallemail, and forwardemail (and the inkbox.messages.* equivalents) — raise StorageLimitExceededError (HTTP 402) when the send would push the mailbox over it.

from inkbox import StorageLimitExceededError

try:
    identity.send_email(to=["[email protected]"], subject="Hi", body_text="…")
except StorageLimitExceededError as e:
    print(e.message)      # human sentence, includes the limit
    print(e.limit_bytes)  # e.g. 2147483648 (2 GiB)
    print(e.upgrade_url)  # console billing page
    # Free space — reclaim is immediate — or upgrade the plan:
    inkbox.messages.delete(identity.email_address, "<message-uuid>")
    inkbox.threads.delete(identity.email_address, "<thread-uuid>")

Read usage off the mailbox (inkbox.mailboxes.get(...)): storageusedbytes and storagelimitbytes (None = the server resolved no cap). The caps are binary — 2 GiB is 2 * 1024**3 = 2,147,483,648 bytes, so divide by 1024 and label GiB/MiB, never GB.

Free plan: a footer is appended to the stored body of outgoing mail, so inkbox.messages.get(...) does not return byte-for-byte what you sent (a body-less send comes back with the footer as its body). Don't assert sentbody == fetchedbody on a Free plan.

Mail Clients (IMAP/SMTP)

An inbox can be attached to a regular mail client (Thunderbird, Apple Mail, mutt, …) with the API key you already have — there is no separate credential to create and no SDK call involved; the gateway speaks IMAP and SMTP, not HTTP.

The password is the same agent-scoped key an identity-scoped Inkbox(...) client authenticates with; mint one with inkbox.apikeys.create(scopedidentity_id=...). Admin-scoped keys are rejected — one key maps to exactly one mailbox. Revoking the key revokes mail-client access.

Constraints that bite:

• From must be the authenticated inbox address, and exactly one address — aliases / "send as" are rejected.
• On the Free plan, signed/encrypted mail (S/MIME, PGP) cannot be sent over SMTP — the required footer can't be injected without breaking the signature, so the send is refused. Send unsigned, or upgrade.
• Leave "save a copy of sent messages" on — Inkbox recognizes the client's copy as the message it already stored, so you get one Sent entry, charged against the storage cap once.

Full walkthrough: https://inkbox.ai/docs/capabilities/email/mail-clients

Phone

# Place outbound call — stream audio via WebSocket
call = identity.place_call(
    to_number="+15551234567",
    client_websocket_url="wss://your-agent.example.com/ws",
)
print(call.status)
print(call.rate_limit.calls_remaining)

# Or let Inkbox Voice AI drive the call — no WebSocket,
# no code. reason is the agent's task brief (required with
# mode="hosted_agent", invalid otherwise; server 422).
call = identity.place_call(
    to_number="+15551234567",
    mode="hosted_agent",     # CallMode.HOSTED_AGENT; default "client_websocket"
    reason="Confirm tomorrow's 3pm appointment; reschedule if needed.",
)
print(call.mode, call.reason)
# where Voice AI isn't available (or is at capacity), the server's
# 503 (hosted_agent_unavailable / hosted_agent_at_capacity) surfaces verbatim.

# List calls (offset pagination). Every call carries mode / reason plus
# post_call_action_items — open items Voice AI recorded
# (seq-ascending; empty for client_websocket calls)
calls = identity.list_calls(limit=10, offset=0)
for c in calls:
    print(c.id, c.direction, c.remote_phone_number, c.status, c.mode)
    for item in c.post_call_action_items:
        print(f"  [{item.seq}] {item.action}: {item.details}")

# Transcript segments (ordered by seq)
for t in identity.list_transcripts(calls[0].id):
    print(f"[{t.party}] {t.text}")   # party: "local" or "remote"

# Hang up a live call from outside it (teardown confirms asynchronously,
# so the returned call can still show its live status; already-ended
# calls surface the server's 409)
call = identity.hangup_call(calls[0].id)

# Organization-scoped voice discovery; no identity ID is needed.
# Entries include id, name, description, available, and optional preview_url.
# Keep unavailable entries for display; do not hardcode a voice allowlist.
catalog = inkbox.hosted_agent.list_voices()
print(catalog.default_voice, catalog.voices)
selected_voice = next((voice for voice in catalog.voices if voice.available), None)

# Per-identity Inkbox Voice AI config: voice and instructions.
# Both are nullable (None means the server default). set is a FULL REPLACE —
# an omitted field resets to the server default.
cfg = identity.get_hosted_agent_config()
if selected_voice is not None:
    cfg = identity.set_hosted_agent_config(
        voice=selected_voice.id,
        instructions=cfg.instructions,  # Preserve when changing only voice.
    )

# Inbound-call handling: auto_accept | auto_reject | webhook | hosted_agent | forward.
# hosted_agent needs no URL; forward needs exactly one phone or SIP target.
identity.set_incoming_call_action(incoming_call_action="hosted_agent")
identity.set_incoming_call_action(
    incoming_call_action="forward",
    forwarding_target_type="phone",
    forwarding_phone_number="+15551234567",
)
print(identity.get_incoming_call_action().incoming_call_action)

Text Messages (SMS/MMS)

Outbound SMS limits and gates (current):

• Allowed only from local numbers, not toll-free.
• 100 recipient sends per phone number per rolling 24h. A 3-recipient group message counts as 3 recipient sends. A single accepted send may push usage past the cap; the next capped send returns 429 senderratelimited.
• New local numbers need ~10-15 min for 10DLC carrier propagation. identity.phonenumber.smsstatus is SmsStatus.PENDING until ready; sends in this window return 409 sendersmspending.
• Recipient must have texted START to any number in the org. Unknown → 403 recipientnotoptedin. STOP → 403 recipientoptedout. Inspect / override consent state via inkbox.smsopt_ins (see below).
• Beta: Group MMS and conversation sends are beta. Some carriers may reject group chats or MMS from 10DLC numbers even when the sender is ready and recipients have opted in.

Customer-managed 10DLC brands/campaigns lift the default per-number cap to the carrier-assigned tier. Toll-free SMS sending is still coming soon.

# Send SMS/MMS from this identity's phone number.
# Returns a queued TextMessage; final delivery state arrives via any
# webhook subscription on the sender's phone number whose event_types
# include the text.* lifecycle events.
sent = identity.send_text(to="+15551234567", text="Hello from Inkbox")
print(sent.id, sent.delivery_status)   # SmsDeliveryStatus.QUEUED

# Group MMS beta: pass a list of recipients plus optional media URLs.
group = identity.send_text(
    to=["+15551234567", "+15557654321"],
    text="Hello group",
    media_urls=["https://example.com/photo.jpg"],
)
print(group.conversation_id, group.recipients)

# Reply to an existing conversation by UUID. Do not pass "to" with this form.
reply = identity.send_text(
    conversation_id=group.conversation_id,
    text="Following up in the same conversation.",
)

# List text messages (offset pagination)
texts = identity.list_texts(limit=20, offset=0)
for t in texts:
    print(t.id, t.direction, t.remote_phone_number, t.text, t.is_read)

# Filter by read state
unread = identity.list_texts(is_read=False)

# Get a single text message
text = identity.get_text("text-uuid")
print(text.type)   # "sms" or "mms"
if text.media:     # MMS media attachments (temporary signed URLs)
    for m in text.media:
        print(m.content_type, m.size, m.url)

# List one-to-one conversation summaries; opt into groups explicitly.
convos = identity.list_text_conversations(limit=20, include_groups=True)
for c in convos:
    print(c.id, c.participants, c.latest_has_media, c.latest_text)

# Get messages in a specific conversation by remote number or conversation UUID.
msgs = identity.get_text_conversation("+15551234567", limit=50)

# Mark a text as read (identity convenience method)
identity.mark_text_read("text-uuid")

# Mark all messages in a conversation as read
result = identity.mark_text_conversation_read("+15551234567")
print(result["updated_count"])

# Admin-only: search, update, delete
results = inkbox.texts.search(phone.id, q="invoice", limit=20)
inkbox.texts.update(phone.id, "text-uuid", status="deleted")

iMessage

iMessage can use the shared service or an organization-owned dedicated line. On shared service, recipients ask the triage number to connect them to @agent_handle; the shared local number is never exposed. Shared service requires the recipient to message first. A dedicated line may start a conversation, subject to consent, contact-rule, and rate-limit checks.

Discover the router (triage) number at runtime — it can change, so never hardcode it:

triage = inkbox.imessages.get_triage_number()
print(triage.number, triage.connect_command)  # "+1646...", "connect @your-handle"
# Humans connect by texting that command to that number.

Reachability is opt-in per identity (imessage_enabled, default False):

identity = inkbox.create_identity("my-agent", imessage_enabled=True)
# or toggle later
identity.update(imessage_enabled=True)
# admin-only: flip contact-rule mode (default "blacklist")
identity.update(imessage_filter_mode="whitelist")
print(identity.imessage_enabled, identity.imessage_filter_mode)

Dedicated lines follow the phone-number resource style: list or claim them on the org-level iMessage resource, then inspect the typed number model. Claims require admin credentials.

numbers = inkbox.imessages.list_numbers()  # attached and unattached
number = inkbox.imessages.claim_number(
    idempotency_key="claim-agent-2026-07-18",
)
print(number.number, number.status, number.agent_identity_id)

Claim and attach atomically during identity create/update. Do not make a separate attach call after an atomic claim. imessagenumberid=None is intentional wire data that moves an identity back to shared service; omitting the argument leaves its attachment unchanged.

New identities default contactsharingenabled=True. When a dedicated line is attached, it automatically offers the identity's display name (or handle as fallback) and optional avatar. Pass contactsharingenabled=False during creation to opt out before the line is claimed, or update the identity later to enable or disable sharing.

dedicated_identity = inkbox.create_identity(
    "dedicated-agent",
    imessage_enabled=True,
    contact_sharing_enabled=False,  # opt out before claiming the line
    claim_imessage_number=True,
)
print(dedicated_identity.imessage_number.number)

dedicated_identity.update(contact_sharing_enabled=True)  # enable later

identity.update(
    claim_imessage_number=True,
    idempotency_key="swap-my-agent-2026-07-18",
)                                                         # claim + swap
identity.update(imessage_number_id=number.id)             # attach owned number
identity.update(imessage_number_id=None)                  # return to shared

claimnumber and atomic identity claims may raise DedicatedIMessageNumberQuotaExceededError, DedicatedIMessageNumberInventoryPendingError, or IdempotencyKeyReusedError. The inventory error exposes retryafter_seconds; do not retry sooner. Reuse the same caller-generated idempotency key when retrying an ambiguous claim.

Messaging (identity convenience methods; inkbox.imessages is the org-level resource with the same operations plus agentidentityid / is_blocked filters):

from inkbox import IMessageSendStyle

# Send to a connected recipient, or reply into a conversation by UUID.
sent = identity.send_imessage(to="+15551234567", text="Hello over iMessage")
group = dedicated_identity.send_imessage(
    to=["+15551234567", "+15557654321"],
    text="Hello group",
    media_urls=["https://example.com/group-photo.jpg"],
    send_style=IMessageSendStyle.CONFETTI,
)  # dedicated line only; 2–8 distinct recipients
group_reply = dedicated_identity.send_imessage(
    conversation_id=group.conversation_id,
    text="Group follow-up",
    media_urls=["https://example.com/follow-up.jpg"],
    send_style=IMessageSendStyle.LASERS,
)
print(sent.service, sent.status)  # IMessageService.IMESSAGE, IMessageDeliveryStatus.QUEUED

# List messages / conversations
msgs = identity.list_imessages(limit=20, is_read=False, include_groups=True)
convos = identity.list_imessage_conversations(limit=20, include_groups=True)
convo = identity.get_imessage_conversation(sent.conversation_id)
# assignment_status tells you whether the recipient is still connected:
# anything other than "active" means sends/reactions will be refused
# until they reconnect through triage.
print(convo.assignment_status)
# Group rows have nullable assignment/remote fields and a best-known participant
# snapshot. group_creation_status is creating, not_created, or ready. A rejected
# initial creation keeps the same conversation; send again by conversation_id to
# retry, and success changes it to ready.
# Group creation and conversation_id replies accept the same 13
# IMessageSendStyle values as one-to-one sends, with or without the media URL.

# Who is actively connected to this identity right now (paginated)?
connections = identity.list_imessage_assignments(limit=20)
for a in connections:
    print(a.remote_number, a.status, a.created_at)

# Tapbacks target inbound one-to-one or group messages by message_id. Sends
# accept seven named reactions (love, like, dislike, laugh, emphasize,
# question, eyes); inbound can also be "custom" with the literal emoji in
# custom_emoji. Arbitrary custom emoji are not sendable.
sent_reaction = identity.send_imessage_reaction(message_id=msgs[0].id, reaction="like")

# Live tapbacks come back on message reads, oldest first.
for r in msgs[0].reactions or []:
    print(r.direction, r.reaction, r.custom_emoji)

# Take your own tapback back. Only the sender can. A failed removal leaves the
# tapback in place rather than clearing it locally, so the call can be retried.
identity.remove_imessage_reaction(sent_reaction.id)

# Read receipts + typing indicator are one-to-one only; groups return 409.
identity.mark_imessage_conversation_read(sent.conversation_id)
identity.send_imessage_typing(sent.conversation_id)

# Media: upload bytes (max 10 MiB), then send the returned URL (one per message)
upload = identity.upload_imessage_media(
    content=open("photo.jpg", "rb").read(),
    filename="photo.jpg",
    content_type="image/jpeg",
)
identity.send_imessage(to="+15551234567", media_urls=[upload.media_url])

Contact rules are scoped to the identity, including when it has a dedicated line:

from inkbox import IMessageRuleAction

rule = inkbox.imessage_contact_rules.create(
    "my-agent", action=IMessageRuleAction.BLOCK, match_target="+15559999999",
)
rules = inkbox.imessage_contact_rules.list("my-agent")
inkbox.imessage_contact_rules.update("my-agent", rule.id, action="allow")  # admin-only
inkbox.imessage_contact_rules.delete("my-agent", rule.id)                   # admin-only
all_rules = inkbox.imessage_contact_rules.list_all()                        # admin-only, org-wide

Inbound messages and reactions arrive via identity-owned webhook subscriptions — see Webhooks below.

SMS Opt-Ins

Per-recipient SMS consent state, keyed by (your org, recipient number). The registry is updated automatically when recipients text START / STOP to any of your numbers (source="sms"). Reads are admin-only; writes are admin-only and require your org to be on its own active, customer-managed 10DLC campaign (Inkbox-default-campaign orgs share consent state and get 409 customercampaignrequired on writes — source="api" writes record an audit event).

from inkbox import SmsOptInStatus

# List your org's consent rows, newest-updated first (server caps limit at 200)
rows = inkbox.sms_opt_ins.list(limit=50)
opted_out = inkbox.sms_opt_ins.list(status=SmsOptInStatus.OPTED_OUT)

# Look up one recipient — 404 → InkboxAPIError if no row exists
row = inkbox.sms_opt_ins.get("+15551234567")
print(row.status, row.source, row.opted_in_at, row.opted_out_at)

# Programmatic writes (customer-managed 10DLC campaign only)
inkbox.sms_opt_ins.opt_in("+15551234567")
inkbox.sms_opt_ins.opt_out("+15551234567")

Agent-to-Agent (A2A)

Invitations: an admin-scoped API key uses inkbox.a2ainvitations.create(peeragenthandles, recipientemail=..., expiresinseconds=...), .list(...), .get(id), and .revoke(id). A claimed agent-scoped key uses .accept(invitation). The value may be an exact-origin share URL or raw token; extracta2ainvitationtoken() performs the same strict local normalization. Unbound create responses may reveal invitationtoken, invitationurl, and agenthandoff_prompt; email-bound creates omit capability fields. Signup accepts the same input and returns the optional invitation summary. Do not retry create or accept automatically.

An identity can inspect work it received, work it requested, or both. Omit direction on a2atasks for the receiver inbox; a2asent_tasks is the outbound-only alias.

page = inkbox.a2a.public_directory(q="research", limit=25)
org_page = inkbox.a2a.organization_directory(q="support")
for item in page.items:
    print(item.card.name, item.card_url, item.visibility)

identity.a2a_set_publicly_discoverable(True)  # admin API key required
identity.a2a_set_allow_public_egress(True)

page = identity.a2a_tasks(
    direction="both",
    requester_handle="coordinator",
    worker_handle="researcher",
    state="working",
    context_id="context-uuid",
    q="quarterly report",
    since="2026-07-01T00:00:00Z",
    limit=25,
)

# Explicit pages expose an opaque next_cursor.
if page.next_cursor:
    next_page = identity.a2a_tasks(
        direction="both",
        requester_handle="coordinator",
        worker_handle="researcher",
        state="working",
        context_id="context-uuid",
        q="quarterly report",
        since="2026-07-01T00:00:00Z",
        cursor=page.next_cursor,
        limit=25,
    )

# Iterators preserve filters while draining every cursor page.
for message in identity.iter_a2a_messages(
    direction="outbound",
    worker_handle="researcher",
    role="agent",
    q="revenue",
):
    print(message.task_id, message.context_id, message.task_state, message.parts)

for context in identity.a2a_contexts(direction="both").items:
    print(context.name, context.id)

identity.a2a_update_context(
    "context-uuid",
    name="Quarterly Research Review",
)

Task filters: direction, requesterhandle, workerhandle, state, contextid, q, since, cursor, limit. Message filters additionally support taskid and role; role is the message author (caller or agent), independent of task direction. Message direction defaults to both. Multiple filters are ANDed. Task search returns tasks containing a matching message; message search returns individual matches with requester/worker and task/context provenance. Search covers string and numeric content values from text and data parts, excludes metadata, and is deterministic newest-first rather than relevance-ranked.

Use a2atask / a2asent_task for a task's current state and message history.

New contexts start with the persisted name New A2A Session. That exact default may be replaced with a name based on the first task message. Either participant can rename a context at any time; automatic naming does not replace a non-default name. Context-level caller and target remain the original opener and recipient. Each nested task carries its own authoritative participants, and tasks in both directions can run concurrently.

The standard client starts a sibling task when contextid is supplied without taskid. Supplying task_id continues that specific task. This cross-endpoint reuse is supported between Inkbox identities; external A2A services may define different behavior.

For a multi-turn worker flow, reply with intent="ask_caller" to request input; the caller continues the same task through the standard A2A client, and the worker later replies with intent="complete" or intent="fail".

Directory methods accept q, cursor, and limit; iterator variants follow all pages. Receiver enablement, public egress, and advertised skills may be changed with the identity's agent-scoped key. Public discoverability and other admission-policy mutations require an admin API key: a2asetpubliclydiscoverable, a2asetfiltermode, a2aaddcontactrule, a2aupdatecontactrule, and a2adeletecontactrule. Use a2areset_skills() to restore the default Agent Card skills. Contact-rule directions are inbound, outbound, or both. Same-organization and public discovery may imply admission. Private cross-organization calls require requester-outbound and worker-inbound permission; explicit blocks always win.

Vault

Encrypted credential vault with client-side Argon2id key derivation and AES-256-GCM encryption. The server never sees plaintext secrets. Requires argon2-cffi and cryptography (included as dependencies).

Initialize

# Initialize a new vault (org ID is fetched automatically from the API key)
result = inkbox.vault.initialize("my-Vault-key-01!")
print(result.vault_id, result.vault_key_id)
for code in result.recovery_codes:
    print(code)  # save these immediately — they cannot be retrieved again

Unlock & Read

from inkbox import LoginPayload, APIKeyPayload, SSHKeyPayload, OtherPayload

# Unlock with a vault key — derives key via Argon2id, decrypts all secrets
unlocked = inkbox.vault.unlock("my-Vault-key-01!")

# Optionally filter to secrets an agent identity has access to
unlocked = inkbox.vault.unlock("my-Vault-key-01!", identity_id="agent-uuid")

# All decrypted secrets from the unlock bundle
for secret in unlocked.secrets:
    print(secret.name, secret.secret_type)
    print(secret.payload)   # LoginPayload, APIKeyPayload, SSHKeyPayload, or OtherPayload

# Fetch and decrypt a single secret by ID
secret = unlocked.get_secret("secret-uuid")
print(secret.payload.username, secret.payload.password)   # for login type

Create & Update

# Create a login secret (secret_type inferred from payload type)
unlocked.create_secret(
    "AWS Production",
    LoginPayload(password="s3cret", username="admin", url="https://aws.amazon.com"),
    description="Production IAM user",
)

# Create an API key secret
unlocked.create_secret(
    "GitHub PAT",
    APIKeyPayload(api_key="ghp_xxx"),
)

# Create an SSH key secret
unlocked.create_secret(
    "Deploy Key",
    SSHKeyPayload(private_key="-----BEGIN OPENSSH PRIVATE KEY-----..."),
)

# Create a freeform secret
unlocked.create_secret("Misc", OtherPayload(data="any freeform content"))

# Update name/description and/or re-encrypt payload
unlocked.update_secret("secret-uuid", name="New Name")
unlocked.update_secret("secret-uuid", payload=LoginPayload(password="new", username="new"))

# Delete
unlocked.delete_secret("secret-uuid")

Metadata (no unlock needed)

info = inkbox.vault.info()                                   # VaultInfo
keys = inkbox.vault.list_keys()                              # list[VaultKey]
keys = inkbox.vault.list_keys(key_type="recovery")           # filter by type
secrets = inkbox.vault.list_secrets()                         # list[VaultSecret] (metadata only)
secrets = inkbox.vault.list_secrets(secret_type="login")     # filter by type
inkbox.vault.delete_secret("secret-uuid")                    # delete without unlocking

Payload Types

secret_type is immutable after creation. To change it, delete and recreate.

Agent Credentials (identity-scoped)

Agent-facing credential access — typed, identity-scoped. The vault stays as the admin surface; identity.credentials is the agent runtime surface.

from inkbox import Credentials

# Unlock the vault first (stores state on the client)
inkbox.vault.unlock("my-Vault-key-01!")

identity = inkbox.get_identity("support-bot")

# Discovery — returns list[DecryptedVaultSecret] with name/metadata
all_creds = identity.credentials.list()
logins    = identity.credentials.list_logins()
api_keys  = identity.credentials.list_api_keys()
ssh_keys  = identity.credentials.list_ssh_keys()
key_pairs = identity.credentials.list_key_pairs()

# Access by UUID — returns typed payload directly
login    = identity.credentials.get_login("secret-uuid")      # → LoginPayload
api_key  = identity.credentials.get_api_key("secret-uuid")    # → APIKeyPayload
ssh_key  = identity.credentials.get_ssh_key("secret-uuid")    # → SSHKeyPayload
key_pair = identity.credentials.get_key_pair("secret-uuid")   # → KeyPairPayload

# Generic access — returns DecryptedVaultSecret
secret = identity.credentials.get("secret-uuid")

• Requires inkbox.vault.unlock() first — raises InkboxError if vault is not unlocked
• Results are filtered to secrets the identity has access to (via access rules)
• Cached after first access; call identity.refresh() to clear the cache
• get_* raises KeyError if not found, TypeError if wrong secret type

One-Time Passwords (TOTP)

TOTP secrets are stored inside LoginPayload.totp in the encrypted vault. Codes are generated client-side — no server call needed.

From an agent identity (recommended)

from inkbox.vault.totp import parse_totp_uri
from inkbox.vault.types import LoginPayload

# Create a login with TOTP
secret = identity.create_secret(
    name="GitHub",
    payload=LoginPayload(
        username="[email protected]",
        password="s3cret",
        totp=parse_totp_uri("otpauth://totp/GitHub:[email protected]?secret=JBSWY3DPEHPK3PXP&issuer=GitHub"),
    ),
)

# Generate TOTP code
code = identity.get_totp_code(str(secret.id))
print(code.code)              # e.g. "482901"
print(code.seconds_remaining) # e.g. 17

# Add/replace TOTP on existing login
identity.set_totp(secret_id, "otpauth://totp/...?secret=...")

# Remove TOTP
identity.remove_totp(secret_id)

From the unlocked vault (admin-only)

unlocked = inkbox.vault.unlock("my-Vault-key-01!")

# Same methods available on UnlockedVault
unlocked.set_totp(secret_id, totp_config_or_uri)
unlocked.remove_totp(secret_id)
code = unlocked.get_totp_code(secret_id)

TOTPCode fields

Admin-only Resources

Mailboxes (inkbox.mailboxes)

mailboxes = inkbox.mailboxes.list()
mailbox   = inkbox.mailboxes.get("[email protected]")

# To rename, use `identity.update(display_name="New Name")` — the
# mailbox PATCH endpoint hard-rejects `display_name` with a 422. To
# attach a webhook receiver, see "Webhooks" below.

# DEPRECATED channel path — the mail filter mode now lives on the identity.
# Prefer `identity.update(mail_filter_mode="whitelist")` (which does NOT return
# a change notice). This legacy mailbox flip still works and still returns one:
updated = inkbox.mailboxes.update(mailbox.email_address, filter_mode="whitelist")
if updated.filter_mode_change_notice:
    # Populated when filter_mode actually changed — tells you how many
    # rules are now redundant under the new mode.
    n = updated.filter_mode_change_notice
    print(n.redundant_rule_count, n.redundant_rule_action, n.new_filter_mode)

# Mailbox responses now also carry mailbox.agent_identity_id when the
# mailbox is linked to an identity.
# `mailbox.sending_domain` is the bare domain the mailbox sends from
# (platform default or a verified custom domain — see "Custom email domains" below).

# Storage (list / get / update all carry these):
print(mailbox.storage_used_bytes)               # bytes stored, e.g. 1288490188
print(mailbox.storage_limit_bytes)              # plan cap, e.g. 2147483648 (2 GiB), or None
used_gib = mailbox.storage_used_bytes / 1024**3  # caps are BINARY — GiB, not GB
# Over-cap sends raise StorageLimitExceededError (402) — see "Storage cap (402)".

results = inkbox.mailboxes.search(mailbox.email_address, q="invoice", limit=20)
# Mailboxes are deleted via the owning identity's cascade — there is no standalone delete:
#   identity.delete()  # removes the mailbox + tunnel atomically (cascade)

Custom email domains (inkbox.domains)

If your org has registered custom sending domains in the console, list them and (admin-only) set the org default. New mailboxes inherit the org default unless you pass `sendingdomainid (standalone) or sending_domain` (identity).

from inkbox import SendingDomainStatus

verified = inkbox.domains.list(status=SendingDomainStatus.VERIFIED)

# Admin-scoped API key only — non-admin keys get 403.
# Returns the bare new default domain name (or None when reverted to platform).
new_default = inkbox.domains.set_default("mail.acme.com")
# Pass the platform domain (e.g. "inkboxmail.com" in prod) to clear the org default.

# Identity create: pick by bare domain name (not id).
inkbox.create_identity("sales-bot", sending_domain="mail.acme.com")
# Force the platform default:
inkbox.create_identity("sales-bot-2", sending_domain=None)
# Standalone mailbox creation is gone — provision via create_identity above.

Phone Numbers (inkbox.phone_numbers)

numbers = inkbox.phone_numbers.list()
number  = inkbox.phone_numbers.get("phone-number-uuid")
number  = inkbox.phone_numbers.provision(agent_handle="my-agent", type="local", state="NY")  # local only; toll_free is rejected (422)

inkbox.phone_numbers.update(
    number.id,
    incoming_call_action="webhook",            # also auto_accept, auto_reject, hosted_agent, or forward
    incoming_call_webhook_url="https://...",
)
inkbox.phone_numbers.update(
    number.id,
    incoming_call_action="auto_accept",
    client_websocket_url="wss://...",
)
inkbox.phone_numbers.update(
    number.id,
    incoming_call_action="hosted_agent",       # no URL — Voice AI answers
)
inkbox.phone_numbers.update(
    number.id,
    incoming_call_action="forward",
    forwarding_target_type="sip",
    forwarding_sip_uri="sip:[email protected]",
)

hits = inkbox.phone_numbers.search_transcripts(number.id, q="refund", party="remote", limit=50)
inkbox.phone_numbers.release(number.id)

Phone numbers carry the same filtermode / agentidentityid / filtermodechangenotice fields as mailboxes; flipping filtermode here is the deprecated channel path (admin-only; returns a change-notice when the value actually changed). Prefer identity.update(phonefilter_mode="whitelist"), which sets the mode on the identity and does not return a change notice.

Contact Rules

Allow/block lists are scoped to the agent identity (mirroring iMessage), addressed by agenthandle. The identity's mailfiltermode / phonefiltermode decides whether each channel's rules act as a whitelist or blacklist. Mail matches by exact email or domain; phone matches by exact E.164 number. Returned rows are MailIdentityContactRule / PhoneIdentityContactRule, keyed by rule.agentidentity_id (not a mailbox/phone-number id).

from inkbox import (
    MailRuleAction, MailRuleMatchType, PhoneRuleAction, PhoneRuleMatchType,
    DuplicateContactRuleError,
)

identity = inkbox.get_identity("sales-agent")

# Mail rules via the identity convenience methods. New rules always start
# active.
rule = identity.create_mail_contact_rule(
    action=MailRuleAction.ALLOW,         # or BLOCK
    match_type=MailRuleMatchType.DOMAIN, # or EXACT_EMAIL
    match_target="example.com",
)
identity.list_mail_contact_rules()
identity.get_mail_contact_rule(rule.id)
identity.update_mail_contact_rule(rule.id, action="allow")  # admin-only
identity.delete_mail_contact_rule(rule.id)                   # admin-only

# Phone rules — same shape, only match_type="exact_number" is supported.
# Phone helpers require the identity to have a phone number (else InkboxError).
identity.create_phone_contact_rule(
    action=PhoneRuleAction.BLOCK,
    match_target="+15551234567",
    match_type=PhoneRuleMatchType.EXACT_NUMBER,
)
identity.list_phone_contact_rules()

# Equivalent org-level resources, keyed by agent_handle, with an org-wide list_all:
inkbox.mail_identity_contact_rules.create(
    "sales-agent", action="allow", match_type="domain", match_target="example.com",
)
inkbox.mail_identity_contact_rules.list("sales-agent")
inkbox.mail_identity_contact_rules.list_all(agent_identity_id=str(identity.id))  # admin-only, org-wide
inkbox.phone_identity_contact_rules.list_all()                                   # admin-only, org-wide

# Duplicate (match_type, match_target) on the same identity raises 409:
try:
    identity.create_mail_contact_rule(action="allow", match_type="domain", match_target="example.com")
except DuplicateContactRuleError as e:
    print(e.existing_rule_id)   # UUID of the rule that already matched

Filter mode

The whitelist/blacklist mode lives on the identity. Flip it with identity.update (admin-only). Unlike the deprecated channel update, this does not return a FilterModeChangeNotice. phonefiltermode requires the identity to have a phone number (else a 422).

identity.update(mail_filter_mode="whitelist", phone_filter_mode="blacklist")
print(identity.mail_filter_mode, identity.phone_filter_mode)

Deprecated: per-mailbox / per-number rules

The legacy per-mailbox inkbox.mailcontactrules and per-number inkbox.phonecontactrules resources still work but hit deprecated server routes (Sunset 2026-08-31). Prefer the identity-keyed surface above.

# Deprecated — per-mailbox mail rule:
inkbox.mail_contact_rules.create(
    mailbox.email_address,
    action="allow", match_type="domain", match_target="example.com",
)
inkbox.mail_contact_rules.list_all(mailbox_id=str(mailbox.id))
# Deprecated — per-number phone rule:
inkbox.phone_contact_rules.create(
    number.id, action="block", match_type="exact_number", match_target="+15551234567",
)

Contacts

Organization-wide address book with lifecycle review, memory, correspondence, and vCard import/export.

Merging requires an admin-scoped API key. Active memories have per-kind and contact-wide limits. Delete a fact from each kind named by a merge error, or any active fact when it names total, then retry. Untyped memories count toward the total.

from inkbox import (
    Contact, ContactCorrespondenceOptions, ContactEmail, ContactPhone,
    ContactAddress, ContactReviewStatus,
)

# CRUD
contact = inkbox.contacts.create(
    given_name="Ada",
    family_name="Lovelace",
    emails=[ContactEmail(label="work", value="[email protected]")],
    phones=[ContactPhone(label="mobile", value="+15551234567")],
)
inkbox.contacts.get(str(contact.id))
inkbox.contacts.list(
    q="ada", order="recent", review_status=[ContactReviewStatus.CONFIRMED]
)
inkbox.contacts.update(str(contact.id), job_title="Analyst")
inkbox.contacts.delete(str(contact.id))
inkbox.contacts.bulk_delete(["contact-uuid-1", "contact-uuid-2"])

# Reverse-lookup — exactly one filter required (else ValueError before HTTP)
inkbox.contacts.lookup(email="[email protected]")
inkbox.contacts.lookup(email_domain="example.com")
inkbox.contacts.lookup(phone="+15551234567")
inkbox.contacts.lookup(email_contains="ada")
inkbox.contacts.lookup(phone_contains="555")

# Compatibility access information is read-only
inkbox.contacts.access.list(str(contact.id))

# Facts, citations, correspondence, and duplicate merging
# fact.kind is "profile", "preference", or "context"; unlocked extracted
# context facts drop out of list() at fact.expires_at. Locked facts stay active.
facts = inkbox.contacts.facts.list(str(contact.id))
inkbox.contacts.facts.list(str(contact.id), include_expired=True)
if facts and facts[0].citations and facts[0].citations[0].source_url:
    source = inkbox.contacts.facts.resolve_citation_url(facts[0].citations[0].source_url)
# Hand-written facts never expire; create/update/delete are admin only.
# Any update makes a fact user-authored, clears expiry, and revives it. Content
# changes also remove confidence and citations; kind-only changes preserve them.
fact = inkbox.contacts.facts.create(
    str(contact.id), content="Prefers email over calls", kind="preference"
)
inkbox.contacts.facts.update(str(contact.id), str(fact.id), kind="profile")
if facts:
    inkbox.contacts.facts.delete(str(contact.id), str(facts[0].id))  # admin only
history = inkbox.contacts.correspondence.get(
    str(contact.id),
    ContactCorrespondenceOptions(identity_id="identity-uuid", channels=["email", "sms"]),
)
survivor = inkbox.contacts.merge(
    str(contact.id), losing_contact_ids=["duplicate-contact-uuid"]
)

# vCards
result = inkbox.contacts.vcards.import_vcards(vcf_text)  # bulk, ≤5 MiB, ≤1000 cards
print(result.created_ids)     # list[UUID]
for item in result.errors:    # list[ContactImportResultItem]
    print(item.index, item.error)
for item in result.conflicts:
    print(item.index, item.conflicting_contact_id)

vcf = inkbox.contacts.vcards.export_vcard(str(contact.id))  # vCard 4.0 string
batch = inkbox.contacts.vcards.export_vcards(["contact-uuid-1", "contact-uuid-2"])
print(batch.vcard)

Notes

Admin-only free-form notes with per-identity access grants. Identities must be granted access explicitly — there is no wildcard for notes.

note = inkbox.notes.create(body="Customer prefers email follow-up.", title="Ada")
inkbox.notes.get(str(note.id))
inkbox.notes.list(q="email", identity_id="agent-uuid", order="recent", limit=50)
inkbox.notes.update(str(note.id), body="Updated body")
inkbox.notes.update(str(note.id), title=None)   # clear title (body cannot be null)
inkbox.notes.delete(str(note.id))

# Access grants (admin + JWT only)
inkbox.notes.access.list(str(note.id))
inkbox.notes.access.grant(str(note.id), identity_id="agent-uuid")
inkbox.notes.access.revoke(str(note.id), "agent-uuid")

Whoami

# Check the authenticated caller's identity
info = inkbox.whoami()
print(info.auth_type)        # "api_key" or "jwt"
print(info.organization_id)

Returns WhoamiApiKeyResponse (with keyid, label, creatortype, authsubtype, etc.) or WhoamiJwtResponse (with email, orgrole, etc.) based on auth_type.

For branching on API-key scope, compare against the exported constants:

from inkbox import (
    AUTH_SUBTYPE_API_KEY_ADMIN_SCOPED,
    AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_CLAIMED,
    AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_UNCLAIMED,
)

if info.auth_type == "api_key" and info.auth_subtype == AUTH_SUBTYPE_API_KEY_ADMIN_SCOPED:
    ...   # admin-only operations (filter_mode flips, rule updates/deletes, etc.)

Tunnels

Bring a local process online at a public https://{name}.inkboxwire.com URL. Outbound HTTP/2 only — no inbound port to open. POSIX only.

# Forward to a local URL (edge mode — Inkbox terminates TLS at the edge)
listener = inkbox.tunnels.connect(
    name="my-app",
    forward_to="http://127.0.0.1:8080",
)
print(listener.public_url)        # https://my-app.inkboxwire.com
print(listener.status, listener.is_connected, listener.last_connected_at)
listener.wait()                   # blocks until close()/Ctrl-C

# Forward to an in-process ASGI app (FastAPI / Starlette / your own)
listener = inkbox.tunnels.connect(name="my-app", forward_to=fastapi_app)

# Passthrough TLS is fixed at create time (see below); the connect() call is
# identical. In passthrough the SDK auto-signs a cert via the control plane.
listener = inkbox.tunnels.connect(
    name="my-app",
    forward_to="http://127.0.0.1:8080",
)

Async usage:

async with ...:
    listener = inkbox.tunnels.connect(name="my-app", forward_to="http://127.0.0.1:8080")
    try:
        await listener.serve_forever()
    finally:
        await listener.aclose()

wait()/close() and serve_forever()/aclose() are mutually exclusive — pick one pair.

listener.status is idle, connecting, connected, reconnecting, closed, or superseded. listener.isconnected is local runtime liveness; listener.lastconnected_at is an aware UTC timestamp retained while reconnecting. listener.tunnel remains the bootstrap resource snapshot. listener.close() raises TimeoutError if its runtime thread cannot stop within 30 seconds.

Tunnels are provisioned atomically by inkbox.createidentity(...); there is no standalone create / delete / restore / rotatesecret surface. For passthrough, opt in at create time: inkbox.createidentity("my-app", tunnel={"tlsmode": "passthrough"}) — tls_mode is fixed at create.

Reads + edit:

inkbox.tunnels.list()                       # list[Tunnel]
inkbox.tunnels.get("tunnel-uuid")
inkbox.tunnels.update(                      # metadata-only
    "tunnel-uuid",
    metadata={"team": "gtm"},
)
# Passthrough only:
inkbox.tunnels.sign_csr("tunnel-uuid", csr_pem=csr_bytes)

Data-plane auth uses the same apikey the Inkbox client was constructed with — admin-scoped or identity-scoped (matching the tunnel's identity). Mint a per-agent identity-scoped key via inkbox.apikeys.create(scopedidentityid=...). Selected connect() kwargs: poolsize (1–32), statedir (default ~/.inkbox/tunnels/{name}), onstatus callback, allowremoteforwarding=False (loopback-only allowlist), forwardtoverifytls=True. In passthrough mode the state dir holds the per-tunnel private key — treat it like an SSH key dir.

For full options, lifecycle notes, and TS examples, see skills/inkbox-tunnels/SKILL.md.

Webhooks & Signature Verification

Webhooks are configured directly on the mailbox or phone number — no separate registration.

import json
from typing import cast
from inkbox import (
    verify_webhook,
    MailWebhookPayload, TextWebhookPayload, PhoneIncomingCallWebhookPayload,
)

# Each agent identity has its own webhook signing key. Create/rotate it
# (plaintext returned once — save it), or read its status:
key = identity.create_signing_key()                        # → SigningKey
status = identity.get_signing_key_status()                 # → SigningKeyStatus(configured, created_at)
# Org-level resource, keyed by agent_handle:
key = inkbox.signing_keys.create_or_rotate("sales-agent")
status = inkbox.signing_keys.get_status("sales-agent")
# DEPRECATED: org-level inkbox.create_signing_key() — with an agent-scoped key it
# still rotates that identity's key; with an admin key the server returns 409.

# Verify, then parse + discriminate
if not verify_webhook(payload=raw_body, headers=request.headers, secret="whsec_..."):
    raise HTTPException(status_code=403)
payload = cast(TextWebhookPayload, json.loads(raw_body))
if payload["event_type"] == "text.delivery_failed":
    msg = payload["data"]["text_message"]
    logger.error("SMS failed: %s (%s)", msg["error_code"], msg["error_detail"])

Algorithm: HMAC-SHA256 over "{request_id}.{timestamp}.{body}".

Event taxonomy:

• Mail (envelope, fire-and-forget) — message.received, message.sent, message.forwarded, message.delivered, message.bounced, message.failed. Subscribe via inkbox.webhooks.subscriptions.create(mailboxid=..., url=..., eventtypes=[...]). On message.received, data["message"] includes the plain-text body (whole under a size cap, else a prefix with bodytruncated: True / bodystate: "truncated"); when truncated, fetch the full message with inkbox.messages.get(message["emailaddress"], message["id"]) — use id (row id), not messageid (RFC 5322 header). These fields are present-with-null on the other events and absent on pre-feature payloads.
• Text (envelope, fire-and-forget) — text.received, text.sent, text.delivered, text.deliveryfailed, text.deliveryunconfirmed. Subscribe via inkbox.webhooks.subscriptions.create(phonenumberid=..., url=..., eventtypes=[...]). The text-message body carries deliverystatus as an outbound message-level rollup; 1:1 traffic also hoists errorcode, errordetail, sentat, deliveredat, and failed_at. On group outbound those legacy detail fields are None and per-recipient state lives in recipients[].
• iMessage (envelope, fire-and-forget) — imessage.received, imessage.reactionreceived, plus the outbound delivery lifecycle imessage.sent, imessage.delivered, imessage.deliveryfailed (declined/error; details on the message object). Subscribe via inkbox.webhooks.subscriptions.create(agentidentityid=..., url=..., eventtypes=[...]) — owned by the agent identity, since shared iMessage pool numbers are not org resources. data["message"] is populated on imessage.received and the three delivery-lifecycle events; data["reaction"] on imessage.reactionreceived. Fan-out only happens while the identity is active and imessage_enabled; contact-rule-blocked traffic is never delivered.
• Call lifecycle (envelope, fire-and-forget + replayable) — call.ended, owned by the agent identity (like iMessage). Subscribe via inkbox.webhooks.subscriptions.create(agentidentityid=..., url=..., eventtypes=["call.ended"]). CallEndedWebhookPayload.data carries the call (WebhookPhoneCall, with derived durationseconds), resolved contacts / agentidentities, an always-present transcripturl (authoritative verbatim, fetch with an admin API key), and an inline transcript block (WebhookCallTranscript, middle-cut/abridged) present when the platform captured a transcript for the call, otherwise None — discriminate a turn from the abridgment marker on "marker" in entry. Voice AI call fields (all optional so pre-Voice AI payloads parse): data["call"] carries mode / reason; data carries outcome ("completed" | "noanswer" | "declined" | "failed", None iff mode is clientwebsocket) and postcallactionitems (open items only, seq-ascending, mirroring PhoneCall.postcallactionitems). Voice AI calls fire call.ended on every terminal state (including never-connected ones like no_answer), not just connected calls. An identity may hold a call.ended sub and an imessage.* sub independently, but one subscription carries a single channel.
• Inbound call (flat, synchronous) — PhoneIncomingCallWebhookPayload on a phone number's incomingcallwebhookurl. Not subscribable; the URL stays on the phone-number resource because the response (action: "answer" | "reject" + optional clientwebsocket_url) decides the call's fate. Non-200, invalid bodies, and timeouts are treated as "decline routing" by Inkbox. (Contrast call.ended above, which is the replayable post-call fan-out.)

Subscription resource: inkbox.webhooks.subscriptions.{list,get,create,update,delete}. Each subscription names exactly one owner (mailbox, phone number, or agent identity), one HTTPS destination URL, and a non-empty subset of one channel's event types. Multiple subscriptions on the same owner fan out independently (cap: 20 active per owner). Identity-owned iMessage, call-lifecycle, and A2A subscriptions use separate rows; disjoint rows may share a destination URL. The SDK runs structural + prefix validation client-side (exactly-one-FK, non-empty distinct events, no phone.incomingcall, one channel per row, and event prefixes compatible with the owner). The server remains authoritative for the exact event-name enum, so a typo with a valid prefix (e.g. message.receivedtypo) passes the SDK's check and is rejected as 422 by the server.

create(...) returns a WebhookSubscriptionCreateResponse. The first subscription created for an identity that has no signing key yet carries that identity's signingkey once (otherwise None) — capture it then, it cannot be retrieved again. Every subscription (read or created) also carries owneridentity_id, the resolved owning agent identity.

created = inkbox.webhooks.subscriptions.create(
    mailbox_id=str(mailbox.id), url="https://example.com/hook", event_types=["message.received"],
)
print(created.owner_identity_id)
if created.signing_key:                # populated once if the identity had no key yet
    save_secret(created.signing_key)

Conversation context: opt a mail, text, or iMessage subscription into per-class history on received events (message.received, text.received, imessage.received) with context_config — email / texts / calls, each {"mode": "count", "count": N} (1..50) or {"mode": "window", "hours": H} (1..168). A2A subscriptions do not support conversation context. On update it is tri-state: omit = unchanged, None = clear, dict = replace. Received-event payloads then carry an optional data["context"] keyed by class; optional fields are absent, not null, so read with .get(...). A skipped class ships items: [] plus a skipped reason; call transcript entries are turns or an abridgment marker, discriminated on "marker" in entry. Config types WebhookContextConfig / WebhookContextClassConfig and receiver wire shapes WebhookContextWire / WebhookContextBlockWire / WebhookTranscriptEntryWire (and the item wire types) are exported from inkbox.

inkbox.webhooks.subscriptions.create(
    mailbox_id=str(mailbox.id), url="https://example.com/hook",
    event_types=["message.received"],
    context_config={"email": {"mode": "count", "count": 10}},
)
inkbox.webhooks.subscriptions.update(created.id, context_config=None)  # clear

Delivery auth token: for endpoints that require their own Authorization header, pass authtoken on create / update — every delivery (and replay) then carries Authorization: Bearer <token> alongside the signature headers. Reads return the stored token as authtoken (None when unset) plus the boolean hasauthtoken flag; both default to unset on servers that predate the fields. On update it is tri-state: omit = unchanged, None = clear, string = replace.

Mail contact / identity resolution: data["contacts"] and data["agentidentities"] are lists of {"bucket", "address", "id", ...} entries (always present, possibly empty). Inbound events resolve from + every cc; outbound events resolve every to + cc + bcc. Pair entries to the source field by (bucket, address). Contact entries carry active memory text newest-first in memories; use match.get("memories", []) for older replays. Outbound payloads also carry data["message"]["bccaddresses"] (None on inbound, since BCC is not visible to recipients).

Phone/text contact / identity resolution: data["contacts"] (text) and top-level contacts (inbound call) are lists of {"id", "name", "memories"} matches; memories contains active memory text newest-first. data["agentidentities"] mirrors contacts for matched agent identities but does not carry memories. Scoped to the identity that owns the receiving phone number; both default to [] when nothing matches. Group text events carry per-recipient delivery rows in data["textmessage"]["recipients"]; outbound group lifecycle events name the event target in data["recipientphonenumber"] (one webhook per recipient leg). Inbound and outbound 1:1 events leave data["recipientphonenumber"] as None — the singular peer is already in data["textmessage"]["remotephonenumber"] (inbound) or data["textmessage"]["recipients"][0] (outbound 1:1).

Exported wire types: MailWebhookPayload, TextWebhookPayload, IMessageWebhookPayload, PhoneIncomingCallWebhookPayload, WebhookContact, WebhookAgentIdentity, WebhookMailContact, WebhookMailAgentIdentity, TextMessageRecipientWire, the conversation-context shapes (WebhookContextWire, WebhookContextBlockWire, WebhookTranscriptEntryWire, and item wire types), plus event-type Literal unions (MailWebhookEventType, TextWebhookEventType, IMessageWebhookEventType) and wire enums (MessageStatus, CallStatusWire, HangupReasonWire, SmsDeliveryStatusWire, etc.). All fields are snake_case TypedDicts to match the raw JSON body.

Error Handling

from inkbox import (
    InkboxAPIError,
    DuplicateContactRuleError,
    RedundantContactAccessGrantError,
    StorageLimitExceededError,
)

try:
    identity = inkbox.get_identity("unknown")
except InkboxAPIError as e:
    print(e.status_code)   # HTTP status (e.g. 404)
    print(e.detail)        # str for legacy errors, dict for structured ones
    print(e.agent_support) # Support Agent instructions, or None

InkboxAPIError.detail can now be a dict for structured responses (e.g. contact-rule / access conflicts). Catch the narrower subclasses when you need the parsed fields:

• DuplicateContactRuleError — 409 when creating a contact rule with an already-taken (matchtype, matchtarget) on the same resource. Exposes .existingruleid: UUID.
• RedundantContactAccessGrantError — 409 when an identity-viewer grant is redundant (e.g. a specific viewer on top of an active wildcard). Exposes .error and .detail_message.
• StorageLimitExceededError — 402 when a send / reply-all / forward would push the mailbox past its plan storage cap. Exposes .message, .upgradeurl, and .limitbytes. Delete messages or threads to free space (immediate), or upgrade. A 402 whose detail is a plain string stays a plain InkboxAPIError.

Key Conventions

• All method and property names are snake_case
• iteremails() / iterunread_emails() return Iterator[Message] — auto-paginated, lazy
• list_calls() returns list[PhoneCall] — offset pagination, not an iterator
• To clear a nullable field (e.g. webhook URL), pass field=None
• The Inkbox client must be used as a context manager (with statement) or .close() called manually
• Mail/phone methods on AgentIdentity raise InkboxError if the relevant channel isn't assigned