Automate OTP Flows in Google Colab with AI Agents

You're automating a signup flow. Everything is scripted — the form fill, the button click, the redirect. Then you hit a wall: an OTP arrives in an email, and your script stops cold.

Someone has to open a browser, read the code, and type it in. Or worse, you've built a brittle parser that breaks every time the email template changes. Or you're sharing a test inbox with five other engineers and can't tell which OTP is yours.

This is one of those problems that sounds simple and isn't. Let me show you a pattern that solves it cleanly.

Why Automating OTP Extraction Is Harder Than It Looks

The core issue is that email is asynchronous. You don't know when the message will arrive. You can't just time.sleep(5) and hope for the best — delivery can take anywhere from half a second to thirty seconds depending on the service, load, and your location.

Beyond timing, there's the parsing problem. OTP emails come in HTML. The six-digit code might be inside a <td>, wrapped in a <strong>, styled with inline CSS, and surrounded by marketing copy. Writing a regex against raw HTML is fragile and will break the moment the vendor's design team touches the template.

And then there's the shared inbox problem. If you're running parallel tests, multiple OTP emails land in the same inbox. You need to know which one belongs to which test run. Most setups don't have a clean answer to this.

A proper solution needs:

Isolated inboxes per test run — no cross-contamination
Reliable polling — wait until the email arrives, up to a configurable timeout
Plain-text access — skip the HTML parser entirely

The Pattern: Create, Wait, Extract

The approach is straightforward:

Create a fresh inbox before each flow
Register (or sign up) using that inbox's address
Poll until the OTP email arrives
Extract the code from the plain-text body

This is a pattern, not a hack. It works in pytest fixtures, Playwright scripts, n8n flows, LangChain agents, and raw Python scripts. The inbox is ephemeral — it gets created for the test and expires automatically.

uncorreotemporal.com provides this infrastructure: a real SMTP backend, isolated inboxes via REST API, and per-inbox session tokens so each test runs independently.

Full Example: OTP Extraction in Python

This is a complete, copy-paste example. No mocks, no shared state.

import requests
import re
import time

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

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

def wait_for_email(
    address: str,
    session_token: str,
    timeout: int = 60,
    poll_interval: int = 3
) -> dict:
    """Poll until at least one email arrives. Raises TimeoutError if none."""
    headers = {"Session-Token": session_token}
    deadline = time.monotonic() + timeout

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

        if messages:
            return messages[0]  # most recent first

        time.sleep(poll_interval)

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

def get_message_body(
    address: str,
    message_id: str,
    session_token: str
) -> str:
    """Fetch the full message and return the plain-text body."""
    headers = {"Session-Token": session_token}
    resp = requests.get(
        f"{BASE_URL}/api/v1/mailboxes/{address}/messages/{message_id}",
        headers=headers
    )
    resp.raise_for_status()
    data = resp.json()
    return data.get("body_text") or ""

def extract_otp(text: str, pattern: str = r'\b\d{6}\b') -> str | None:
    """Extract the first OTP match from text."""
    matches = re.findall(pattern, text)
    return matches[0] if matches else None

# -------------------------------------------------------
# Main flow
# -------------------------------------------------------
address, session_token = create_inbox(ttl_minutes=10)
print(f"[+] Inbox: {address}")

# Use this address in your actual signup/login flow
# e.g. playwright.fill("#email", address)
# e.g. requests.post("https://app.example.com/register", json={"email": address})
print(f"[*] Submitting signup with {address}...")

# Wait for the OTP email
try:
    summary = wait_for_email(address, session_token, timeout=60)
    print(f"[+] Email arrived: '{summary['subject']}'")
except TimeoutError as e:
    print(f"[-] {e}")
    raise

# Fetch full body and extract OTP
body = get_message_body(address, summary["id"], session_token)
otp = extract_otp(body)

if otp:
    print(f"[+] OTP: {otp}")
else:
    print("[-] OTP not found in message body")
    print(f"    Body preview: {body[:200]}")

The session_token is scoped to the inbox — it can only read messages from that specific address. Each test gets its own token, so parallel runs are fully isolated.

Handling Edge Cases

Timeout and retries

The 60-second default covers most services. For slower flows (welcome emails, queued notifications), increase it:

summary = wait_for_email(address, session_token, timeout=120)

If you want exponential backoff instead of fixed polling:

def wait_for_email_backoff(address, session_token, timeout=60):
    headers = {"Session-Token": session_token}
    deadline = time.monotonic() + timeout
    interval = 2

    while time.monotonic() < deadline:
        resp = requests.get(
            f"{BASE_URL}/api/v1/mailboxes/{address}/messages",
            headers={"Session-Token": session_token}
        )
        messages = resp.json()
        if messages:
            return messages[0]
        interval = min(interval * 1.5, 15)
        time.sleep(interval)

    raise TimeoutError(f"Timeout at {address}")

Multiple emails in the inbox

If the flow sends more than one email (e.g., a welcome email followed by the OTP), filter by subject:

messages = resp.json()
otp_email = next(
    (m for m in messages if "verification" in (m["subject"] or "").lower()),
    None
)

OTP pattern variations

Not every service uses 6-digit codes. Adjust the regex to match what you're working with:

# 4-digit PIN
extract_otp(body, pattern=r'\b\d{4}\b')

# 8-digit code
extract_otp(body, pattern=r'\b\d{8}\b')

# Alphanumeric token (e.g. "ABC-123")
extract_otp(body, pattern=r'\b[A-Z]{3}-\d{3}\b')

# UUID-style token
extract_otp(body, pattern=r'[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}')

HTML-only emails

Some services send HTML with no plain-text alternative. If body_text is empty, fall back to stripping HTML:

import html
import re

def strip_html(html_body: str) -> str:
    text = re.sub(r'<[^>]+>', ' ', html_body)
    return html.unescape(text)

body = data.get("body_text") or strip_html(data.get("body_html") or "")

Using This in Google Colab

Colab is useful here for quick experiments: no local setup, shareable notebooks, free GPU if you're running a model alongside the flow. The code above runs as-is in a Colab cell.

For AI agent workflows specifically — LangChain, CrewAI, LlamaIndex, or raw function-calling — the inbox creation and OTP extraction functions map directly to tool definitions. An agent can call create_inbox(), hand the address to a browser automation step, then call wait_for_email() and extract_otp() to complete the verification without human intervention.

The uncorreotemporal.com MCP server exposes the same capabilities as structured tools, so Claude Desktop and any MCP-compatible agent can call them natively without writing a single line of HTTP code.

Why This Works Reliably

Three things matter here:

Real SMTP, not mocks. The inbox receives email over actual SMTP from real sending services. There's no simulation layer that might behave differently from production.

Isolated inboxes. Each call to POST /api/v1/mailboxes creates a new address. Messages delivered to that address are only accessible via its session_token. Parallel test runs cannot interfere with each other.

Async delivery handled by the API. The polling loop is simple because the backend handles the async complexity — SES inbound webhook, message storage, indexed delivery time. You're polling a fast read query, not an external SMTP server.

Real Use Cases

CI/CD pipelines: Register a test account, verify the email, run authenticated tests, expire the inbox. Fully automated, zero manual steps.
Automated QA: Playwright or Selenium tests that complete full registration flows including email verification.
Account creation bots: Bulk testing of onboarding flows across multiple accounts in parallel, each with its own isolated inbox.
AI agents: Agents that complete multi-step signup flows autonomously — create inbox, fill form, extract OTP, submit verification, proceed.

The pattern is the same in all cases. The inbox is a throwaway resource. Create it, use it, let it expire.

Get Started

The API requires no upfront setup for anonymous inboxes — just POST to create an inbox and start receiving email. Free tier includes enough capacity for development and small test suites.

Try it now at uncorreotemporal.com. The full API reference is in the docs, and the MCP server config is available if you're building agent workflows in Claude Desktop or any MCP-compatible client.

OTP extraction doesn't have to be the flaky, manual step that breaks your automation. With isolated inboxes and a clean polling loop, it's just another function call.