Receive email with Python

This guide shows how to receive email with Python. api.email accepts mail for your domain, parses it and sends each message to your app as a signed JSON webhook, so there is no IMAP polling and no MIME parsing in your code. You will add an MX record and a route, write a Flask handler that verifies signatures, see the FastAPI version, turn replies into support tickets and download attachments.

How inbound email works

Mail for your receiving domain is delivered to mx.api.email. api.email checks the sender’s SPF, DKIM and DMARC results, splits the message into text, HTML and attachments, removes the quoted thread from replies and then POSTs an email.received event to the URL your route points at. Your handler verifies the signature, answers with a 2xx status and processes the message. If your app is unreachable, delivery is retried with backoff for up to 24 hours.

Pick addresses that describe the job they do. A shared address such as [email protected] or [email protected] suits a team workflow, while per-object addresses such as [email protected] let you route replies without parsing subjects. You can mix both on one domain with separate routes, each forwarding to the handler that owns that workflow.

Add an MX record and a route

Use a subdomain that receives nothing else, such as in.acme.com, so your existing mailboxes are untouched:

in.acme.com.   MX   10   mx.api.email.

Then create a route. A catch-all lets you encode ids in the address, which is handy for replies to tickets or orders:

import os
import requests

resp = requests.post(
    "https://api.email/v1/routes",
    headers={"Authorization": f"Bearer {os.environ['API_EMAIL_KEY']}"},
    json={"match": "*@in.acme.com", "forward": "https://acme.com/hooks/inbound"},
    timeout=10,
)
print(resp.json())  # {'id': 'rt_3k9d', 'secret': 'whsec_…'}

Store the route’s secret as API_EMAIL_ROUTE_SECRET: every delivery from this route is signed with it. The receiving documentation lists every route option.

A Flask handler with signature verification

Verify the api-email-signature header against the raw body before you parse it. The header has a timestamp t and an HMAC-SHA256 signature v1 over t + "." + body, keyed with the route’s secret; right after a secret rotation it carries two v1 values.

# webhooks.py
import hashlib, hmac, json, os, re, time
from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["API_EMAIL_ROUTE_SECRET"]

def verify(raw: bytes, header: str | None, secret: str, tolerance: int = 300) -> bool:
    # Same helper as in the webhooks docs: rejects replays and forged or malformed
    # signatures, and accepts both signatures during a secret rotation.
    t, sigs = None, []
    for part in (header or "").split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isdigit():
            t = int(value)
        elif key == "v1" and re.fullmatch(r"[0-9a-f]{64}", value):
            sigs.append(value)
    if t is None or not sigs or abs(time.time() - t) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, s) for s in sigs)

@app.post("/hooks/inbound")
def inbound():
    raw = request.get_data()
    if not verify(raw, request.headers.get("api-email-signature"), SECRET):
        abort(400)
    try:
        enqueue("inbound-email", json.loads(raw))  # hand off, then acknowledge
    except Exception:
        abort(503)  # api.email retries the delivery, so nothing is lost
    return "", 200

Acknowledge only once the event is safely in your job queue, such as Celery or RQ. The webhook should only verify, hand off and answer.

The same handler in FastAPI

import json
from fastapi import FastAPI, HTTPException, Request

from webhooks import SECRET, verify  # the helper from the Flask example

api = FastAPI()

@api.post("/hooks/inbound")
async def inbound(request: Request):
    raw = await request.body()
    if not verify(raw, request.headers.get("api-email-signature"), SECRET):
        raise HTTPException(status_code=400)
    await queue.enqueue("inbound-email", json.loads(raw))
    return {}

Read await request.body() rather than a parsed model, because the signature covers the exact bytes that were sent. If enqueue raises, FastAPI answers with an error and api.email retries the delivery.

What the payload contains

The fields you will use most are:

• from and to: the sender and the address on your domain that received the message.
• subject, text and html: the full message.
• reply_text: only the new part of a reply, without the quoted thread or the signature.
• in_reply_to: the id of the email you sent, such as em_8f2c1a, when the message is a reply to it.
• attachments: a list with filename, content_type, size and a download url.
• auth: the SPF, DKIM and DMARC results for the sender’s domain.
• id and received_at: a unique event id and the time the message arrived.

Turn replies into tickets

Most inbound processing is about replies: a customer answers a notification and the answer should land in the right ticket. in_reply_to points to the email you sent. For reply addresses that carry your own id, sign the id: a bare sequential address such as ticket-482@… would let anyone post into anyone’s ticket.

import hashlib, hmac, os, re

TOKEN_SECRET = os.environ["REPLY_TOKEN_SECRET"].encode()

def sign(ticket_id: int) -> str:
    return hmac.new(TOKEN_SECRET, str(ticket_id).encode(), hashlib.sha256).hexdigest()[:16]

def reply_address(ticket_id: int) -> str:
    return f"ticket-{ticket_id}.{sign(ticket_id)}@in.acme.com"  # use as From or Reply-To

def ticket_id_from(to: str) -> int | None:
    m = re.fullmatch(r"ticket-(\d+)\.([0-9a-f]{16})@in\.acme\.com", to.lower())
    return int(m.group(1)) if m and hmac.compare_digest(m.group(2), sign(int(m.group(1)))) else None

def handle_inbound(event: dict) -> None:
    if Event.objects.filter(inbound_id=event["id"]).exists():
        return  # already processed: retries and replays are safe
    Event.objects.create(inbound_id=event["id"])

    ticket = Ticket.objects.filter(sent_email_id=event.get("in_reply_to")).first()
    if ticket is None and (tid := ticket_id_from(event["to"])):
        ticket = Ticket.objects.filter(pk=tid).first()
    if ticket is None:
        ticket = Ticket.objects.create(customer_email=event["from"], subject=event["subject"])

    trusted = event["auth"]["dmarc"] == "pass" or event["from"] == ticket.customer_email
    ticket.messages.create(author=event["from"], body=event["reply_text"] or event["text"], needs_review=not trusted)

The first lines make processing idempotent, and the trusted flag keeps messages from senders who fail DMARC and don’t match the ticket out of the customer-visible thread until someone reviews them.

Download attachments

Attachments arrive as URLs rather than inline data, which keeps webhook bodies small. Download each file with your API key and store it in your own bucket.

import uuid
from pathlib import PurePath

for file in event["attachments"]:
    resp = requests.get(
        file["url"],
        headers={"Authorization": f"Bearer {os.environ['API_EMAIL_KEY']}"},
        timeout=30,
    )
    resp.raise_for_status()  # the job retries on failure
    safe_name = re.sub(r"[^\w.-]", "_", PurePath(file["filename"]).name)[:100]  # sender-controlled
    storage.save(f"{event['id']}/{uuid.uuid4()}-{safe_name}", resp.content)

Check content_type and size before processing, and treat every attachment as untrusted input: scan it if your workflow opens files automatically.

Skip auto-replies and loops

Out-of-office messages, bounces and other systems’ notifications will reach your inbound address sooner or later. Before you create a ticket or send an answer, look at the headers in event["headers"]: ignore messages where auto-submitted is present and not no, messages with precedence set to bulk or auto_reply, and anything sent from your own domain. Without these checks, an auto-reply from a customer’s mail server can trigger your own confirmation email, which triggers another auto-reply, and so on until someone notices.

Answer the sender

To reply, send a normal email with POST /v1/emails to the address in from. Keep the subject, add the original message-id to the In-Reply-To and References headers so the conversation stays threaded in the customer’s mail client, and send it from a reply address on your receiving domain. Store the id the API returns: when the customer answers again, in_reply_to points straight back to it.

Security notes

Treat every inbound message as untrusted input. Escape html before displaying it, or better, display text. Never follow links or open attachments automatically without scanning them. Use the auth results to decide how much to trust the sender, and keep your webhook secret in the environment, rotating it from the dashboard if it ever leaks.

Test it end to end

Expose your local server with a tunnel, for example cloudflared tunnel --url http://localhost:5000, point a route at the tunnel URL and send yourself an email. The dashboard shows each delivery attempt with the status code your handler returned, and you can replay an event after fixing a bug instead of sending a new email every time.