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 qa@yourcompany.com, 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)  # mango-panda-42@uncorreotemporal.com

For CI/CD where you have an API key, use Bearer token auth instead — this lets you manage multiple mailboxes and get higher TTL limits depending on your plan:

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

The address format (adjective-noun-number@uncorreotemporal.com) 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. The list endpoint returns metadata only; fetching a single message returns the full body and marks it as read.

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: If you prefer push over poll, connect to the WebSocket endpoint. The server emits {"event": "new_message", "message_id": "<uuid>"} the instant delivery completes — driven by Redis pub/sub — with a keepalive ping every 30 seconds.

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 6-digit codes not adjacent to other digits
    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 from the API — the core/parser.py module handles multipart MIME, quoted-printable, and base64 internally. You apply your regex directly to a plain Python string.

For other OTP formats:

# 6 or 8 digit codes
re.search(r"(?<!\d)(\d{6}|\d{8})(?!\d)", body)

# Alphanumeric codes like "A3F9K2"
re.search(r"\b([A-Z0-9]{6,8})\b", body)

---

Full Python Example

Here's a complete, self-contained script for automated OTP retrieval:

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:
    """Replace with your application's signup endpoint."""
    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 in email body")
    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

Wrap inbox creation in a session-scoped or function-scoped fixture. Each test gets its own isolated inbox — no shared state, no cross-test contamination:

import pytest, requests, re, time

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"]
    # Inbox expires automatically — no teardown needed


def test_signup_sends_otp(temp_inbox, page):  # page = Playwright fixture
    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")

Selenium

from selenium.webdriver.common.by import By

def test_otp_flow(driver, temp_inbox):
    address, token = temp_inbox
    driver.get("https://yourapp.com/signup")
    driver.find_element(By.NAME, "email").send_keys(address)
    driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()

    message = wait_for_message(address, token)
    otp = extract_otp(message["body_text"])

    driver.find_element(By.NAME, "otp").send_keys(otp)
    driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()

CI/CD (GitHub Actions)

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

Store your API key as a repository secret. Pass it as a Bearer token in your Python code. Each job gets isolated inboxes, and parallel matrix jobs never collide.

---

Bonus: Using This with AI Agents

LLM-powered agents increasingly need to perform actions that require email verification — creating accounts on behalf of users, confirming subscriptions, or completing onboarding flows autonomously.

UnCorreoTemporal ships a Model Context Protocol (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 call create_mailbox(), use the address during registration, call get_messages(address) to check for the verification email, then read_message(address, message_id) to retrieve the body and extract the OTP — all without human intervention. For agents built with LangChain, CrewAI, or custom tool loops, the same REST API works directly via requests using the same pattern shown above.

---

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: no shared inboxes, no IMAP credentials, no MIME parsing, no unpredictable delivery delays.

UnCorreoTemporal is built specifically for this use case — async FastAPI backend, PostgreSQL message storage, Redis-driven WebSocket push, and AWS SES for production email ingestion. The API surface is small enough to integrate in an afternoon, and the architecture scales cleanly from a single pytest fixture to a full multi-agent CI pipeline.

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