Skip to content
Sign inGet startedDashboard

FrameworksFlask

Send email from Flask with Shipmail

Send email from Flask with Shipmail, the client in the Shipmail Python SDK. Call it from any view, and return an error response when Shipmail refuses the send.

Last updated

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.

# emails.py
import os

from shipmail import Shipmail

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


def send_email(to: str, subject: str, text: str, idempotency_key: str | None = None) -> None:
    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 view

The view answers 502 if Shipmail refuses the send. Its idempotency key makes Shipmail send the receipt once, even when the SDK retries a 429 or a server error. See Idempotency.

# app.py
from flask import Flask, request
from shipmail import ShipmailError

from emails import send_email

app = Flask(__name__)


@app.post("/orders/<order_id>/receipt")
def send_receipt(order_id: str):
    email = request.get_json()["email"]
    try:
        send_email(
            email,
            f"Your receipt for order {order_id}",
            "Thanks for your order.",
            idempotency_key=f"receipt-{order_id}",
        )
    except ShipmailError:
        app.logger.exception("Receipt email not sent")
        return {"error": "Email not sent"}, 502
    return {"ok": True}, 201

The send runs inside the request. To answer before the email goes out, hand send_email to your job queue instead.

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.

SettingValue
Endpoint URLhttps://api.example.com/webhooks/shipmail
Eventsmessage.received
Include reply textOn, so data.reply_text carries what the sender wrote.
Signing secretShown 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:

# 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 request.get_data(as_text=True), not request.get_json(), 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. Hand slow work to a job queue.
# app.py (continued)
import os

from shipmail import WebhookVerificationError, verify_webhook

from events import mark_handled, was_handled


def handle_reply(data: dict) -> None:
    sender = (data.get("from") or {}).get("address")
    app.logger.info("Reply from %s: %s", sender, data.get("reply_text"))


@app.post("/webhooks/shipmail")
def shipmail_webhook():
    body = request.get_data(as_text=True)
    try:
        event = verify_webhook(body, dict(request.headers), os.environ["SHIPMAIL_WEBHOOK_SECRET"])
    except WebhookVerificationError:
        return "Invalid signature", 401

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

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

Related guides