Inkbox Python SDK
API-first communication infrastructure for AI agents — email, phone, encrypted vault, and identities.
Install & Init
pythonpip install inkbox
Always use the context manager — it manages the underlying HTTP session:
pythonfrom 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
Handles are fixed at creation. Change the display name to update an agent's presentation.
pythonidentity = inkbox.create_identity("sales-agent") identity = inkbox.get_identity("sales-agent") identities = inkbox.list_identities() # → list[AgentIdentitySummary] identity.update(display_name="Sales assistant") 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).
Import historical mail
pythonfrom 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
pythonsent = 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
pythonfrom 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).
pythonfrom 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.
pythonfrom 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:
Frommust 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) # connections includes incoming forwarding, live handoffs, and conference guests. # Missing means an older response; an empty list is authoritative. if c.connections is not None: for connection in c.connections: print(connection.kind, connection.trigger, connection.status, connection.target) else: for forwarding in c.forwardings: # Automatic incoming forwarding only. print(forwarding.status, forwarding.target) 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): # remote can be the original caller or a conference guest; party is not a person ID. # phone_number is nullable; shared local lines and unknown attribution have no number. print(f"[{t.phone_number or t.party}] {t.text}") # To press keypad keys during a client-driven call, send this JSON # through the call's media WebSocket: {"event": "dtmf", "digits": "1"} # Each event accepts 1-30 keys from 0-9, *, and #, in order. # At most 20 bursts per call can be outstanding, including the one # being sent; further bursts are ignored. # No acknowledgment is sent; wait for the menu's response before another # burst. Do not retry blindly. # 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_statusisSmsStatus.PENDINGuntil ready; sends in this window return409 sender_sms_pending. - Recipient must have texted
STARTto any number in the org. Unknown →403 recipient_not_opted_in.STOP→403 recipient_opted_out. Inspect / override consent state viainkbox.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 identity 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 personal 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 personal line may start a conversation, subject to consent, contact-rule, and rate-limit checks.
Personal lines can also receive one-to-one messages from non-phone sender
addresses, such as sender@example.com. Read responses and imessage.received
webhooks preserve the address as a string; do not normalize it into a phone number
or assume every received message can be answered. These conversations are
receive-only: sending messages, reactions, typing indicators, or read receipts
returns HTTP 422 with recipient_not_e164. Outbound to recipients must still
be E.164 phone numbers. Reading message history does not send a read receipt.
Discover the router (triage) number at runtime — it can change, so never hardcode it:
pythontriage = 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):
pythonidentity = 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)
Personal 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.
pythonnumbers = inkbox.imessages.list_numbers() # personal lines, 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 personal 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.
pythondedicated_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):
pythonimport re 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, ) # personal 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. # This example selects a phone-number 1:1 message, not a receive-only sender. target = next((m for m in msgs if m.direction == "inbound" and not m.is_group and not m.is_blocked and re.fullmatch(r"\+[1-9][0-9]{1,14}", m.remote_number or "")), None) if target is not None: target_convo = identity.get_imessage_conversation(target.conversation_id) if target_convo.assignment_status == "active": sent_reaction = identity.send_imessage_reaction(message_id=target.id, reaction="like") # Live tapbacks come back on message reads, oldest first. for r in target.reactions or []: print(r.direction, r.reaction, r.custom_emoji) # Take your own tapback back. A failed removal leaves it in place. 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.
pythonfrom 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).
pythonfrom 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.
pythonpage = 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
pythonfrom 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)
pythoninfo = 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.
pythonfrom 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 — raisesInkboxErrorif 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_*raisesKeyErrorif not found,TypeErrorif 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)
pythonfrom 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)
pythonunlocked = 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)
pythonmailboxes = 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, 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 sending_domain_id (standalone) or sending_domain
(identity).
pythonfrom 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)
pythonnumbers = 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:agent@voice.example.com", ) 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 filter_mode / agent_identity_id / filter_mode_change_notice fields as mailboxes; flipping filter_mode here is the deprecated channel path (admin-only; returns a change-notice when the value actually changed). Prefer identity.update(phone_filter_mode="whitelist"), which sets the mode on the identity and does not return a change notice.
Contact Rules
All mutation examples in this section require an admin API key. Exact-address allow/block choices override the email or phone mode. Without an exact choice, matching email domain entries apply in their corresponding mode, then the mode's default applies. Identity-level phone rules do not require a dedicated number. Releasing a number preserves permissions.
Allow/block lists are scoped to the agent identity (mirroring iMessage), addressed by agent_handle. The identity's mail_filter_mode / phone_filter_mode 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.agent_identity_id (not a mailbox/phone-number id).
pythonfrom 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. phone_filter_mode requires the identity to have a phone
number (else a 422).
pythonidentity.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.mail_contact_rules and per-number
inkbox.phone_contact_rules resources still work but hit deprecated server routes
(Sunset 2026-08-31). Prefer the identity-keyed surface above.
python# 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
Shared address book with whole-group email/phone visibility and separate communication choices for each address. Phone covers SMS, calls, and iMessage. Profile and Memories do not grant communication access. Existing-contact identifier changes and suggestion absorption require admin credentials.
With admin credentials, inkbox.contacts.access.get(handle, contact_id) returns email/phone objects with visible and contactable, plus profile and memories. Import ContactChannelAccessUpdate from inkbox; .update(handle, contact_id, email=ContactChannelAccessUpdate(visible=True, contactable=[])) gives View-only email access. A nonempty contactable list allows those current addresses and blocks the rest. Omitted fields are preserved; profile=False hides omitted or empty groups and omitted Memories, while explicit choices win. Visibility alone does not grant communication. The roster's optional access field has the same effective shape. The legacy .access.list(contact_id) remains read-only metadata.
For atomic group-access creation, pass ContactCreatePermissions(identity_id=identity_id, email=ContactChannelAccessUpdate(visible=True, contactable=["ada@example.com"])), with the matching contact email. Use group objects or the older boolean maps, not both.
Use inkbox.contacts.permissions.get(handle, contact_id) with admin credentials to read effective emails and phones boolean maps plus profile and memories booleans. Call .update(handle, contact_id, emails={"ada@example.com": True}, profile=True, memories=False) to save explicit choices. Omitted fields and addresses stay unchanged; no revision is required.
For atomic creation, pass permissions=ContactCreatePermissions(identity_id=identity_id, emails={"ada@example.com": True}, profile=True, memories=False) to inkbox.contacts.create, along with the matching contact email. Import ContactCreatePermissions from inkbox. All initial choices are saved with the contact.
Advanced policies remain under inkbox.contacts.communication_policy: .get(contact_id, identity_id) returns addresses, effective_visibility, visibility, and revision. .replace(contact_id, expected_revision=..., identity_id=..., addresses=[ContactAddressUpdate(kind="email", value="ada@example.com", action="allow", expected_action="inherit")]) makes guarded edits; omitted visibility is preserved. .preview(contact_id, identity_id) and .list_for_identity(handle) return filtered saved views. Agent keys can list only their own view.
With admin credentials, .list_management_for_identity(handle, q="Jane", order="name", limit=20) includes hidden contacts and returns compact contact summaries, visibility settings, and effective access. Email/Phone results are all, some, none, or no_identifiers; Profile/Memories are booleans. Identity-owned communication rules have nullable rule.contact cards, filtered for the caller and without memories.
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.
pythonfrom inkbox import ( Contact, ContactCorrespondenceOptions, ContactEmail, ContactPhone, ContactAddress, ContactReviewStatus, ) # CRUD contact = inkbox.contacts.create( given_name="Ada", family_name="Lovelace", emails=[ContactEmail(label="work", value="ada@example.com")], 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="ada@example.com") 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", "slack"]), ) 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)
contacts.create saves a matching suggested contact instead of failing: when an
email or phone in the request already belongs to an unreviewed contact, that
contact is confirmed and returned with its memories and existing identifiers.
Name fields are replaced; omitted non-name profile fields are preserved, and
supplied non-name fields are applied. This also works with agent-scoped API keys.
When the address belongs to a saved contact, to more than one contact, or is in
conflict, the call still fails with HTTP 409
duplicate_contact_identifier.
Notes
Admin-only free-form notes with per-identity access grants. Identities must be granted access explicitly — there is no wildcard for notes.
pythonnote = 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
python# Check the authenticated caller's identity info = inkbox.whoami() print(info.auth_type) # "api_key" or "jwt" print(info.organization_id)
Returns WhoamiApiKeyResponse (with key_id, label, creator_type, auth_subtype, etc.) or WhoamiJwtResponse (with email, org_role, etc.) based on auth_type.
For branching on API-key scope, compare against the exported constants:
pythonfrom 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. Linux, macOS, and Windows.
python# 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:
pythonasync 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.is_connected is local runtime liveness;
listener.last_connected_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.create_identity(...); there is no standalone create / delete / restore / rotate_secret surface. For passthrough, opt in at create time: inkbox.create_identity("my-app", tunnel={"tls_mode": "passthrough"}) — tls_mode is fixed at create.
Reads + edit:
pythoninkbox.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 api_key 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.api_keys.create(scoped_identity_id=...). Selected connect() kwargs: pool_size (1–32), state_dir (default ~/.inkbox/tunnels/{name}), on_status callback, allow_remote_forwarding=False (loopback-only allowlist), forward_to_verify_tls=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
Availability: Mixed-event identity subscriptions and explicit identity-wide list scope
require SDK/CLI 0.7.8 or later and GET /webhooks/catalog returning
supports_identity_subscriptions: true. Until then, use separate mail, text, and
identity-event subscriptions with their existing mailbox, phone, and identity selectors;
omit the scope option. The examples in this section that combine families or create
mail/text subscriptions by identity assume that capability is available. Explicit
identity scope fails clearly when the capability is missing; it does not fall back to
a partial list.
Webhooks are configured directly on the mailbox or phone number — no separate registration.
pythonimport 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:
Use inkbox.webhooks.subscriptions.list(agent_identity_id=identity.id, scope="identity") to list every notification family, including mixed subscriptions. Omitting scope preserves the legacy single-family views.
- Mail (envelope, fire-and-forget) —
message.received,message.sent,message.forwarded,message.delivered,message.bounced,message.failed. Subscribe viainkbox.webhooks.subscriptions.create(agent_identity_id=..., url=..., event_types=[...]). Onmessage.received,data["message"]includes the plain-textbody(whole under a size cap, else a prefix withbody_truncated: True/body_state: "truncated"); when truncated, fetch the full message withinkbox.messages.get(message["email_address"], message["id"])— useid(row id), notmessage_id(RFC 5322 header). These fields are present-with-nullon the other events and absent on pre-feature payloads. - Text (envelope, fire-and-forget) —
text.received,text.sent,text.delivered,text.delivery_failed,text.delivery_unconfirmed. Subscribe viainkbox.webhooks.subscriptions.create(agent_identity_id=..., url=..., event_types=[...]). The text-message body carriesdelivery_statusas an outbound message-level rollup; 1:1 traffic also hoistserror_code,error_detail,sent_at,delivered_at, andfailed_at. On group outbound those legacy detail fields areNoneand per-recipient state lives inrecipients[]. - iMessage (envelope, fire-and-forget) —
imessage.received,imessage.reaction_received, plus the outbound delivery lifecycleimessage.sent,imessage.delivered,imessage.delivery_failed(declined/error; details on the message object). Subscribe viainkbox.webhooks.subscriptions.create(agent_identity_id=..., url=..., event_types=[...])— owned by the agent identity, since shared iMessage pool numbers are not org resources.data["message"]is populated onimessage.receivedand the three delivery-lifecycle events;data["reaction"]onimessage.reaction_received. Fan-out only happens while the identity is active andimessage_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 viainkbox.webhooks.subscriptions.create(agent_identity_id=..., url=..., event_types=["call.ended"]).CallEndedWebhookPayload.datacarries thecall(WebhookPhoneCall, with derivedduration_seconds), resolvedcontacts/agent_identities, an always-presenttranscript_url(authoritative verbatim, fetch with an admin API key), and an inlinetranscriptblock (WebhookCallTranscript, middle-cut/abridged) present when the platform captured a transcript for the call, otherwiseNone— 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"]carriesmode/reason;datacarriesoutcome("completed" | "no_answer" | "declined" | "failed",Noneiffmodeisclient_websocket) andpost_call_action_items(open items only, seq-ascending, mirroringPhoneCall.post_call_action_items). Voice AI calls firecall.endedon every terminal state (including never-connected ones likeno_answer), not just connected calls. These events can share one subscription with mail, text and A2A notifications. - Inbound call (flat, synchronous) —
PhoneIncomingCallWebhookPayloadon a phone number'sincoming_call_webhook_url. Not subscribable; the URL stays on the phone-number resource because the response (action: "answer" | "reject"+ optionalclient_websocket_url) decides the call's fate. Non-200, invalid bodies, and timeouts are treated as "decline routing" by Inkbox. (Contrastcall.endedabove, which is the replayable post-call fan-out.)
Subscription resource: inkbox.webhooks.subscriptions.{list,get,create,update,delete}. One identity owns each subscription, and its nonempty distinct event selection may combine all notification families even without configured channels. Multiple destinations are supported; overlapping events at the same identity and URL conflict. Use the identity selector; legacy mailbox/phone selectors remain mutually exclusive and resolve to the owning identity. PATCH replaces the full event list. The API validates exact catalog names; the SDK validates event prefixes and excludes phone.incoming_call.
create(...) returns a WebhookSubscriptionCreateResponse. The first subscription created for an identity that has no signing key yet carries that identity's signing_key once (otherwise None) — capture it then, it cannot be retrieved again. Every subscription (read or created) also carries owner_identity_id, the resolved owning agent identity.
pythoncreated = inkbox.webhooks.subscriptions.create( agent_identity_id=identity.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). Other event types, including A2A, ignore this configuration. 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.
pythoninkbox.webhooks.subscriptions.create( agent_identity_id=identity.id, url="https://example.com/hook", event_types=["message.received"], context_config={"email": {"mode": "count", "count": 10}}, ) inkbox.webhooks.subscriptions.update(created.id, scope="identity", context_config=None) # clear
Delivery auth token: for endpoints that require their own Authorization header, pass auth_token on create / update — every delivery (and replay) then carries Authorization: Bearer <token> alongside the signature headers. Reads return the stored token as auth_token (None when unset) plus the boolean has_auth_token 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["agent_identities"] 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"]["bcc_addresses"] (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["agent_identities"] 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["text_message"]["recipients"]; outbound group lifecycle events name the event target in data["recipient_phone_number"] (one webhook per recipient leg). Inbound and outbound 1:1 events leave data["recipient_phone_number"] as None — the singular peer is already in data["text_message"]["remote_phone_number"] (inbound) or data["text_message"]["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
pythonfrom 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(match_type, match_target)on the same resource. Exposes.existing_rule_id: UUID.RedundantContactAccessGrantError— 409 when an identity-viewer grant is redundant (e.g. a specific viewer on top of an active wildcard). Exposes.errorand.detail_message.StorageLimitExceededError— 402 when a send / reply-all / forward would push the mailbox past its plan storage cap. Exposes.message,.upgrade_url, and.limit_bytes. Delete messages or threads to free space (immediate), or upgrade. A402whosedetailis a plain string stays a plainInkboxAPIError.
Key Conventions
- All method and property names are snake_case
iter_emails()/iter_unread_emails()returnIterator[Message]— auto-paginated, lazylist_calls()returnslist[PhoneCall]— offset pagination, not an iterator- To clear a nullable field (e.g. webhook URL), pass
field=None - The
Inkboxclient must be used as a context manager (withstatement) or.close()called manually - Mail/phone methods on
AgentIdentityraiseInkboxErrorif the relevant channel isn't assigned
Use subscriptions.update(sub.id, scope="identity", event_types=events) to replace the selected events and subscriptions.delete(sub.id, scope="identity") to remove the subscription. Delivery rows expose replayable and replay_unavailable_reason without replacing their original webhook_subscription_id.
Custom email signatures
Custom signatures are saved per mailbox and require an eligible paid plan to set or enable.
Each HTML/text field supports up to 16,384 characters. HTML is sanitized; logos
must use absolute HTTPS URLs. Updating HTML without text generates a plain-text
fallback. Omitted fields stay unchanged; null clears saved content. If saved text
is null, sending derives it from HTML when possible. Disable without deleting to
pause automatic insertion. Disabling and clearing remain available on every plan.
Signatures are inserted when mail is sent, including replies, forwards, and sent
drafts; do not append them manually. The Inkbox watermark is controlled separately.
Signed/encrypted SMTP mail cannot have a custom signature inserted; disable
automatic insertion before sending those messages.
A .sig file is not a standardized attachment format: read its UTF-8 text or HTML
content into the corresponding field; images and proprietary formats are not imported.
pythonfrom pathlib import Path inkbox.mailboxes.update( "alex@example.com", signature_html=Path("signature.html").read_text(encoding="utf-8"), signature_enabled=True, ) # Pause without deleting; use None for both content fields to clear them. inkbox.mailboxes.update("alex@example.com", signature_enabled=False)
Verified domains
An organization admin can prove DNS control and attach a domain to an agent.
The domain follows the agent's visibility: public agents show it publicly;
private agents show it to their organization and authorized A2A peers. Proof expires at the
returned valid_until; assertions do not establish legal identity or endorse an
agent. Keep the TXT record in place. Domain certification is separate from custom
email sending domains.
See verified domains for expiry, ownership, and recovery rules. These methods require version 0.7.12 or later.
pythonfrom inkbox import Inkbox client = Inkbox() claim = client.organization_domains.create("example.com") print(claim.dns_record.name, claim.dns_record.value) # Add the exact TXT record before verifying. claim = client.organization_domains.verify(claim.id) if claim.state == "verified": client.identities.set_domain_affiliation("helper", claim.id) for item in client.a2a.iter_public_directory(q="example.com"): print(item.card.name)
Claim methods: create, list, get, verify, delete.
Use client.identities.get_domain_affiliation(handle) to inspect saved settings,
set_domain_affiliation(handle, claim_id) to attach a domain, and
remove_domain_affiliation(handle) to remove the selection. Task and context
participants expose optional .affiliation; message assertions identify the author.
Directory q searches handles, descriptions, skills, and published verified
domains, including domain fragments. Text matches can include unverified agents.
Use verified_domain="example.com" to require an exact current domain in public
results. Preserve the query and filters when following a cursor.
Message retries
Send methods generate a key and preserve it during bounded request retries.
Use an explicit idempotency_key only when application code must retry the same
logical message across calls. Reuse the exact input. A replay returns the original
response; read the message for current delivery status. Request retries do not
guarantee delivery retries. Recover a failed send's key with the public
get_message_request_key(error) helper. Recover a lost ID with
inkbox.message_sends.lookup(...) or lookup_email(...). Never treat a missing
result as proof that submitting a second message is safe.
Slack
In 0.7.14+, client.slack.contact_rules manages directional person and workspace rules.
Use slack_filter_mode or separate slack_inbound_filter_mode /
slack_outbound_filter_mode identity updates for defaults. import_contacts reads
one page per call; follow its cursor. Rule writes,
default changes and import require an organization admin key. Claimed
agent keys can read their own rules. Import never grants messaging or Companion access.
See the Slack API and onboarding guide for implemented SDK/CLI methods.
Use an existing identity. Select a workspace explicitly for live reads and mutations.
Organization-member sessions, organization admin API keys, and claimed agent keys can
save and list setup workspaces in their organization. Claimed agent keys can prepare
and install only their own identity’s app; organization credentials can select an
identity in their organization. Installation availability does not imply preparation
is ready. Save both app-configuration tokens for the target workspace, then prepare
the identity’s app using that saved provisioning-workspace UUID.
Use save_provisioning_workspace / saveProvisioningWorkspace or CLI
slack provisioning-workspace save --credentials-file <path> (use - for stdin).
Reuse safe metadata from list_provisioning_workspaces / listProvisioningWorkspaces.
Tokens are write-only; never put them in command arguments or output.
Pass the saved ID to start_setup (Python/Rust), startSetup (TypeScript), or
slack setup start --provisioning-workspace-id <uuid>.
Poll connection reads until setup.status is ready, with a bounded wait and a few
seconds between reads. For needs_credentials, select a saved workspace or save valid configuration credentials.
Do not blindly retry an unknown setup outcome. Start installation only when ready.
Open the returned URL in a browser; treat the full URL as a secret. After approval,
list connections again to confirm the expected workspace is connected. The app is
bound to its chosen workspace; no client-invitation workflow is supported.
Join accessible
public channels or invite the agent to selected private channels. Slack Connect is
supported when the selected connection has access.
App removal (0.7.23+). For user-requested app removal, use an organization admin API key;
claimed agent keys cannot remove apps or read removal status. Removal is not the
default repair for a reauthorization error. Read the connection list and check
application_created / applicationCreated before using its application generation.
Generate a fresh request UUID once and save it with that generation before removal;
retry uncertain requests with those same values, not a newly read generation.
The removal operation ID equals its saved request UUID.
Use admin_client authenticated with an organization admin key:
pythonfrom uuid import uuid4 current = admin_client.slack.list_connections(identity_id) if not current.application_created or current.application_generation is None: raise RuntimeError("No Slack app is available to remove") generation = current.application_generation request_id = uuid4() # Save both values before removal; reuse them on retries. removed = admin_client.slack.remove_application( identity_id, expected_application_generation=generation, request_id=request_id, ) status = admin_client.slack.get_application_removal(removed.operation_id)
Poll the saved operation ID every few seconds with a bounded wait while deletion
is pending, including when the error is credentials_busy. failed and unknown
are terminal; replay does not retry deletion. Removal always attempts permanent
Slack-side deletion, but credential errors or timeouts do not prove the app is gone.
If needed, check and delete the original app in Slack with its original creator.
Removal never prepares another app. Previous connection history becomes unavailable;
the Inkbox identity, other channels, and memory are unchanged. For a fresh app, select
the intended saved workspace-and-creator record (or save another token pair), start
setup explicitly, wait until ready, authorize installation, and invite the new bot
back to channels. A retained non-null app generation does not mean an app exists.
Use explicit connection IDs and stable caller-provided idempotency keys for sends and
utility mutations (reactions, pins, own-message edits/deletions, join/leave, uploads,
and native processing status). Poll sends only while sending and operations only while
in_progress. Unknown is terminal uncertainty and must not be blindly repeated.
Recover lost send responses with get_action_by_key / getActionByKey or CLI
slack action get-by-key --connection-id <uuid> --idempotency-key <key>.
A 404 does not prove no send occurred; never use missing lookup data to justify a
new key. Fresh failed rate-limited sends may include retry_after / retryAfter
seconds; honor that delay before a deliberate new attempt. Stored action reads and
same-key replays do not retain this hint. A recorded terminal action is never resent
by replaying its key. Send
and utility keys use independent per-connection namespaces; utilities emit no outcome
webhook, so read their status through operation lookup. Inspect
capabilities for missing scopes; native processing support remains workspace-dependent.
General file uploads accept standard base64 for 1 byte..10 MiB (CLI: a local --file).
Retained history is separate from bounded live reads and webhook diagnostics. Capture
is automatic for observed messages in accessible conversations, with no time-based
retention limit. Organization-member sessions and organization admin API keys can
disconnect connections, set retention, or purge retained history; claimed agent keys
cannot. Purging history does not stop capture. Omitted retention resets to no time limit.
For search, start with client.slack.search_messages("release notes") across the identity's workspace connections.
Do not loop over connections or require a connection ID for a general search.
Agent credentials infer the identity; other credentials require an explicit identity.
Use a connection filter only to narrow the search. Results include connection IDs.
Use plain English keywords, ranked by relevance then recency; not Slack query
operators or semantic search. Attachment bodies are not indexed. Follow the returned
cursor with the same filters even for short or empty pages, until no cursor remains.
Search errors are not evidence of no matches.
Use archive listing, bounded backfill/restart, and coverage; do not infer complete workspace/thread history
from one page or a completed channel import. Purge deletes retained history without stopping new capture. Archive reads require current connection/conversation access.
In 0.7.15+, use list_archived_messages(connection_id, roots_only=True, include=["conversation", "sender", "reactions", "files"]) for a main timeline with cached display context.
Use list_cached_emoji, download_cached_media, and download_file_preview for cached media. Follow emoji cursors explicitly and inspect directory status;
missing permissions or pending synchronization can leave an empty incomplete page.
Unknown reaction/thread counts are not zero, and known reaction counts do not imply complete actor lists.
Omit the roots-only option when reading a selected thread. Expansions do not apply to ranked search.
Keep caches separate by connection and invalidate them when its generation changes. Never forward
Inkbox credentials to fallback image URLs. Current message and file access checks still apply.
Slack webhooks select incoming messages with slack.dm_received,
slack.group_dm_received, slack.channel_message_received, slack.mention_received,
and slack.thread_reply_received. Overlapping selections produce one logical delivery per subscription,
choosing the first selected match in mention, thread, DM, group DM, channel priority.
Subscriptions cover all accessible conversations across connected workspaces.
There are no Slack-specific filters. Context applies only to received mail, text,
and iMessage events; Slack historical delivery replay is unsupported. The runtime owns attention rules,
watched threads, and its own memory. Webhook delivery order is not guaranteed.
Threaded iMessage replies
Requires SDK/CLI 0.7.13 or later. These reply examples require a phone-number conversation or a supported group; non-phone one-to-one conversations remain receive-only, including threaded replies.
pythonmessage = identity.get_imessage(message_id) reply = identity.send_imessage( conversation_id=message.conversation_id, reply_to_message_id=message.id, text="Agreed — let's use that option.", ) page = identity.get_imessage_thread(message.id, limit=50) if page.next_cursor: next_page = identity.get_imessage_thread(message.id, cursor=page.next_cursor) if message.thread_id: page = identity.get_imessage_conversation_thread(message.conversation_id, message.thread_id) rows = identity.list_imessages(conversation_id=message.conversation_id, thread_id=message.thread_id)
IMessage and message webhook payloads expose nullable reply_to_message_id,
thread_id, and thread_root_message_id. Resource-level equivalents are
imessages.get, get_thread, get_conversation_thread, send, and list;
these also accept agent_identity_id.
Thread IDs are opaque and distinct from message IDs. Thread pages include the
root and its replies in chronological order; follow next_cursor (Python/Rust)
or nextCursor (TypeScript/CLI) until null. A standalone message can have its own
thread ID and use its own message ID as the root, even before anyone replies.
A non-null reply_to_message_id / replyToMessageId identifies a visible reply
parent; a thread ID alone does not mean the message has replies. Thread metadata
may be null for pending or older messages, and the root or direct parent can be
unavailable. Conversation message lists remain flat and newest-first. A thread filter requires its conversation ID; it uses the existing
limit/offset pagination, unlike the chronological thread endpoints.
Native replies work in supported one-to-one and group iMessage conversations. Use a message from the same conversation. With plain fallback enabled (the default), the API sends an ordinary message in that conversation when native threading is unsupported: the target is known to use SMS/RCS or was downgraded. This does not force a particular transport. An ordinary fallback has no reply parent and does not join the target's native thread. Invalid or inaccessible targets, unsettled messages, and missing reply metadata still fail. Delivery errors are not retried as new ordinary messages. If a previously supported target can no longer receive a native reply, that send may fail even with fallback enabled. Read the message status and error fields after queueing; delivery webhooks do not cover every failure before dispatch. Omitting the target preserves ordinary sending.
Set plain_reply_fallback=False on send_imessage or imessages.send to
require a native reply instead.


