Send email with FastAPI
Prerequisites
Before you start, you need:
- A mails.ai API key. A test key (
mk_test_…) runs everything on this page and sends no real email. - Python 3.10 or newer.
Guide
Install
Get FastAPI and the mails.ai Python SDK.
pip install "fastapi[standard]" mailsaiSet your API key
Put the key in an environment variable. The SDK reads
MAILS_API_KEYon its own, so the key never appears in your code.export MAILS_API_KEY="mk_test_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"Send from a route
fromnames the agent that sends the message: an agent with that name is used if it exists, and created if it does not when the key may create agents (themanagescope) and your plan has room for another; withoutmanagethe send answers403 insufficient_scope. Becausefromis a reserved word in Python, the fields go in a dictionary.from fastapi import FastAPI, HTTPException from mailsai import Client, MailsError app = FastAPI() @app.post("/send") def send(): message = { "from": "hello", "to": "reply@test.mails.ai", "subject": "Welcome aboard", "body": "Thanks for signing up. Reply any time and I will read it.", } try: client = Client() # reads MAILS_API_KEY sent = client.send(**message) except MailsError as error: detail = {"code": error.code, "message": error.message} raise HTTPException(status_code=error.status or 500, detail=detail) return {"id": sent["id"], "status": sent["status"], "from": sent["from"]}Start the app and call the route from a second terminal:
fastapi dev main.py curl -X POST http://127.0.0.1:8000/sendreply@test.mails.aiis an address of ours that answers within about a second, so there is already a reply waiting for the next step. How the test address works.Receive replies on a webhook
mails.ai signs every event it posts to your endpoint with an
X-Mails-Signatureheader.verify_webhookchecks that header against the exact bytes that arrived and returns the event, orNonewhen the check fails. For 24 hours after you rotate the signing secret the header carries twov1signatures, soverify_eventchecks each one. Read the body withrequest.body(): a parsed and re-serialized body no longer matches the signature. Add this tomain.py:import os from fastapi import Request from mailsai import verify_webhook def verify_event(body: str, header: str, secret: str): """The event when any v1 signature in the header matches the secret, else None.""" parts = header.split(",") t = next((p for p in parts if p.startswith("t=")), "") for v1 in (p for p in parts if p.startswith("v1=")): event = verify_webhook(body, f"{t},{v1}", secret) if event is not None: return event return None @app.post("/webhooks/mails") async def mails_webhook(request: Request): body = (await request.body()).decode() signature = request.headers.get("x-mails-signature", "") event = verify_event(body, signature, os.environ["MAILS_WEBHOOK_SECRET"]) if event is None: raise HTTPException(status_code=400, detail="bad signature") if event["type"] == "reply.received" and not event.get("quarantined"): reply = event["data"] print(reply["subject"], reply["body_text_excerpt"], event.get("injection_score")) # Hand the reply to your agent here. return {"received": True}quarantinedistruewhen the message scored as a likely prompt injection. Skip those, and answer with a2xxwithin 5 seconds: any other answer, or none, counts as a failed delivery and is tried again, 3 attempts in all.Register the webhook
Deploy the app, then tell mails.ai where the route lives. The address must be
https://and reachable from the internet. This call needs a key with themanagescope.curl https://api.mails.ai/v1/webhooks \ -H "Authorization: Bearer $MAILS_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "url": "https://your-app.com/webhooks/mails", "event_types": ["reply.received", "message.received"] }'The answer carries a
signing_secretthat starts withwhsec_and is shown once. Set it asMAILS_WEBHOOK_SECRETwhere the app runs.
client.list_events(event_type="reply.received") returns the same events, and Events & streaming shows how to keep a live connection open.Next steps
Was this page helpful?