DEV Community

Francisco Perez
Francisco Perez

Posted on • Originally published at uncorreotemporal.com

Give Your AI Agent a Real Email Inbox in Google Colab

#ai #python #automation #tutorial

The Gap Nobody Talks About

You can run a Python agent in Google Colab. It can call APIs, write code, scrape pages, parse documents, and call LLMs. But the moment that agent needs to complete a real-world account registration — the kind that sends a verification email — it stops cold.

No inbox. No way to receive an OTP. The flow is blocked.

This is not a minor inconvenience. A large fraction of real-world automation depends on email as a control point:

  • Account signups with email confirmation
  • Two-factor authentication via OTP
  • Subscription confirmations and approval flows
  • Password resets and verification links

The agent can get to the "Check your email" page. It just can't do anything after that.

This article shows how to fix that. You'll have a working, programmable email inbox running in Google Colab by the end — one the agent controls entirely, no human in the loop.

Why Traditional Solutions Fail

The obvious workarounds don't hold up in practice.

Mocking the email step is the most common approach. You bypass the verification entirely in your test environment. The problem: the agent doesn't learn to handle the real flow, and when you need production-level automation (provisioning accounts on external services, for example), the mock is useless.

Using a shared inbox like qa-team@yourdomain.com introduces race conditions the moment anything runs in parallel. Two agent runs trigger two verification emails into the same inbox. You can't reliably match which code belongs to which session.

Gmail IMAP is fragile at a level that makes it unsuitable for automated use. OAuth2 tokens expire, app passwords get blocked by workspace policies, IMAP polling adds unpredictable latency, and you're parsing raw RFC 2822 MIME messages — base64-encoded, quoted-printable — to get to the text body. It breaks every few months.

Manual verification defeats the purpose. Agents that pause for human input are not agents; they're wizards.

What Agents Actually Need

An agent that can handle email needs four things:

  1. An email address it controls, created on demand
  2. Real SMTP delivery — not simulated, not mocked
  3. A programmatic way to read incoming messages
  4. Automatic cleanup when the session ends

This maps exactly to what a programmable temporary email infrastructure provides.

UnCorreoTemporal is that infrastructure. It accepts real SMTP traffic, stores messages in a PostgreSQL database, exposes a REST API for reading them, and pushes real-time events via WebSockets. Every inbox is ephemeral by design — it expires on a configurable schedule without any teardown code required.

It is also MCP-native, which means Claude Desktop and any other MCP-compatible agent runtime can call it as a tool directly. More on that at the end.

Setting Up in Google Colab

Open a new Colab notebook. No special setup required — requests is available in every Colab runtime.

# Cell 1 — install nothing, requests is pre-installed
import requests
import re
import time

BASE_URL = "https://api.uncorreotemporal.com/api/v1"
Enter fullscreen mode Exit fullscreen mode

Create an Inbox

A single POST call creates an ephemeral inbox. No authentication required for anonymous use. The response includes the address and a session_token that authenticates all subsequent operations.

# Cell 2 — create inbox
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(f"Inbox: {address}")
# Output: mango-panda-42@uncorreotemporal.com
Enter fullscreen mode Exit fullscreen mode

The address is human-readable (adjective-noun-number@uncorreotemporal.com) and unique per creation. The inbox is live immediately — it starts accepting SMTP delivery the moment it's created.

Read Incoming Messages

The list endpoint returns message metadata without loading full bodies, keeping responses fast. Fetch the full body only when a message has arrived.

# Cell 3 — wait for incoming email
def wait_for_email(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(f"No email received within {timeout}s")
Enter fullscreen mode Exit fullscreen mode

The full message response includes body_text and body_html as pre-decoded Python strings. No MIME parsing on your end — the server handles all of that internally via core/parser.py.

End-to-End Example: OTP Automation in Colab

Here's a complete flow you can run cell by cell. It simulates a signup that triggers a verification email, waits for it, and extracts the OTP.

# Cell 4 — simulate a service that sends a verification email
def simulate_signup(email: str) -> None:
    """
    In a real scenario this would be:
        requests.post("https://yourservice.com/register", json={"email": email})
    """
    print(f"[→] Registering with email: {email}")
Enter fullscreen mode Exit fullscreen mode 
# Cell 5 — extract OTP from body text
def extract_otp(body_text: str) -> str:
    match = re.search(r"(?<!d)(d{6})(?!d)", body_text)
    if not match:
        raise ValueError("No 6-digit OTP found in email body")
    return match.group(1)
Enter fullscreen mode Exit fullscreen mode 
# Cell 6 — full end-to-end flow
import requests, re, time

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

# Step 1: create inbox
resp = requests.post(f"{BASE_URL}/mailboxes", params={"ttl_minutes": 10})
resp.raise_for_status()
data = resp.json()
address, token = data["address"], data["session_token"]
print(f"[✓] Inbox created: {address}")
print(f"[✓] Expires at: {data['expires_at']}")

# Step 2: use the address to sign up (replace with your service)
print(f"[→] Submitting signup with {address}...")
# requests.post("https://yourservice.com/register", json={"email": address})

# Step 3: wait for the verification email
print("[…] Waiting for verification email...")
headers = {"X-Session-Token": token}
deadline = time.time() + 30

message = None
while time.time() < deadline:
    resp = requests.get(
        f"{BASE_URL}/mailboxes/{address}/messages",
        headers=headers,
    )
    resp.raise_for_status()
    msgs = resp.json()
    if msgs:
        msg_id = msgs[0]["id"]
        full = requests.get(
            f"{BASE_URL}/mailboxes/{address}/messages/{msg_id}",
            headers=headers,
        )
        full.raise_for_status()
        message = full.json()
        break
    time.sleep(1.5)

if not message:
    raise TimeoutError("Email not received in 30 seconds")

print(f"[✓] Email received from: {message['from_address']}")
print(f"[✓] Subject: {message['subject']}")

# Step 4: extract OTP
body = message["body_text"] or ""
match = re.search(r"(?<!d)(d{6})(?!d)", body)
if match:
    otp = match.group(1)
    print(f"[✓] OTP extracted: {otp}")
else:
    print("[!] No OTP found — check body_html if service sends HTML-only email")
    print(message["body_html"][:500])

print("[✓] Flow complete")
Enter fullscreen mode Exit fullscreen mode

Real-Time Option: WebSocket Instead of Polling 

import asyncio
import json
import websockets

async def wait_for_email_ws(address: str, token: str) -> str:
    url = f"wss://api.uncorreotemporal.com/ws/inbox/{address}?token={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"]
Enter fullscreen mode Exit fullscreen mode

The event is {"event": "new_message", "message_id": "<uuid>"}. The server sends a keepalive ping every 30 seconds to prevent connection drops in long-running workflows.

How This Works Internally 

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)  channel: mailbox:{address}
                │
                ▼
        WebSocket /ws/inbox/{address}
        {"event": "new_message", "message_id": "..."}
Enter fullscreen mode Exit fullscreen mode

Key components:

  • FastAPI async backend: All API endpoints use async SQLAlchemy with PostgreSQL. The ORM models define body_text, body_html, from_address, subject, and attachments as top-level fields — no raw MIME on your side.
  • SMTP ingestion: A local aiosmtpd-based server handles direct SMTP delivery in development. In production, AWS SES receives inbound mail and forwards it via SNS webhook to /api/v1/ses/inbound. Both paths funnel through the same deliver_raw_email() function.
  • Redis pub/sub: When a message is stored, it publishes to mailbox:{address}. The WebSocket handler in ws/inbox.py subscribes to that channel and forwards the event to connected clients.
  • Background expiration: Inboxes carry an expires_at timestamp. A background worker soft-deletes expired mailboxes automatically. No cleanup code in your automation.

Why This Matters for AI Agents

Email is a coordination primitive. It's how external services signal back to your agent. Without it:

  • Agents can't complete onboarding flows on third-party platforms
  • QA pipelines that test registration flows require human testers
  • Multi-step automation that depends on email confirmations stalls indefinitely

Concrete use cases this enables:

  • Autonomous signup testing: Create an account, confirm the email, assert the welcome message arrived — fully automated, repeatable in CI
  • Agent-driven account provisioning: An agent registers on a third-party service, receives and processes the confirmation, continues the workflow
  • Parallel testing without collision: 20 test runs, 20 inboxes, zero shared state
  • Ephemeral research agents: An agent registers for access to a dataset or service, uses it, and the inbox expires when the job is done

MCP Integration for Claude and Compatible Agents 

{
  "mcpServers": {
    "uncorreotemporal": {
      "command": "python",
      "args": ["-m", "uncorreotemporal.mcp.server"],
      "env": { "UCT_API_KEY": "uct_your_key_here" }
    }
  }
}
Enter fullscreen mode Exit fullscreen mode

The agent can then call create_mailbox(), get_messages(address), and read_message(address, message_id) as first-class tools.

What You Can Build

The pattern shown here — create inbox, trigger external flow, wait for email, extract code or link — composes into larger systems:

  • A pytest fixture that creates a fresh inbox per test case and tears down automatically
  • A CI/CD stage that provisions test accounts on every build with zero shared credentials
  • A Colab-based agent that navigates real-world signup flows end to end
  • A research tool that registers for services, collects confirmation emails, and logs structured data

The API surface is small. The core loop is three HTTP calls. The inbox handles the rest.

Try it at uncorreotemporal.com — your first inbox is one requests.post() away.

Top comments (0)