xlambda-email

Send email from your own domain, and — the reason this package exists — turn on receiving with almost no glue code.

pip install xlambda-email

Python port of @xlambda-tech/email; same API, same webhook signature scheme.

Receiving inbound mail

from xlambda.email import create_inbound_handler
from xlambda.core.webhooks.adapters.flask import create_flask_blueprint

handler = create_inbound_handler(
    secret=os.environ["XLAMBDA_WEBHOOK_SECRET"],
    on_received=lambda msg: InboundEmail.objects.create(
        sender=msg["from"], subject=msg["subject"], html=msg["htmlBody"]
    ),
)

app.register_blueprint(create_flask_blueprint(handler), url_prefix="/webhooks/email")

That's the whole integration. Point your project's webhookUrl (dashboard → Settings → Webhooks) at this endpoint and inbound mail on your managed domain starts landing in your own database.

This is the only place a message's body ever exists. The platform stores no message content server-side by design — only metadata (from, to, subject, size) shows up in client.domains.messages(). If your handler doesn't persist what it needs on receipt, it's gone. Deliveries do retry 3× with exponential backoff if your endpoint errors, so a transient failure isn't fatal — an on_received that never runs at all is.

On FastAPI, use the async handler so your callback doesn't block the loop:

from xlambda.email import create_async_inbound_handler
from xlambda.core.webhooks.adapters.fastapi import create_fastapi_router

handler = create_async_inbound_handler(secret=..., on_received=save_to_db)
app.include_router(create_fastapi_router(handler), prefix="/webhooks/email")

No framework at all:

handler.to_wsgi_app()   # sync handler
handler.to_asgi_app()   # async handler
result = handler.handle(raw_bytes, signature_header)   # verify it yourself

on_sent / on_delivered / on_bounced are also available if you want one endpoint handling the outbound-side events too. Anything that isn't an email.* event is acknowledged with a 200 and ignored, so pointing a shared webhook URL here is safe.

Sending

from xlambda.email import EmailClient

client = EmailClient(api_key=os.environ["XLAMBDA_API_KEY"])

sent = client.emails.send(
    from_="you@yourdomain.com",
    to="them@example.com",          # or ["a@x.com", "b@y.com"]
    subject="Hello",
    html="<p>Hello</p>",
)
print(sent["id"])

from_ carries the trailing underscore because from is a Python keyword; it goes out on the wire as from, and comes back as from in responses. Everything else matches Resend's argument names (to/cc/bcc/reply_to/ subject/text/html), so switching from Resend — or back — is close to a find-and-replace on the import.

Async:

from xlambda.email import AsyncEmailClient

async with AsyncEmailClient(api_key=...) as client:
    await client.emails.send(from_=..., to=..., subject=..., html=...)

Domains

Sending from a domain requires registering it and publishing the DNS records that come back:

domain = client.domains.create(project_id="proj_123", domain="notify.example.com")

for record in domain["dkimRecords"]:
    print(record["name"], record["content"])
print(domain["mxTarget"], domain["spfRecord"], domain["dmarcRecord"])

client.domains.verify(domain["id"])     # re-checks MX/SPF/DKIM/DMARC

Message metadata (bodies are never stored — see above):

page = client.domains.messages(domain["id"], direction="received", limit=50)
page["nextCursor"]      # pass back as cursor= for the next page

SMTP credentials, for systems that speak SMTP rather than HTTP:

users = client.domains.sending_users(domain["id"])
user = users.create()   # user["password"] is shown here and never again

Errors

from xlambda.email import XlambdaError

try:
    client.emails.send(from_=..., to=..., subject=..., html=...)
except XlambdaError as err:
    if err.status == 429:
        ...     # rate limited, or the platform's outbound warm-up cap
    print(err.code, err.message, err.details)

XlambdaError and the webhook types are re-exported here, so you never need to depend on xlambda-core directly.

Conventions

Arguments are snake_case; responses are the API's own camelCase, returned as plain dicts typed by TypedDict. See xlambda-core for why.