Send email from FastAPI with Shipmail

Send email from FastAPI with AsyncShipmail, the async client in the Shipmail Python SDK. Send in a background task so the response does not wait, or await the send when the route needs the result.

Last updated November 1, 2026

Before you start

A verified sending subdomain, such as notify.example.com, with an active sender. See Transactional email.
An API key with send access to the sender's mailbox.
The sender address, such as hello@notify.example.com. Use the bare address. Shipmail adds the sender name.

Install the SDK

Add the Python SDK (Python 3.10 or later) and put the key and the sender address in the environment of your app.

pip install shipmail

SHIPMAIL_API_KEY=sm_live_...
EMAIL_FROM=hello@notify.example.com

Add a send_email helper

Create one client for the whole app. It keeps its HTTP connections open between sends.

# app/emails.py
import os

from shipmail import AsyncShipmail

shipmail = AsyncShipmail(os.environ["SHIPMAIL_API_KEY"])


async def send_email(to: str, subject: str, text: str, idempotency_key: str | None = None) -> None:
    await shipmail.messages.send(
        {"from": os.environ["EMAIL_FROM"], "to": [to], "subject": subject, "text": text},
        {"idempotency_key": idempotency_key} if idempotency_key else None,
    )

Send from a route

The lifespan function closes the client when the app shuts down. BackgroundTasks runs the welcome email after the response is sent. The receipt route awaits the send and answers 502 if Shipmail refuses it. Its idempotency key makes Shipmail send the receipt once, even when the SDK retries a 429 or a server error. See Idempotency.

# app/main.py
from contextlib import asynccontextmanager

from fastapi import BackgroundTasks, FastAPI, HTTPException
from pydantic import BaseModel
from shipmail import ShipmailError

from app.emails import send_email, shipmail


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await shipmail.aclose()


app = FastAPI(lifespan=lifespan)


class Customer(BaseModel):
    email: str


@app.post("/signup")
async def sign_up(customer: Customer, background_tasks: BackgroundTasks):
    # Create the account here.
    background_tasks.add_task(
        send_email, customer.email, "Welcome to Acme", "Thanks for signing up."
    )
    return {"ok": True}


@app.post("/orders/{order_id}/receipt", status_code=201)
async def send_receipt(order_id: str, customer: Customer):
    try:
        await send_email(
            customer.email,
            f"Your receipt for order {order_id}",
            "Thanks for your order.",
            idempotency_key=f"receipt-{order_id}",
        )
    except ShipmailError as error:
        raise HTTPException(status_code=502, detail="Email not sent") from error
    return {"ok": True}

An error in a background task does not reach the client, so log it there or use a job queue for email that must go out.

Test without sending

Use a sm_test_ key while you build. Shipmail accepts the request, delivers nothing and lists the send under Test in Activity. See Sandbox mode.

Receive replies

Replies to your sender address land in its Shipmail mailbox. To act on them in your app, add a webhook under Developer, then Webhooks, or with the Webhooks API, and add the secret to your environment.

Setting	Value
Endpoint URL	https://api.example.com/webhooks/shipmail
Events	message.received
Include reply text	On, so data.reply_text carries what the sender wrote.
Signing secret	Shown once after you save. Store it as SHIPMAIL_WEBHOOK_SECRET.

SHIPMAIL_WEBHOOK_SECRET=...

Shipmail can deliver the same event more than once. This table records each event_id once it is processed, so a retry of a handled event is skipped and a failed one is tried again:

# app/events.py
import sqlite3


# SQLite keeps the sample short. Use your app's database.
def _db() -> sqlite3.Connection:
    db = sqlite3.connect("events.db")
    db.execute("CREATE TABLE IF NOT EXISTS shipmail_events (event_id TEXT PRIMARY KEY)")
    return db


def was_handled(event_id: str) -> bool:
    with _db() as db:
        row = db.execute("SELECT 1 FROM shipmail_events WHERE event_id = ?", (event_id,))
        return row.fetchone() is not None


def mark_handled(event_id: str) -> None:
    # Call this only after the event was processed, so a failure gets a retry.
    with _db() as db:
        db.execute("INSERT OR IGNORE INTO shipmail_events (event_id) VALUES (?)", (event_id,))

Read the raw body with await request.body(), not a Pydantic model, and verify it before parsing. A bad signature or a timestamp older than five minutes gets a 401.
Answer with a 2xx within 10 seconds. Shipmail retries a failed delivery and disables the endpoint after five failures in a row. Handle the event in the request, not in a background task, so a failure returns an error and gets a retry. Hand slow work to a job queue.

# app/main.py (continued)
import os

from fastapi import Request, Response
from shipmail import WebhookVerificationError, verify_webhook

from app.events import mark_handled, was_handled


def handle_reply(data: dict) -> None:
    sender = (data.get("from") or {}).get("address")
    print("Reply from", sender, data.get("reply_text"))


@app.post("/webhooks/shipmail")
async def shipmail_webhook(request: Request):
    body = (await request.body()).decode()
    try:
        event = verify_webhook(body, dict(request.headers), os.environ["SHIPMAIL_WEBHOOK_SECRET"])
    except WebhookVerificationError:
        return Response("Invalid signature", status_code=401)

    # A retry carries the same event_id. Skip one that was already processed.
    if was_handled(event["event_id"]):
        return Response(status_code=204)
    if event["event_type"] == "message.received":
        # Handle it before answering: an exception returns 500 and Shipmail retries.
        handle_reply(event["data"])
    mark_handled(event["event_id"])
    return Response(status_code=204)

Signature details are in Webhook verification. To read the conversation and reply to it, see Receive email in an AI agent.