Automating OTP Retrieval from Emails in Python

Every time a signup flow sends a verification email, an automation pipeline stalls — waiting for a human to open their inbox, copy a six-digit code, and paste it somewhere. For developers, this bottleneck shows up in:

• E2E test suites that exercise account registration flows
• CI/CD pipelines that provision test accounts on every build
• Signup automation scripts that need to complete email verification
• AI agents that create accounts autonomously and need to confirm them

The problem isn't just inconvenience. Real email addresses accumulate spam, shared inboxes cause test pollution, and Gmail's IMAP API has rate limits and OAuth friction that make programmatic access painful to maintain. You end up with flaky hacks — sleeping for five seconds and hoping the email arrived.

There's a better way.

---

The Problem with Traditional Email Testing

Using a real inbox for automated OTP retrieval introduces several failure modes:

Shared inbox pollution. If your test suite uses [e-mail], parallel test runs receive emails from multiple jobs into the same inbox, making it impossible to reliably match an OTP to a specific test run.

Gmail IMAP fragility. Gmail's IMAP access requires OAuth2 tokens, app passwords, or Less Secure App settings — all of which require manual setup, expire, and get blocked by workspace policies. Polling via IMAP also adds 5–30 seconds of latency.

Regex-on-raw-MIME hell. If you do get IMAP working, you're parsing RFC 2822 messages with base64 parts, quoted-printable encoding, and multipart boundaries. One encoding change in the sender's template breaks your parser.

Rate limits and delivery delays. Cloud email providers throttle connections, and deliverability adds unpredictable latency. A test that needs to run in under 30 seconds becomes a 90-second flake.

---

A Better Approach: Programmable Temporary Email

A programmable temporary email infrastructure solves all of this by giving you:

• On-demand inbox creation via a REST API — one per test, discarded afterwards
• Real SMTP reception — actual email protocols, not simulation
• Instant message retrieval — read the body milliseconds after delivery
• No authentication overhead — anonymous inboxes with a session token, or API key access for CI

UnCorreoTemporal is exactly this kind of infrastructure. It provides temporary email addresses at @uncorreotemporal.com, accepts real SMTP delivery, stores messages in a PostgreSQL database, and exposes them through a REST API and a WebSocket stream. You get a full programmable email inbox in a single HTTP call.

---

Architecture Overview

Understanding how the system works helps you use it correctly in production automation.

External SMTP sender
        │
        ▼
AWS SES (spam/virus filter)
        │
        ▼ SNS webhook
POST /api/v1/ses/inbound
        │
        ▼
core/delivery.deliver_raw_email()
        │
   ┌────┴────┐
   ▼         ▼
PostgreSQL  Redis pub/sub
(messages)  (mailbox:{address})
                │
                ▼
        WebSocket /ws/inbox/{address}
        {"event": "new_message", "message_id": "..."}

The deliver_raw_email() function in core/delivery.py is the shared ingestion point — used by both the AWS SES webhook (api/routers/ses_inbound.py) and the local aiosmtpd-based SMTP server (smtp/handler.py). Once an email is stored, it publishes a Redis event on the channel mailbox:{address}, which the WebSocket endpoint in ws/inbox.py forwards to any connected clients.

Messages are stored with parsed body_text and body_html fields extracted via Python's standard email library, plus the full RFC 2822 bytes for re-parsing. You work directly with decoded strings — no MIME handling on your end.

---

Creating a Temporary Inbox

The simplest call creates an anonymous inbox with no authentication required. The response includes a session_token that authenticates all subsequent operations on that mailbox.

import requests

BASE_URL = "https://api.uncorreotemporal.com/api/v1"

def create_inbox(ttl_minutes: int = 10) -> tuple[str, str]:
    resp = requests.post(
        f"{BASE_URL}/mailboxes",
        params={"ttl_minutes": ttl_minutes},
    )
    resp.raise_for_status()
    data = resp.json()
    return data["address"], data["session_token"]

address, token = create_inbox(ttl_minutes=10)
print(address)  # [e-mail]

For CI/CD where you have an API key, use Bearer token auth instead:

headers = {"Authorization": "Bearer uct_your_api_key_here"}
resp = requests.post(f"{BASE_URL}/mailboxes", headers=headers)

The address format ([e-mail]) is human-readable and collision-resistant. Inboxes expire automatically at expires_at — no cleanup code required.

---

Waiting for the Email

After triggering your signup flow, poll the messages endpoint until an email arrives.

import time

def wait_for_email(
    address: str,
    session_token: str,
    timeout: int = 30,
    poll_interval: float = 1.5,
) -> dict:
    headers = {"X-Session-Token": session_token}
    deadline = time.time() + timeout

    while time.time() < deadline:
        resp = requests.get(
            f"{BASE_URL}/mailboxes/{address}/messages",
            headers=headers,
        )
        resp.raise_for_status()
        messages = resp.json()

        if messages:
            msg_id = messages[0]["id"]
            full = requests.get(
                f"{BASE_URL}/mailboxes/{address}/messages/{msg_id}",
                headers=headers,
            )
            full.raise_for_status()
            return full.json()

        time.sleep(poll_interval)

    raise TimeoutError(f"No email received within {timeout}s")

Real-time alternative: Connect to the WebSocket endpoint for push-based delivery notification:

import asyncio, json
import websockets

async def wait_for_email_ws(address: str, session_token: str) -> str:
    url = f"wss://api.uncorreotemporal.com/ws/inbox/{address}?token={session_token}"
    async with websockets.connect(url) as ws:
        async for raw in ws:
            event = json.loads(raw)
            if event.get("event") == "new_message":
                return event["message_id"]

---

Extracting the OTP Code

OTP codes in verification emails are almost always 6 or 8 isolated digits. A targeted regex on body_text is reliable and fast:

import re

def extract_otp(body: str) -> str:
    match = re.search(r"(?<!\d)(\d{6})(?!\d)", body)
    if not match:
        raise ValueError("No 6-digit OTP found in message body")
    return match.group(1)

The body_text field comes pre-decoded — core/parser.py handles multipart MIME, quoted-printable, and base64 internally.

---

Full Python Example

import re
import time
import requests

BASE_URL = "https://api.uncorreotemporal.com/api/v1"


def create_inbox(ttl_minutes: int = 10) -> tuple[str, str]:
    resp = requests.post(f"{BASE_URL}/mailboxes", params={"ttl_minutes": ttl_minutes})
    resp.raise_for_status()
    data = resp.json()
    return data["address"], data["session_token"]


def trigger_signup(email: str) -> None:
    requests.post("https://yourapp.com/api/signup", json={"email": email})


def wait_for_message(address: str, token: str, timeout: int = 30) -> dict:
    headers = {"X-Session-Token": token}
    deadline = time.time() + timeout
    while time.time() < deadline:
        resp = requests.get(f"{BASE_URL}/mailboxes/{address}/messages", headers=headers)
        resp.raise_for_status()
        messages = resp.json()
        if messages:
            msg_id = messages[0]["id"]
            full = requests.get(f"{BASE_URL}/mailboxes/{address}/messages/{msg_id}", headers=headers)
            full.raise_for_status()
            return full.json()
        time.sleep(1.5)
    raise TimeoutError("Email not received in time")


def extract_otp(body: str) -> str:
    match = re.search(r"(?<!\d)(\d{6})(?!\d)", body)
    if not match:
        raise ValueError("No OTP found")
    return match.group(1)


if __name__ == "__main__":
    address, token = create_inbox(ttl_minutes=5)
    print(f"Inbox: {address}")
    trigger_signup(email=address)
    message = wait_for_message(address, token, timeout=30)
    otp = extract_otp(message["body_text"])
    print(f"OTP: {otp}")

---

Using This in Automated Testing

pytest fixture

import pytest, requests

BASE_URL = "https://api.uncorreotemporal.com/api/v1"

@pytest.fixture()
def temp_inbox():
    resp = requests.post(f"{BASE_URL}/mailboxes", params={"ttl_minutes": 10})
    resp.raise_for_status()
    data = resp.json()
    yield data["address"], data["session_token"]


def test_signup_sends_otp(temp_inbox, page):
    address, token = temp_inbox
    page.goto("https://yourapp.com/signup")
    page.fill('[name="email"]', address)
    page.click('[type="submit"]')
    message = wait_for_message(address, token, timeout=30)
    otp = extract_otp(message["body_text"])
    page.fill('[name="otp"]', otp)
    page.click('[type="submit"]')
    page.wait_for_url("**/dashboard")

CI/CD (GitHub Actions)

- name: Run E2E tests
  env:
    UCT_API_KEY: ${{ secrets.UCT_API_KEY }}
  run: pytest tests/e2e/ -v

---

Bonus: Using This with AI Agents

UnCorreoTemporal ships an MCP server that exposes inbox management directly as tools callable by Claude Desktop and other MCP-compatible agents:

{
  "mcpServers": {
    "uncorreotemporal": {
      "command": "python",
      "args": ["-m", "uncorreotemporal.mcp.server"],
      "env": { "UCT_API_KEY": "uct_your_key" }
    }
  }
}

With this configured, an agent can create a mailbox, complete a registration flow, retrieve the verification email, extract the OTP, and proceed — all without human intervention.

---

Conclusion

Automated OTP retrieval from email doesn't have to be fragile. The combination of on-demand inbox creation, real SMTP delivery, and a clean REST API removes every layer of indirection that makes traditional email testing painful.

UnCorreoTemporal is built for this use case — async FastAPI, PostgreSQL, Redis-driven WebSocket push, and AWS SES. The API surface is small enough to integrate in an afternoon.

Try the API at uncorreotemporal.com. Your first inbox is one HTTP call away.

Originally published at uncorreotemporal.com