# Inkbox Python

> Use when writing Python code that imports from `inkbox`, uses `pip install inkbox`, or when adding email, mailbox imports, phone, text/SMS, iMessage, A2A task/message history, contacts, notes, contact rules, vault, tunnels, mailbox storage, mail clients (IMAP/SMTP), or agent identity features using the Inkbox Python SDK.

- Skill: `inkbox-ai/inkbox-python` (Agent Skill)
- Install (CLI): `npx skillmds@latest add inkbox-ai/inkbox-python`
- Raw SKILL.md: https://api.skillmd.com/api/skills/inkbox-ai/inkbox-python/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: inkbox-ai (https://skillmd.com/u/inkbox-ai)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/inkbox-ai/inkbox-python

---


# Inkbox Python SDK

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

## Install & Init

```python
pip install inkbox
```

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

```python
from inkbox import Inkbox

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

Constructor: `Inkbox(api_key, base_url="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  (.permissions, .communication_policy, .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(...) / ...  (writes require admin credentials)
├── .get_signing_key_status() / .create_signing_key()
├── .list_contact_communication_policies() → ContactCommunicationPolicyPage
├── 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.verify_signup(api_key, ...)`, `Inkbox.resend_signup_verification(api_key)`, `Inkbox.get_signup_status(api_key)`.

## Identities

```python
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

```python
# Identity is created with a mailbox AND tunnel atomically — both come back on the response
print(identity.email_address)            # e.g. "sales-agent@inkboxmail.com"
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

```python
from inkbox import MailImportFormat

created = inkbox.mailboxes.imports.create(
    email,
    source_format=MailImportFormat.AUTO,
    original_addresses=["old@example.com"],
)
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: `refresh_upload_target(email, job_id)`
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
`original_addresses` per job, 65,000 entries per ZIP, 20 jobs per organization
per 24 hours (`MailImportQuotaExceededError.retry_after_seconds`), and one
in-flight import per mailbox.

### Send

```python
sent = identity.send_email(
    to=["user@example.com"],
    subject="Hello",
    body_text="Hi there!",          # plain text (optional)
    body_html="<p>Hi there!</p>",   # HTML (optional)
    cc=["cc@example.com"],          # optional
    bcc=["bcc@example.com"],        # 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

```python
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=["user@example.com"]),
    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 `idempotency_key` 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 `forward_message_id`.
Use the latest returned `generation` for every mutation. A `part_index` 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 `draft_generation_conflict` and retry
the same ID and generation on `draft_send_in_progress`. Never resend
`draft_delivery_uncertain`; after checking sent mail, duplicate or delete it instead.

### Read

```python
# 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).

```python
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** — `send_email`, `reply_all_email`, and `forward_email` (and the `inkbox.messages.*` equivalents) — raise `StorageLimitExceededError` (HTTP 402) when the send would push the mailbox over it.

```python
from inkbox import StorageLimitExceededError

try:
    identity.send_email(to=["user@example.com"], 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(...)`): `storage_used_bytes` and `storage_limit_bytes` (`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 `sent_body == fetched_body` 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.

| Setting | Value |
|---|---|
| IMAP host | `imap.inkboxmail.com` |
| IMAP port | `993` (IMAPS / implicit TLS) |
| SMTP host | `smtp.inkboxmail.com` |
| SMTP port | `465` (SMTPS / implicit TLS) or `587` (STARTTLS) |
| Username | the inbox address (e.g. `sales-agent@inkboxmail.com`) |
| Password | an **identity-scoped** API key (`ApiKey_...`) |

The password is the same agent-scoped key an identity-scoped `Inkbox(...)` client authenticates with; mint one with `inkbox.api_keys.create(scoped_identity_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

```python
# 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.",
    # Optional: on_voicemail (OnVoicemail.LEAVE_MESSAGE | HANG_UP | IGNORE;
    # hosted calls default to leave_message) and voicemail_message (what Voice
    # AI says; requires leave_message). VoicemailDetection is deprecated.
)
print(call.mode, call.reason, call.on_voicemail)
# 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 sender_rate_limited`.
- New local numbers need **~10-15 min** for 10DLC carrier propagation. `identity.phone_number.sms_status` is `SmsStatus.PENDING` until ready; sends in this window return `409 sender_sms_pending`.
- Recipient must have texted **`START`** to any number in the org. Unknown → `403 recipient_not_opted_in`. `STOP` → `403 recipient_opted_out`. Inspect / override consent state via `inkbox.sms_opt_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.

```python
# 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:

```python
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`):

```python
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.

```python
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. `imessage_number_id=None` is
intentional wire data that moves an identity back to shared service; omitting
the argument leaves its attachment unchanged.

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

```python
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
```

`claim_number` and atomic identity claims may raise
`DedicatedIMessageNumberQuotaExceededError`,
`DedicatedIMessageNumberInventoryPendingError`, or
`IdempotencyKeyReusedError`. The inventory error exposes
`retry_after_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 `agent_identity_id` / `is_blocked` filters):

```python
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)
identity.release_imessage_assignment(connections[0].id)  # admin key only; they can reconnect via triage
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:

Phone rules cover SMS, calls, and iMessage together. Creation, updates, and deletion require admin credentials. An agent key can inspect permitted rules but cannot authorize itself; a user changes permissions in the Inkbox Console.

```python
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 customer_campaign_required` on writes — `source="api"` writes record an audit event).

```python
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.a2a_invitations.create(peer_agent_handles, recipient_email=...,
expires_in_seconds=...)`, `.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; `extract_a2a_invitation_token()` performs the same strict local
normalization. Unbound create responses may reveal `invitation_token`,
`invitation_url`, and `agent_handoff_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 `a2a_tasks` for the receiver inbox; `a2a_sent_tasks` is the
outbound-only alias.

```python
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`, `requester_handle`, `worker_handle`, `state`,
`context_id`, `q`, `since`, `cursor`, `limit`. Message filters additionally
support `task_id` 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 `a2a_task` / `a2a_sent_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 `context_id` is supplied without
`task_id`. 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:
`a2a_set_publicly_discoverable`, `a2a_set_filter_mode`, `a2a_add_contact_rule`, `a2a_update_contact_rule`, and
`a2a_delete_contact_rule`. Use `a2a_reset_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

```python
# 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

```python
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

```python
# Create a login secret (secret_type inferred from payload type)
unlocked.create_secret(
    "Example dashboard",
    LoginPayload(password="example-password", username="admin", url="https://dashboard.example.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)

```python
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

| Type | Class | Fields |
|------|-------|--------|
| `login` | `LoginPayload` | `password`, `username?`, `email?`, `url?`, `notes?` |
| `api_key` | `APIKeyPayload` | `api_key`, `endpoint?`, `notes?` |
| `key_pair` | `KeyPairPayload` | `access_key`, `secret_key`, `endpoint?`, `notes?` |
| `ssh_key` | `SSHKeyPayload` | `private_key`, `public_key?`, `fingerprint?`, `passphrase?`, `notes?` |
| `other` | `OtherPayload` | `data` |

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

```python
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)

```python
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="user@example.com",
        password="s3cret",
        totp=parse_totp_uri("otpauth://totp/GitHub:user@example.com?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)

```python
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

| Field | Type | Description |
|---|---|---|
| `code` | `str` | The OTP code (e.g. `"482901"`) |
| `period_start` | `int` | Unix timestamp when the code became valid |
| `period_end` | `int` | Unix timestamp when the code expires |
| `seconds_remaining` | `int` | Seconds until expiry |

## Admin-only Resources

### Mailboxes (`inkbox.mailboxes`)

```python
mailboxes = inkbox.mailboxes.list()
mailbox   = inkbox.mailboxes.get("abc@inkboxmail.com")

# 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, f

…(truncated)