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. Tick Read messages and Create and delete inboxes (inbox:read + inbox:manage): a new key can only read, so opening an inbox fails with 403 until you add the second one. 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 UTC (ends with Z) | 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.
On the Free plan you can send to your own addresses (your account email or any of your inboxes). To send anywhere else, connect and verify a domain. Free includes one; Pro plans include more.
# 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.