Email testing with Python
The official Python client wraps the REST API in two calls: open an inbox, wait for the code. Works in pytest, unittest, Robot Framework or a plain script.
Before you begin
- A MailFlat accountFree to start, no card. Every plan can open inboxes from the API.
- An account API keyAgents → API keys in the dashboard. It looks like mf_live_… and goes in the X-API-Key header. Keep it in an environment variable, never in the repo. See API keys and authentication.
- Python 3.9+That is the whole toolchain requirement.
Install
pip install mailflat
Package: mailflat
Your first inbox and one-time code
The whole loop: open an address, let your app mail it, read the code back, clean up.
import os
from mailflat import MailFlat
mf = MailFlat(api_key=os.environ["MAILFLAT_API_KEY"])
# 1. a real, deliverable address
# prefix = the local part of the address. label is a display name only.
inbox = mf.create(prefix="signup")
print(inbox.address) # signup@a7f2c.mailflat.net
# 2. your app sends the code to that address
my_app.register(email=inbox.address)
# 3. read it back, no mocking anywhere
otp = inbox.wait_for_otp(timeout=30)
print(otp) # "482913"
# 4. done with it
inbox.delete()
Line by linewhat each step of this Python example does
- create(...)
- Returns immediately with a real, deliverable address. Nothing is queued or simulated.
- retention_hours
- Optional. Messages purge themselves after it, so a skipped teardown never leaks.
- wait_for_otp
- Polls for you and fails loudly on timeout, instead of returning nothing three lines before the real error.
- delete()
- Optional but tidy. Retention would clean up anyway.
The address is real
Mail actually travels: SMTP, DKIM, the lot. Nothing is stubbed, so a broken template or a misconfigured sender fails here exactly like it would in production.
Waiting for the code
We extract the one-time code server-side and hand it to you as a field, so you never write a regex against an email body.
# wait_for_otp polls /latest for you and raises instead of returning None
otp = inbox.wait_for_otp(timeout=30, poll_interval=1)
# need the whole message, not just the code?
msg = inbox.wait_for_message(timeout=30)
print(msg.subject, msg.sender, msg.text)
# both raise on timeout, so a missing email fails the test loudly
from mailflat import OTPTimeoutError
try:
inbox.wait_for_otp(timeout=5)
except OTPTimeoutError as e:
print("no code arrived:", e)
# Waiting for MANY codes at once? Use the async client. Twenty inboxes wait in
# one event loop instead of twenty parked threads.
import asyncio
from mailflat.aio import AsyncMailFlat
async def main():
async with AsyncMailFlat() as mf:
inboxes = await asyncio.gather(*(mf.create(label=f"run-{i}") for i in range(20)))
codes = await asyncio.gather(*(i.wait_for_otp(timeout=60) for i in inboxes))
print(codes)
asyncio.run(main())
Always set a timeout
A poll loop without a deadline turns a missing email into a hung job. Fail loudly instead: the error message should name the address you were waiting on.
Reading every message
for msg in inbox.messages(): # received mail, newest first
print(msg.subject, msg.sender, msg.received_at)
print(msg.text) # plain text body
print(msg.links) # URLs found in the body
print(msg.otp) # extracted code, or None
# mail YOU sent from this address is not in the list above. Ask for it:
sent = inbox.messages(direction="out")
latest = inbox.latest() # None until the first message lands
if latest and "Reset" in latest.subject:
latest.delete() # drop one message, keep the inbox
| Field | Type | What it is | JSON field (msg.raw) |
|---|---|---|---|
| subject | string | Subject line | subject |
| sender | string | From address | sender |
| text | string | Plain text body | body_text |
| html | string | HTML body | body_html |
| otp | string | null | One-time code, extracted by us | otp_code |
| to_address | string | The exact address it was sent to, tag included | to_address |
| tag | string | null | Plus-addressing tag, if the sender used one | tag |
| received_at | ISO 8601 | When it landed | received_at |
| is_encrypted | boolean | True on end-to-end encrypted inboxes, where body and code are unavailable | is_encrypted |
Sending mail from the inbox
Useful in reverse: point your own inbound pipeline at a MailFlat address and check that it handles what arrives.
# mail leaves from the inbox address, DKIM-signed by our own MTA
result = inbox.send(
"someone@example.com",
subject="Welcome",
body="Plain text body",
html="<p>Optional HTML body</p>",
cc=["team@example.com"], # written into the headers
bcc=["archive@example.com"], # in NO header, not even in its own copy
attachments=["invoice.pdf"], # a path: read and base64-encoded for you
)
# send() returns when the mail is ACCEPTED (HTTP 202), not when it is delivered:
# delivery runs on a queue. The response carries the id you follow it with.
print(result["message_id"], result["queued"])
from mailflat import SendFailedError, SendTimeoutError
try:
sent = inbox.wait_until_sent(result["message_id"], timeout=120)
print(sent.send_status) # "sent" (or "unsigned" if DKIM was skipped)
except SendFailedError as e:
print("permanently failed:", e) # the queue gave up; it will not retry
except SendTimeoutError:
# NOT a failure. The queue is still retrying. Sending again delivers it twice.
print("still queued; check back or subscribe to the message.delivered webhook")
# attachments also take bytes you already hold
inbox.send("someone@example.com", attachments=[
{"filename": "report.csv", "content": b"a,b\n1,2\n"},
])
# Attachment size and count depend on your plan. Free is deliberately small.
# Ask the API rather than hard-coding the numbers:
# GET https://mailflat.net/api/plans → max_attachment_bytes · max_attachments
# Going over is rejected with the limit spelled out, not silently dropped.
Cleaning up
inbox.delete() # inbox and every message, immediately
# or leave it: messages expire on their own at the retention you asked for
inbox = mf.create(label="ci", retention_hours=2)
Two safety nets, use both
Delete in teardown so the list stays readable, and set retention_hours so a crashed run still cleans itself up.
In a test suite
Works with pytest, unittest, Robot Framework, Selenium and anything else that gives you a setup and teardown hook.
# conftest.py: one isolated inbox per test, torn down even on failure
import os, uuid, pytest
from mailflat import MailFlat
@pytest.fixture(scope="session")
def mf():
return MailFlat(api_key=os.environ["MAILFLAT_API_KEY"])
@pytest.fixture
def inbox(mf):
box = mf.create(prefix=f"ci-{uuid.uuid4().hex[:8]}", retention_hours=2)
yield box
box.delete()
# test_signup.py
def test_user_can_sign_up(inbox, page):
page.goto("/signup")
page.fill("#email", inbox.address)
page.click("#submit")
page.fill("#code", inbox.wait_for_otp(timeout=60))
assert page.url.endswith("/dashboard")
Line by linewhat each step of this Python example does
- unique prefix
- One inbox per test, so parallel workers never read each other's mail.
- teardown hook
- Runs even when the test fails. That is exactly when inboxes get left behind.
- real code, real email
- No test-mode backdoor in your app: the path under test is the one your users take.
Python reference
| Method | What it does |
|---|---|
| mf.create(prefix=…, label=…, subdomain=…, domain=…, retention_hours=…) | Open an inbox, returns Inbox |
| mf.list() | Every inbox this key opened |
| mf.inbox(address) | Attach to an existing address without an API call |
| inbox.wait_for_otp(timeout=…, poll_interval=…) | Poll until a code arrives, raises OTPTimeoutError |
| inbox.wait_for_message(timeout=…) | Same, but returns the whole Message |
| inbox.messages() / inbox.latest() | All messages (newest first) / the newest one or None |
| inbox.send(to, subject=…, body=…, html=…) | Send from this address, DKIM-signed |
| inbox.delete() / inbox.delete_message(id) | Drop the inbox / one message |
Full endpoint reference, including error shapes and rate limits: Agent API and MCP.