npx claudepluginhub inkbox-ai/inkbox --plugin inkboxThis skill uses the workspace's default tool permissions.
API-first communication infrastructure for AI agents — email, phone, encrypted vault, and identities.
Integrates @inkbox/sdk in TypeScript/JavaScript for AI agent features: identities, email, phone/SMS, contacts, notes, rules, vault, tunnels.
Handles Resend email API: send single/batch transactional emails, inbound webhooks, manage templates/domains/contacts/broadcasts/events/logs/API keys/automations, setup SDK with gotchas like idempotency keys.
Send and receive SMS/MMS with Telnyx Python SDK, handle opt-outs and delivery webhooks. For notifications, 2FA, or messaging apps.
Share bugs, ideas, or general feedback.
API-first communication infrastructure for AI agents — email, phone, encrypted vault, and identities.
pip install inkbox
Always use the context manager — it manages the underlying HTTP session:
from inkbox import Inkbox
with Inkbox(api_key="ApiKey_...") as inkbox:
...
Constructor: Inkbox(api_key, base_url="https://inkbox.ai", timeout=30.0)
Inkbox (admin-only client)
├── .create_identity(handle) → AgentIdentity
├── .get_identity(handle) → AgentIdentity
├── .list_identities() → list[AgentIdentitySummary]
├── .mailboxes → MailboxesResource
├── .phone_numbers → PhoneNumbersResource
├── .texts → TextsResource
├── .mail_contact_rules → MailContactRulesResource
├── .phone_contact_rules → PhoneContactRulesResource
├── .contacts → ContactsResource (.access, .vcards)
├── .notes → NotesResource (.access)
├── .vault → VaultResource
├── .whoami() → WhoamiResponse
└── .create_signing_key() → SigningKey
AgentIdentity (identity-scoped helper)
├── .mailbox → IdentityMailbox | None
├── .phone_number → IdentityPhoneNumber | None
├── .credentials → Credentials (requires vault unlocked)
├── 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.
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).
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.update(status="paused") # or "active"
identity.refresh() # re-fetch from API, updates cached channels
identity.delete() # unlinks channels
# Identity is created with a mailbox automatically — provision a phone number
phone = identity.provision_phone_number(type="toll_free") # or type="local", state="NY"
print(identity.email_address) # e.g. "sales-agent@inkboxmail.com"
print(phone.number) # e.g. "+18005551234"
# Link existing channels
identity.assign_mailbox("mailbox-uuid")
identity.assign_phone_number("phone-number-uuid")
# Unlink without deleting
identity.unlink_mailbox()
identity.unlink_phone_number()
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>",
}],
)
# 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)
# Get full thread (oldest-first)
thread = identity.get_thread(msg.thread_id)
for m in thread.messages:
print(f"[{m.from_address}] {m.subject}")
Threads carry a folder field: inbox, spam, archive, or blocked (server-assigned, never client-set).
from inkbox import ThreadFolder
# Thread.folder / ThreadDetail.folder is always one of the four values above.
Low-level folder listing / per-thread updates (list(folder=…), list_folders(email), update(..., folder=…)) live on ThreadsResource. Passing folder="blocked" to update raises ValueError before the HTTP call.
# Place outbound call — stream audio via WebSocket
call = identity.place_call(
to_number="+15167251294",
client_websocket_url="wss://your-agent.example.com/ws",
)
print(call.status)
print(call.rate_limit.calls_remaining)
# List calls (offset pagination)
calls = identity.list_calls(limit=10, offset=0)
for c in calls:
print(c.id, c.direction, c.remote_phone_number, c.status)
# Transcript segments (ordered by seq)
for t in identity.list_transcripts(calls[0].id):
print(f"[{t.party}] {t.text}") # party: "local" or "remote"
Outbound SMS limits and gates (current):
identity.phone_number.sms_status is SmsStatus.PENDING until ready; sends in this window return 409 sender_sms_pending.START to any number in the org. Unknown → 403 recipient_not_opted_in. STOP → 403 recipient_opted_out.Coming soon: toll-free SMS sending, customer-managed 10DLC brands/campaigns (drastically higher per-number limits).
# Send an SMS from this identity's phone number.
# Returns a queued TextMessage; final delivery state arrives via the
# incoming_text_webhook_url configured on the sender.
sent = identity.send_text(to="+15167251294", text="Hello from Inkbox")
print(sent.id, sent.delivery_status) # SmsDeliveryStatus.QUEUED
# 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 conversation summaries (one row per remote number)
convos = identity.list_text_conversations(limit=20)
for c in convos:
print(c.remote_phone_number, c.latest_text, c.unread_count, c.total_count)
# Get messages in a specific conversation
msgs = identity.get_text_conversation("+15167251294", 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("+15167251294")
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")
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 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
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 a login secret (secret_type inferred from payload type)
unlocked.create_secret(
"AWS Production",
LoginPayload(password="s3cret", username="admin", url="https://aws.amazon.com"),
description="Production IAM user",
)
# Create an API key secret
unlocked.create_secret(
"GitHub PAT",
APIKeyPayload(api_key="ghp_xxx"),
)
# Create an SSH key secret
unlocked.create_secret(
"Deploy Key",
SSHKeyPayload(private_key="-----BEGIN OPENSSH PRIVATE KEY-----..."),
)
# Create a freeform secret
unlocked.create_secret("Misc", OtherPayload(data="any freeform content"))
# Update name/description and/or re-encrypt payload
unlocked.update_secret("secret-uuid", name="New Name")
unlocked.update_secret("secret-uuid", payload=LoginPayload(password="new", username="new"))
# Delete
unlocked.delete_secret("secret-uuid")
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
| 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-facing credential access — typed, identity-scoped. The vault stays as the admin surface; identity.credentials is the agent runtime surface.
from inkbox import Credentials
# Unlock the vault first (stores state on the client)
inkbox.vault.unlock("my-Vault-key-01!")
identity = inkbox.get_identity("support-bot")
# Discovery — returns list[DecryptedVaultSecret] with name/metadata
all_creds = identity.credentials.list()
logins = identity.credentials.list_logins()
api_keys = identity.credentials.list_api_keys()
ssh_keys = identity.credentials.list_ssh_keys()
key_pairs = identity.credentials.list_key_pairs()
# Access by UUID — returns typed payload directly
login = identity.credentials.get_login("secret-uuid") # → LoginPayload
api_key = identity.credentials.get_api_key("secret-uuid") # → APIKeyPayload
ssh_key = identity.credentials.get_ssh_key("secret-uuid") # → SSHKeyPayload
key_pair = identity.credentials.get_key_pair("secret-uuid") # → KeyPairPayload
# Generic access — returns DecryptedVaultSecret
secret = identity.credentials.get("secret-uuid")
inkbox.vault.unlock() first — raises InkboxError if vault is not unlockedidentity.refresh() to clear the cacheget_* raises KeyError if not found, TypeError if wrong secret typeTOTP secrets are stored inside LoginPayload.totp in the encrypted vault. Codes are generated client-side — no server call needed.
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)
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)
| 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 |
inkbox.mailboxes)mailboxes = inkbox.mailboxes.list()
mailbox = inkbox.mailboxes.get("abc@inkboxmail.com")
inkbox.mailboxes.update(mailbox.email_address, display_name="New Name")
inkbox.mailboxes.update(mailbox.email_address, webhook_url="https://example.com/hook")
inkbox.mailboxes.update(mailbox.email_address, webhook_url=None) # remove webhook
# Switch contact-rule filter mode (admin-only — agent-scoped keys get 403)
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).
results = inkbox.mailboxes.search(mailbox.email_address, q="invoice", limit=20)
inkbox.mailboxes.delete(mailbox.email_address)
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).
from inkbox import SendingDomainStatus
verified = inkbox.domains.list(status=SendingDomainStatus.VERIFIED)
# Admin-scoped API key only — non-admin keys get 403.
# Returns the bare new default domain name (or None when reverted to platform).
new_default = inkbox.domains.set_default("mail.acme.com")
# Pass the platform domain (e.g. "inkboxmail.com" in prod) to clear the org default.
# Standalone mailbox: pick by domain id.
inkbox.mailboxes.create(
agent_handle="sales-bot",
sending_domain_id=verified[0].id, # verified custom domain
)
inkbox.mailboxes.create(
agent_handle="sales-bot",
sending_domain_id=None, # force platform default
)
# Identity create: pick by bare domain name (not id).
inkbox.create_identity("sales-bot", sending_domain="mail.acme.com")
inkbox.phone_numbers)numbers = inkbox.phone_numbers.list()
number = inkbox.phone_numbers.get("phone-number-uuid")
number = inkbox.phone_numbers.provision(agent_handle="my-agent", type="toll_free")
local = inkbox.phone_numbers.provision(agent_handle="my-agent", type="local", state="NY")
inkbox.phone_numbers.update(
number.id,
incoming_call_action="webhook", # "webhook", "auto_accept", or "auto_reject"
incoming_call_webhook_url="https://...",
)
inkbox.phone_numbers.update(
number.id,
incoming_call_action="auto_accept",
client_websocket_url="wss://...",
)
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 is admin-only and returns a change-notice when the value actually changed.
Per-mailbox or per-phone-number allow/block lists, enforced server-side. The active filter_mode on the owning resource controls whether the rules are interpreted as a whitelist or blacklist. Mail matches by exact email or domain; phone matches by exact E.164 number.
from inkbox import (
MailRuleAction, MailRuleMatchType, PhoneRuleAction, PhoneRuleMatchType,
DuplicateContactRuleError,
)
# Mail rules — scoped to a single mailbox. New rules always start active;
# call `update(..., status="paused")` afterwards to pause one.
rule = inkbox.mail_contact_rules.create(
mailbox.email_address,
action=MailRuleAction.ALLOW, # or BLOCK
match_type=MailRuleMatchType.DOMAIN, # or EXACT_EMAIL
match_target="example.com",
)
inkbox.mail_contact_rules.list(mailbox.email_address)
inkbox.mail_contact_rules.get(mailbox.email_address, rule.id)
inkbox.mail_contact_rules.update(mailbox.email_address, rule.id, status="paused") # admin-only
inkbox.mail_contact_rules.delete(mailbox.email_address, rule.id) # admin-only
# Admin-only list; optionally narrow to a single mailbox_id
all_rules = inkbox.mail_contact_rules.list_all(mailbox_id=str(mailbox.id))
# Duplicate (match_type, match_target) on the same mailbox raises 409:
try:
inkbox.mail_contact_rules.create(
mailbox.email_address,
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
# Phone rules — same shape, only match_type="exact_number" is supported.
inkbox.phone_contact_rules.create(
number.id,
action=PhoneRuleAction.BLOCK,
match_type=PhoneRuleMatchType.EXACT_NUMBER,
match_target="+15551234567",
)
inkbox.phone_contact_rules.list(number.id)
inkbox.phone_contact_rules.list_all(phone_number_id=str(number.id))
Admin-only address book with per-identity access grants and vCard import/export.
from inkbox import (
Contact, ContactEmail, ContactPhone, ContactAddress,
RedundantContactAccessGrantError,
)
# 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")],
# access_identity_ids defaults to "wildcard" (every active identity);
# pass [] for admin-only, or a list of identity UUIDs for explicit grants.
)
inkbox.contacts.get(str(contact.id))
inkbox.contacts.list(q="ada", order="recent", limit=50, offset=0)
inkbox.contacts.update(str(contact.id), job_title="Analyst") # JSON-merge-patch via kwargs
inkbox.contacts.delete(str(contact.id))
# 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")
# Access grants (admin + JWT only; agents can self-revoke)
inkbox.contacts.access.list(str(contact.id))
inkbox.contacts.access.grant(str(contact.id), identity_id="agent-uuid")
inkbox.contacts.access.grant(str(contact.id), wildcard=True) # every active identity
inkbox.contacts.access.revoke(str(contact.id), "agent-uuid")
# Redundant grants (e.g. per-identity on top of wildcard) raise 409
try:
inkbox.contacts.access.grant(str(contact.id), identity_id="agent-uuid")
except RedundantContactAccessGrantError as e:
print(e.error, e.detail_message)
# 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)
vcf = inkbox.contacts.vcards.export_vcard(str(contact.id)) # vCard 4.0 string
Admin-only free-form notes with per-identity access grants. Identities must be granted access explicitly — there is no wildcard for notes.
note = inkbox.notes.create(body="Customer prefers email follow-up.", title="Ada")
inkbox.notes.get(str(note.id))
inkbox.notes.list(q="email", identity_id="agent-uuid", order="recent", limit=50)
inkbox.notes.update(str(note.id), body="Updated body")
inkbox.notes.update(str(note.id), title=None) # clear title (body cannot be null)
inkbox.notes.delete(str(note.id))
# Access grants (admin + JWT only)
inkbox.notes.access.list(str(note.id))
inkbox.notes.access.grant(str(note.id), identity_id="agent-uuid")
inkbox.notes.access.revoke(str(note.id), "agent-uuid")
# 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:
from inkbox import (
AUTH_SUBTYPE_API_KEY_ADMIN_SCOPED,
AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_CLAIMED,
AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_UNCLAIMED,
)
if info.auth_type == "api_key" and info.auth_subtype == AUTH_SUBTYPE_API_KEY_ADMIN_SCOPED:
... # admin-only operations (filter_mode flips, rule updates/deletes, etc.)
Bring a local process online at a public https://{name}.inkboxwire.com URL. Outbound HTTP/2 only — no inbound port to open. POSIX only.
# Forward to a local URL (edge mode — Inkbox terminates TLS at the edge)
listener = inkbox.tunnels.connect(
name="my-app",
forward_to="http://127.0.0.1:8080",
)
print(listener.public_url) # https://my-app.inkboxwire.com
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 (you terminate; SDK auto-signs a cert via the control plane)
listener = inkbox.tunnels.connect(
name="my-app",
tls_mode="passthrough",
forward_to="http://127.0.0.1:8080",
)
Async usage:
async with ...:
listener = inkbox.tunnels.connect(name="my-app", forward_to="http://127.0.0.1:8080")
try:
await listener.serve_forever()
finally:
await listener.aclose()
wait()/close() and serve_forever()/aclose() are mutually exclusive — pick one pair.
CRUD:
inkbox.tunnels.list() # list[Tunnel]
inkbox.tunnels.get("tunnel-uuid")
created = inkbox.tunnels.create(name="my-app", tls_mode="edge")
print(created.connect_secret) # returned ONCE — save it
inkbox.tunnels.delete("tunnel-uuid") # → pending_removal (24h grace)
inkbox.tunnels.restore("tunnel-uuid")
inkbox.tunnels.rotate_secret("tunnel-uuid")
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. The state dir holds the connect secret and (in passthrough) the private key — treat it like an SSH key dir.
For full options, lifecycle notes, and TS examples, see skills/inkbox-tunnels/SKILL.md.
Webhooks are configured directly on the mailbox or phone number — no separate registration.
from inkbox import verify_webhook
# Rotate signing key (plaintext returned once — save it)
key = inkbox.create_signing_key()
# Verify an incoming webhook request
is_valid = verify_webhook(
payload=raw_body, # bytes
headers=request.headers,
secret="whsec_...",
)
Algorithm: HMAC-SHA256 over "{request_id}.{timestamp}.{body}".
from inkbox import (
InkboxAPIError,
DuplicateContactRuleError,
RedundantContactAccessGrantError,
)
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
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 a contact-access grant is redundant (e.g. per-identity grant on top of an active wildcard). Exposes .error and .detail_message.iter_emails() / iter_unread_emails() return Iterator[Message] — auto-paginated, lazylist_calls() returns list[PhoneCall] — offset pagination, not an iteratorfield=NoneInkbox client must be used as a context manager (with statement) or .close() called manuallyAgentIdentity raise InkboxError if the relevant channel isn't assigned