Shipmail Python SDK

Official Python SDK for the Shipmail API. Provides both synchronous and asynchronous clients. Requires Python 3.10+.

Installation

pip install shipmail

Quick Start

from shipmail import Shipmail

client = Shipmail("sm_live_...")

# Create a domain
domain = client.domains.create({"name": "example.com"})

# Send an email
message = client.messages.send({
    "mailbox_id": "mbx_...",
    "to": [{"address": "user@example.com"}],
    "subject": "Hello",
    "text": "Hi there",
    "client_reference": "crm-123",
    "metadata": {"campaign": "onboarding"},
    "source_rfc_message_id": "<crm-123@example.com>",
})
same_message = client.messages.list({"client_reference": "crm-123"})

Async

from shipmail import AsyncShipmail

async with AsyncShipmail("sm_live_...") as client:
    domain = await client.domains.create({"name": "example.com"})

    message = await client.messages.send({
        "mailbox_id": "mbx_...",
        "to": [{"address": "user@example.com"}],
        "subject": "Hello",
        "text": "Hi there",
    })

Configuration

from shipmail import Shipmail

client = Shipmail(
    "sm_live_...",
    base_url="https://shipmail.to/api/v1",  # default
    max_retries=2,      # default, retries on 5xx and 429
    timeout=30.0,       # default, in seconds
    organization_id="00000000-0000-4000-8000-000000000123",
)

Resources

Domains

domain = client.domains.create({"name": "example.com"})
domains = client.domains.list({"limit": 10})
domain = client.domains.get("dom_...")
updated = client.domains.update("dom_...", {"catch_all_mailbox_id": "mbx_..."})
client.domains.delete("dom_...")
result = client.domains.verify("dom_...")
records = client.domains.get_dns_records("dom_...")

Mailboxes

mailbox = client.mailboxes.create({
    "domain_id": "dom_...",
    "address": "hello",
    "password": "StrongPass123",
    "display_name": "Hello",
})
mailboxes = client.mailboxes.list({"domain_id": "dom_..."})
mailbox = client.mailboxes.get("mbx_...")
updated = client.mailboxes.update("mbx_...", {"display_name": "New Name"})
client.mailboxes.suspend("mbx_...")
client.mailboxes.resume("mbx_...")
updated = client.mailboxes.reset_password("mbx_...", {"password": "NewPassword1"})
forwarding = client.mailboxes.create_forwarding("mbx_...", {"destination": "owner@example.net"})
forwarding_list = client.mailboxes.list_forwarding("mbx_...")
client.mailboxes.delete_forwarding("mbx_...", forwarding["id"])
app_password = client.mailboxes.create_app_password("mbx_...", {
    "name": "Desktop mail",
    "expires_at": "2026-10-01T00:00:00Z",
})
app_passwords = client.mailboxes.list_app_passwords("mbx_...")
client.mailboxes.revoke_app_password("mbx_...", app_password["id"])
folders = client.mailboxes.list_folders("mbx_...")
folder = client.mailboxes.create_folder("mbx_...", {"name": "VIP", "parent_id": None})
client.mailboxes.update_folder("mbx_...", folder["id"], {"name": "VIP Clients"})
client.mailboxes.delete_folder("mbx_...", folder["id"])
identities = client.mailboxes.list_identities("mbx_...")
rules = client.mailboxes.get_rules("mbx_...")
rules = client.mailboxes.update_rules("mbx_...", {
    "rules": [
        *rules["rules"],
        {
            "id": "4f5a9d74-b0f1-49a7-bbfb-1f2af841f5b2",
            "name": "Flag invoices",
            "enabled": True,
            "position": len(rules["rules"]),
            "match_mode": "all",
            "stop": False,
            "conditions": [{"type": "subject_contains", "value": "invoice"}],
            "actions": [{"type": "star"}, {"type": "send_webhook"}],
        },
    ],
})
client.mailboxes.delete("mbx_...")

mailbox_id = "550e8400-e29b-41d4-a716-446655440000"
queue = client.mailboxes.list_inbox_threads(mailbox_id, {
    "attention_state": "needs_reply",
    "after": "2025-07-20T00:00:00.000Z",
})
candidate = queue["data"][0]
draft = client.mailboxes.create_inbox_reply_draft(
    mailbox_id,
    candidate["thread_id"],
    {"text": "Thanks for the note.", "expected_version": candidate["version"]},
)
# Apply your approval policy first. Stale versions fail with 409 without delivery.
client.mailboxes.send_inbox_reply_draft(mailbox_id, candidate["thread_id"], draft["id"])

Messages

message = client.messages.send({
    "mailbox_id": "mbx_...",
    "to": [{"address": "user@example.com", "name": "User"}],
    "cc": [{"address": "cc@example.com"}],
    "subject": "Hello",
    "html": "<p>Hi there</p>",
    "text": "Hi there",
})

message = client.messages.get("msg_...")

analytics = client.messages.list_analytics({
    "updated_after": "2026-07-01T00:00:00.000Z",
    "limit": 100,
})
# Follow analytics["pagination"]["next_cursor"], then persist snapshot_at.

Scheduled messages and attachments

Stage raw files up to 25 MB and use the returned opaque ID instead of embedding base64 in JSON. Base64 attachments remain supported for existing clients.

with open("invoice.pdf", "rb") as file:
    attachment = client.mailboxes.stage_attachment(
        "mbx_...",
        filename="invoice.pdf",
        content_type="application/pdf",
        data=file.read(),
    )

scheduled = client.messages.send({
    "mailbox_id": "mbx_...",
    "to": ["customer@example.com"],
    "subject": "Invoice",
    "text": "Attached.",
    "staged_attachment_ids": [attachment["id"]],
    "scheduled_at": "2026-08-01T08:00:00.000Z",
})

pending = client.scheduled_messages.list()
detail = client.scheduled_messages.get(scheduled["id"])
client.scheduled_messages.update(scheduled["id"], {
    "to": detail["to"],
    "subject": detail["subject"],
    "text": detail.get("text", ""),
    "staged_attachment_ids": [attachment["id"]],
    "scheduled_at": "2026-08-02T08:00:00.000Z",
})
client.scheduled_messages.cancel(scheduled["id"])

Staged IDs expire after 24 hours and are bound to the API key, organization, and mailbox that created them.

Browser-hosted components can keep the Shipmail API key off the page by calling client.mailboxes.prepare_staged_attachment_upload(...). It returns a five-minute, single-use upload URL bound to the filename, MIME type, exact byte size, and lowercase SHA-256 digest. Upload the raw bytes without credentials or redirects, then use the returned sat_... ID in a send.

Sandbox

Use an sm_test_... API key to simulate sends and inbound replies without sending real email:

test_client = Shipmail("sm_test_...")
test_client.messages.send({
    "mailbox_id": "mbx_...",
    "to": ["customer@example.com"],
    "subject": "Sandbox test",
    "text": "Not delivered",
    "sandbox_outcome": "bounced",
})
test_client.mailboxes.inject_sandbox_inbound("mbx_...", {
    "from_": "customer@example.com",
    "subject": "Re: Sandbox test",
    "text": "Fake inbound reply",
})

Threads

threads = client.threads.list({"mailbox_id": "mbx_..."})
thread = client.threads.get(threads["data"][0]["id"], {"mailbox_id": "mbx_..."})
reply = client.threads.reply(threads["data"][0]["id"], {
    "mailbox_id": "mbx_...",
    "text": "Thanks for your email",
    "to": [{"address": "user@example.com"}],
})

Reply scans

Use a durable, atomically captured scan for a historical window, then page every result using the opaque cursor unchanged. Creation returns the completed snapshot; retry a 409 with bounded backoff while historical header classification finishes. Scans are retained for 30 days.

from datetime import datetime, timedelta, timezone

scan = client.reply_scans.create({
    "mailbox_ids": ["550e8400-e29b-41d4-a716-446655440000"],
    "after": (datetime.now(timezone.utc) - timedelta(days=365)).isoformat(),
})
results = client.reply_scans.list_results(scan["id"], {"limit": 100})
print(results["data"])

Audiences

audience = client.audiences.create({
    "name": "Newsletter",
    "consent_source": "Website signup form",
})

client.audiences.subscribers.add(audience["id"], {
    "email_address": "jane@example.com",
    "merge_fields": {"plan": "pro"},
})

client.audiences.feeds.update(audience["id"], {
    "enabled": True,
    "title": "Release notes",
    "canonical_url": "https://example.com/feed.xml",
    "entry_limit": 25,
})
# Graceful migration: the current URL redirects to the replacement.
client.audiences.feeds.rotate(audience["id"])
# Leaked URL: immediately invalidate both current and previous URLs.
client.audiences.feeds.revoke(audience["id"])

Newsletters

sender_identities = client.newsletters.sender_identities.list({"limit": 25})
with open("hero.png", "rb") as f:
    hero = client.newsletters.assets.upload({
        "filename": "hero.png",
        "content_type": "image/png",
        "data": f.read(),
    })
existing_hero = client.newsletters.assets.register_from_url({
    "url": "https://cdn.shipmail.to/newsletter-images/org_123/hero.png",
    "filename": "hero.png",
})

newsletter = client.newsletters.create({
    "audience_id": "aud_...",
    "sender_identity_id": sender_identities["data"][0]["id"],
    "name": "July changelog",
    "subject": "What shipped in July",
    "preview_text": "A quick product update",
    "blocks": [
        {"type": "heading", "level": 1, "text": "July updates"},
        {"type": "callout", "variant": "info", "title": "Quick note", "body": "A short intro."},
        {"type": "paragraph", "body": "A quick product update."},
        {"type": "image", "url": hero["url"], "alt": "Product screenshot"},
        {"type": "image", "url": existing_hero["url"], "alt": "Existing CDN screenshot"},
        {
            "type": "columns",
            "ratio": "50-50",
            "left": {"title": "For teams", "body": "Shared inbox improvements."},
            "right": {"title": "For agents", "body": "API and MCP improvements."},
        },
    ],
})

client.newsletters.preview(newsletter["id"])

client.newsletters.send_test(newsletter["id"], {
    "recipient_email": "owner@example.com",
})

client.newsletters.preflight(newsletter["id"])

client.newsletters.schedule(newsletter["id"], {
    "scheduled_at": "2026-08-01T09:00:00.000Z",
})

Newsletter test sends and schedules must pass preflight. Guardrail failures raise ValidationError with the failed preflight items in err.details. Preflight responses include url_breakdown so you can see which links, image URLs, and video thumbnails contribute to deliverability checks. Paragraph, quote, callout, list-item, and column bodies accept bare text or sanitized inline HTML. Use <p> and <br> for line breaks. Allowed tags are a, b, br, code, em, i, p, s, span, strong, and u. Use body_html or custom_html for a fully custom email-safe layout. Concurrent newsletter updates can raise ConflictError (409). Fetch the latest newsletter, merge your changes, and retry the update.

Webhooks

webhook = client.webhooks.create({
    "url": "https://example.com/webhook",
    "events": ["message.received", "message.sent"],
    "description": "My webhook",
})
# webhook["secret"] is only available at creation time

webhooks = client.webhooks.list()
webhook = client.webhooks.get("whk_...")
updated = client.webhooks.update("whk_...", {"active": False})
client.webhooks.delete("whk_...")

rotated = client.webhooks.rotate_secret("whk_...")
test = client.webhooks.test("whk_...")
deliveries = client.webhooks.list_deliveries("whk_...")
delivery = client.webhooks.get_delivery("whk_...", "dlv_...")
replay = client.webhooks.replay_delivery(
    "whk_...",
    "dlv_...",
    {"idempotency_key": "replay-dlv-123"},
)

Partner beta

Approved partner accounts can create isolated operator-owned organizations. Use a separate client for delegated infrastructure:

child = client.partner.create_organization(
    {
        "name": "Operator",
        "external_reference": "operator_123",
        "owner_email": "owner@example.com",
        "mailbox_limit": 3,
        "data_classification": "internal_test",
    },
    {"idempotency_key": "operator-123"},
)

delegated = Shipmail(
    "sm_live_...",
    organization_id=child["organization_id"],
)
domains = delegated.domains.list()
mailbox = delegated.mailboxes.create({
    "domain_id": "dom_...",
    "address": "support",
    "generate_password": True,
})
grants = client.partner.list_mailbox_credential_grants()
credential = client.partner.consume_mailbox_credential_grant(
    grants["data"][0]["id"],
    {"name": "Embedded webmail"},
)
usage = client.partner.usage()

The beta requires Shipmail approval and externally owned domains. Delegated context cannot access mail content, exports, suppressions, billing, or password endpoints. Delegated mailbox creation must use generate_password: True; the generated primary password is never returned to the partner. The operator creates a one-time credential grant. Consuming it requires the exact partner:mailbox_credentials:issue scope and returns the app-password secret once. App-password creation and grant consumption do not accept idempotency keys because their plaintext response must never be cached.

Status

status = client.status.get()

Pagination

List methods return a paginated response with cursor-based pagination:

page = client.domains.list({"limit": 10})
print(page["data"])        # list of domains
print(page["pagination"])  # {"next_cursor": ..., "has_more": ...}

# Fetch next page
if page["pagination"]["has_more"]:
    next_page = client.domains.list({
        "cursor": page["pagination"]["next_cursor"],
        "limit": 10,
    })

Cursors are opaque and operation-specific. Never parse, modify, or fabricate them. Inbox and reply queue cursors are bound to their mailbox, time window, sort, and filters.

Auto-pagination iterates through all pages automatically:

for domain in client.domains.list_auto_paginating(limit=25):
    print(domain["name"])

# Async
async for domain in client.domains.list_auto_paginating(limit=25):
    print(domain["name"])

Webhook Verification

Verify incoming webhook signatures without instantiating a client:

from shipmail import verify_webhook, WebhookVerificationError

try:
    event = verify_webhook(raw_body, headers, webhook_secret)
    print(event["event_type"])  # e.g., "message.received"
    print(event["data"])
except WebhookVerificationError:
    # Invalid signature
    pass

Error Handling

The SDK raises typed exceptions that map to API error responses:

from shipmail import (
    ShipmailError,
    AuthenticationError,
    AuthorizationError,
    ValidationError,
    NotFoundError,
    RateLimitError,
    ConflictError,
    InternalServerError,
    APIConnectionError,
)

try:
    client.domains.create({"name": ""})
except ValidationError as err:
    print(err)             # Error message
    print(err.details)     # Field-level validation errors
except RateLimitError as err:
    print(err.retry_after) # Seconds to wait
except ShipmailError as err:
    print(err.status)      # HTTP status code
    print(err.type)        # Error type string
    print(err.request_id)  # Request ID for support
    print(err.retryable)   # Whether the request can be retried

Retries

The SDK automatically retries on 5xx errors and 429 (rate limit) responses with exponential backoff and jitter. Configure with max_retries (default: 2, meaning up to 3 total attempts).

client = Shipmail("sm_live_...", max_retries=0)  # Disable retries

License

MIT