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"]})
updated = client.mailboxes.update_spam_filter("mbx_...", {"threshold": 8})
client.mailboxes.delete("mbx_...")

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_...")

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"}],
})

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"},
})

Newsletters

newsletter_domains = client.newsletters.domains.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_...",
    "newsletter_domain_id": newsletter_domains["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. Block prose fields are plain text in API requests. Use newlines for paragraph breaks, and use body_html or custom_html only when you need raw HTML. 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,
    })

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