Skip to main content

Python SDK

The official Python SDK supports both async and sync workflows, includes retries for transient API failures, and exposes webhook verification helpers.

Python 3.9+ is required.

Install​

pip install postmx

Quickstart​

Async​

import asyncio

from postmx import PostMX

async def main() -> None:
    async with PostMX("pmx_live_...") as postmx:
        inbox = await postmx.create_inbox({
            "label": "signup-test",
            "lifecycle_mode": "temporary",
            "ttl_minutes": 15,
        })

        print("Send your app email to:", inbox["email_address"])

        message = await postmx.wait_for_message(
            inbox["id"],
            timeout=30.0,
            interval=1.0,
        )

        print("OTP:", message["otp"])
        print("Intent:", message["intent"])

asyncio.run(main())

Sync​

from postmx import PostMXSync

postmx = PostMXSync("pmx_live_...")
inbox = postmx.create_inbox({
    "label": "signup-test",
    "lifecycle_mode": "temporary",
    "ttl_minutes": 15,
})

message = postmx.wait_for_message(inbox["id"])
print(message["otp"])

Create the client​

Async client​

from postmx import PostMX

client = PostMX("pmx_live_...")

Sync client​

from postmx import PostMXSync

client = PostMXSync("pmx_live_...")

Constructor options​

OptionTypeDefaultNotes
base_urlstrhttps://api.postmx.coOverride for staged or local environments.
max_retriesint2Retries apply to 429, 500, 502, 503, and 504.
timeoutfloat30.0Request timeout in seconds.

Common workflows​

Create an inbox​

inbox = await client.create_inbox({
    "label": "checkout-flow",
    "lifecycle_mode": "temporary",
    "ttl_minutes": 15,
})

ttl_minutes is optional for temporary inboxes. Current API limits are 10 to 60.

List inboxes​

result = await client.list_inboxes(limit=20)

print(result["inboxes"])
print(result["page_info"])
print(result["wildcard_address"])

List messages for an inbox​

result = await client.list_messages(inbox["id"], limit=10)
print(result["messages"])

Get a message​

message = await client.get_message("msg_123")

print(message["subject"])
print(message["text_body"])
print(message["html_body"])
print(message["otp"])
print(message["links"])

Fetch only the part you need​

otp_only = await client.get_message("msg_123", content_mode="otp")
print(otp_only["otp"])

links_only = await client.get_message("msg_123", content_mode="links")
print(links_only["links"])

text_only = await client.get_message("msg_123", content_mode="text_only")
print(text_only["text_body"])

Supported modes:

  • full
  • otp
  • links
  • text_only

Wait for a message​

message = await client.wait_for_message(
    inbox["id"],
    interval=1.0,
    timeout=60.0,
)

Notes:

  • interval defaults to 1.0 seconds.
  • timeout defaults to 60.0 seconds.
  • interval must be at least 0.2 seconds.

Create a webhook​

result = await client.create_webhook({
    "label": "production-events",
    "target_url": "https://example.com/webhooks/postmx",
})

print(result["webhook"]["id"])
print(result["signing_secret"])

Use a final public HTTPS endpoint. The API rejects localhost targets, embedded credentials, and private or reserved IP literals. PostMX does not follow redirects, and each delivery attempt times out after 10 seconds.

Verify a webhook signature​

from postmx import verify_webhook_signature

event = verify_webhook_signature(
    payload=raw_body,
    signature=request.headers.get("x-postmx-signature") or request.headers["postmx-signature"],
    timestamp=request.headers["x-postmx-timestamp"],
    signing_secret=signing_secret,
)

print(event["type"])
print(event["data"]["message"]["otp"])

Important:

  • Store the returned signing_secret immediately. It is returned once.
  • Pass the raw body as bytes or str.
  • Read X-PostMX-Delivery-Id for attempt tracing and X-PostMX-Event-Id for diagnostics.
  • The expected signature format is v1=<base64url(hmac_sha256(signing_secret, timestamp + "." + raw_body))>.
  • The default timestamp tolerance is 300 seconds.
  • The payload already includes the full message plus extracted fields such as otp, links, and intent.

Error handling​

from postmx import PostMXApiError, PostMXNetworkError

try:
    await client.get_message("msg_missing")
except PostMXApiError as error:
    print(error.status)
    print(error.code)
    print(error.request_id)
    print(error.retry_after_seconds)
except PostMXNetworkError as error:
    print(error)

Async vs sync​

  • Use PostMX in async web apps, async workers, and async tests.
  • Use PostMXSync in scripts, sync apps, and basic automation.
  • PostMXSync cannot be used inside an already running event loop. In async environments, use PostMX directly.
  • POST requests are safe to retry because the SDK auto-generates an idempotency_key when you do not pass one.

Method reference​

PostMX(api_key, *, base_url="https://api.postmx.co", max_retries=2, timeout=30.0)
await client.list_inboxes(*, limit=None, cursor=None)
await client.create_inbox(params, *, idempotency_key=None)
await client.list_messages(inbox_id, *, limit=None, cursor=None)
await client.get_message(message_id, *, content_mode=None)
await client.create_webhook(params, *, idempotency_key=None)
await client.wait_for_message(inbox_id, *, interval=1.0, timeout=60.0)