Email API

/

Python

Self-Hosted Email API in Python

This guide shows you how to send email from Python using a self-hosted email API. Install requests or httpx, point them at your Cloudflare Worker endpoint, and go live — sync or async, with webhooks for inbound messages. No SMTP relay, no per-email bill, no SDK lock-in.

Install + auth

Two packages cover the same wire contract — requests for sync code, httpx for asyncio. Pick whichever your service already uses. The API key comes from /dashboard/settings/api-keys and is bound to one of your domains.

requirements.txtbash
# Sync client (requests)
pip install requests

# Async client (httpx) — same wire contract, non-blocking
pip install httpx

Send with requests (sync)

POST /api/email/send with a JSON body. The from address must land on a domain bound to the API key — the Worker rejects cross-domain sends with a 422 from_domain_mismatch.

send.pypython
import os
import requests

API_KEY = os.environ["FLOWMAILS_API_KEY"]   # fm_live_…
ENDPOINT = "https://your-worker.workers.dev/api/email/send"

resp = requests.post(
    ENDPOINT,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    },
    json={
        "from": "hello@yourdomain.com",
        "to": "customer@example.com",
        "subject": "Order #1234 confirmed",
        "text": "Thanks for your order — we will ship it tomorrow.",
    },
    timeout=15,
)
resp.raise_for_status()
print(resp.json())
# {"results": [{"id": "msg_8421", "status": "queued"}]}

Send async with httpx

Same wire contract, async I/O — drop this into a FastAPI handler, a Celery worker, or any asyncio app. The to field accepts a single address or a list.

send_async.pypython
import asyncio
import httpx

async def send():
    async with httpx.AsyncClient(timeout=15) as client:
        r = await client.post(
            "https://your-worker.workers.dev/api/email/send",
            headers={
                "Authorization": f"Bearer {os.environ['FLOWMAILS_API_KEY']}",
                "Content-Type": "application/json",
            },
            json={
                "from": "support@yourdomain.com",
                "to": ["customer@example.com", "team@yourdomain.com"],
                "subject": "Weekly digest",
                "html": "<h1>This week</h1><p>Three new signups.</p>",
            },
        )
        r.raise_for_status()
        return r.json()

asyncio.run(send())

Receive inbound webhooks

The Worker forwards the JSON-shaped envelope to whatever URL you register. Filter spam, persist to your DB, fan out to your CRM — the Worker has done the message-id / threading bookkeeping for you.

webhook.pypython
# FastAPI webhook receiver — drop into your service
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.post("/webhooks/inbound")
async def inbound(request: Request):
    # Cloudflare Email Routing posts the raw RFC 5322 envelope;
    # the Worker already extracted the JSON-shaped payload.
    payload = await request.json()
    if payload.get("subject") and "spam" in payload["subject"].lower():
        raise HTTPException(status_code=202, detail="filtered")
    # … persist to your DB, fan out to CRM, etc.
    return {"ok": True}

Send your first email from Python in under five minutes.

Start free