DEV Community: Francisco Perez

How to Test Email Signup Flows in CI/CD Pipelines

Francisco Perez — Mon, 30 Mar 2026 22:19:52 +0000

The Pain Every CI Pipeline Knows

It works locally. The signup form submits, the confirmation email arrives in your inbox, you paste the OTP, it verifies. Ship it.

Then CI runs. The test sends the signup request, polls... and times out. Or worse: it passes because you mocked the email sender and never actually tested that the email arrives, renders correctly, or contains a working code.

Email signup flows break in CI for a handful of predictable reasons:

Shared inboxes — two parallel test runs both register and poll the same address. Whichever test gets the email first passes; the other times out.
Mocked senders — you assert that send_email() was called, not that an email was received. Template bugs, SMTP credential expiry, and malformed links all pass through silently.
Brittle fixtures — hardcoded email addresses that accumulate stale messages across runs.
No isolation — there's no API to create a fresh inbox per test run, so you bolt on a shared test account and cross your fingers.

The fix is not a better mock. It's a real, isolated, programmable inbox that your test owns for its lifetime.

Quick Start

All examples in this article are from the temporary-email-api-examples repo. The python-example/ folder is a single-file script that demonstrates the full flow with plain requests — no extra libraries required.

Set two environment variables before running:

export UCT_API_BASE_URL="https://uncorreotemporal.com/api/v1"
export UCT_EMAIL_API_KEY="your_api_key_here"

Then:

git clone https://github.com/francofuji/temporary-email-api-examples
cd temporary-email-api-examples/python-example
pip install requests
python main.py

You will see a temporary inbox created, a polling loop start, and — once you trigger a real email to the address — an OTP extracted and printed.

The Flow: Create Inbox - Wait for Email - Extract OTP

The pattern is three steps. Here is how the example implements each one.

1. Create the Inbox

import os, requests

API_BASE_URL = os.getenv("UCT_API_BASE_URL")
API_KEY = os.getenv("UCT_EMAIL_API_KEY")
HEADERS = {"X-API-Key": API_KEY}

create_resp = requests.post(f"{API_BASE_URL}/inboxes", headers=HEADERS, timeout=10)
create_resp.raise_for_status()

inbox = create_resp.json()
inbox_id = inbox.get("id")

One POST, one inbox. No signup flow, no shared state. The inbox_id is the key you use for the next two steps.

2. Poll for Messages

import time, re

pattern = re.compile(r"\b(\d{6})\b")
start_time = time.time()

while True:
    if time.time() - start_time > 60:
        raise SystemExit("Timed out waiting for OTP")

    try:
        msg_resp = requests.get(
            f"{API_BASE_URL}/inboxes/{inbox_id}/messages",
            headers=HEADERS,
            timeout=10,
        )
        if msg_resp.status_code >= 500:
            time.sleep(3)
            continue
        msg_resp.raise_for_status()
        messages = msg_resp.json()
    except requests.RequestException as exc:
        print(f"Request failed: {exc}. Retrying...")
        time.sleep(3)
        continue

    if not messages:
        time.sleep(3)
        continue

The loop handles three distinct cases: server errors (5xx, which you retry), request failures (network blips), and an empty inbox (normal while waiting). Separating these is what makes polling reliable in CI — you don't bail on transient failures.

3. Extract the OTP

    body = messages[0].get("body", "")
    match = pattern.search(body)
    if match:
        otp = match.group(1)
        print(f"OTP found: {otp}")
        break

The regex \b(\d{6})\b matches any standalone 6-digit sequence. Adjust the pattern to match your application's format.

CI/CD Tips

Set an explicit timeout and fail loudly. The example uses 60 seconds. In CI, an infinite loop or a silently hanging test is worse than a hard failure. If the email doesn't arrive in 60 seconds, something upstream is broken — raise, don't sleep forever.

Use fixed poll intervals, not adaptive backoff. Poll every 3 seconds. Predictable timing makes failures easier to diagnose.

Handle 5xx separately from 4xx. A 500 from the messages endpoint is probably transient. A 401 or 404 means your credentials or inbox ID are wrong — retrying won't help. The example retries 5xx and raises immediately on 4xx via raise_for_status().

Don't share inbox IDs across test cases. Create one inbox per test run.

Store credentials in CI secrets, not in code.

# GitHub Actions
env:
  UCT_API_BASE_URL: ${{ vars.UCT_API_BASE_URL }}
  UCT_EMAIL_API_KEY: ${{ secrets.UCT_EMAIL_API_KEY }}

Why This Works in CI

Most email testing approaches fail in CI because they depend on something external: a shared inbox, a stubbed SMTP server, or a hardcoded test account that accumulates state across runs.

This approach is stateless from the pipeline's perspective. Each run calls POST /inboxes, gets a fresh address and ID back, and owns that inbox exclusively until the test exits or the TTL expires. There's nothing to clean up, nothing to coordinate between parallel jobs, and no leftover messages from a previous run to confuse the polling logic.

The inbox also receives real SMTP delivery — the same delivery path your production email goes through. If your email provider is down, your template is broken, or your SMTP credentials expired, the test will fail. That's the point. A test that can't catch those failures isn't testing email; it's testing whether your mock function was called.

Conclusion

Email signup testing is broken in most CI pipelines not because it's technically hard but because the default tools — mocks, shared inboxes, SMTP dev servers — don't give you the properties you need: isolation, real delivery, and an API to read the result programmatically.

The pattern in this article is straightforward: create a dedicated inbox per test, poll for the message with a hard timeout, extract the OTP or link, and proceed. The Python example is a working implementation you can drop into any pipeline today.

Next steps:

Browse the full temporary-email-api-examples repo for Node.js, shell/cURL, Playwright, and pytest examples
Get your API key at uncorreotemporal.com
Wrap the flow in a pytest fixture for per-test inbox isolation in larger test suites

Playwright + Temporary Email API: Full E2E Auth Flow Testing

Francisco Perez — Wed, 25 Mar 2026 02:23:35 +0000

The Problem with Testing Email Auth Flows

Most Playwright suites stop at the form submission. The test fills the email field, clicks Submit, asserts a success banner, and calls it a day. But the auth flow doesn't end there — it ends when the user clicks the link in their inbox. That second half is invisible to your test.

The common workarounds all have the same failure mode in disguise:

Mocking the email sender — you assert that sendEmail() was called, not that an email was received, rendered correctly, and contains a working link.
Shared test inboxes — two parallel CI runs both register and poll the same inbox. The first email gets consumed by whichever test polls first. Flaky tests guaranteed.
Hardcoded OTP bypass routes — test backdoors that sooner or later end up in a production build.
Mailhog / Mailpit — one more container to deploy, a different API surface from production, and zero validation of your real SMTP path.

The issue isn't test tooling. It's the assumption that email delivery is a side effect you can safely skip or stub. In practice, your verification link can be malformed, your SMTP credentials can expire, your template can break — and a mocked test will pass through all of it.

What We Will Build

By the end of this article you will have a fully working TypeScript Playwright test that:

Creates a temporary inbox via the uncorreotemporal.com API
Submits a signup form using that inbox address
Polls for the incoming email with a configurable timeout
Extracts the OTP code or verification link from the message body
Completes the auth flow inside Playwright
Cleans up the inbox in teardown

The test uses real SMTP delivery — your application sends to the temporary inbox exactly as it would send to a real user. No interceptors, no stubs.

Architecture Overview

[Your App] --signup form--> [FastAPI / Node backend]
                                    |
                              sends real email via SMTP
                                    |
                                    v
                         [uncorreotemporal.com SMTP receiver]
                                    |
                               stores message
                                    |
                  +------------------------------------------+
                  |  REST API: GET /api/v1/mailboxes/        |
                  |           {address}/messages             |
                  +------------------------------------------+
                                    |
                             Playwright test
                            polls -> extracts -> continues

The temporary inbox acts as a controllable endpoint on the SMTP receive side. You created it with one HTTP call; you read from it with another; you tear it down when the test is done. The application under test never knows the difference.

Setup

Install Playwright

npm install --save-dev @playwright/test
npx playwright install chromium

Environment Variables

# .env.test (never commit this)
UCT_API_KEY=uct_your_key_here
DEMO_BASE_URL=http://localhost:3000   # your app under test

Get your API key at uncorreotemporal.com. The free tier is sufficient for local development; for parallel CI pipelines use a paid key with higher inbox quotas.

Playwright Config

// playwright.config.ts
import { defineConfig } from "@playwright/test";

export default defineConfig({
  testDir: "./tests",
  timeout: 60_000,
  workers: process.env.CI ? 4 : undefined,
  use: {
    headless: true,
    baseURL: process.env.DEMO_BASE_URL ?? "http://localhost:3000",
  },
});

Step-by-Step Implementation

Step 1 — Create an Inbox via API

// tests/helpers/email.ts

const BASE = "https://uncorreotemporal.com/api/v1";

interface Inbox {
  address: string;
  expires_at: string;
  session_token: string;
}

export async function createInbox(ttlMinutes = 15): Promise<Inbox> {
  const res = await fetch(`${BASE}/mailboxes`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-API-Key": process.env.UCT_API_KEY!,
    },
    body: JSON.stringify({ ttl_minutes: ttlMinutes }),
  });

  if (!res.ok) {
    throw new Error(`Failed to create inbox: ${res.status} ${await res.text()}`);
  }

  return res.json();
}

The response:

{
  "address": "crisp-falcon-77@uncorreotemporal.com",
  "expires_at": "2026-03-24T15:00:00+00:00",
  "session_token": "dGhpcyBpcyBhIHNhbXBsZSB0b2tlbg"
}

Step 2 — Use the Inbox in the Signup Form

test("signup -> email verification -> dashboard", async ({ page }) => {
  const inbox = await createInbox();

  try {
    await page.goto("/register");
    await page.fill('[name="email"]', inbox.address);
    await page.fill('[name="password"]', "Secure1234!");
    await page.click('[type="submit"]');

    await page.waitForURL("**/confirm**");
    await expect(page.locator("h1")).toContainText("Check your email");

    const email = await waitForEmail(inbox.address, inbox.session_token);
    // ... extract and complete
  } finally {
    await deleteInbox(inbox.address, inbox.session_token);
  }
});

Step 3 — Wait for the Email (Polling with Retry)

export async function waitForEmail(
  address: string,
  sessionToken: string,
  options: { timeout?: number; interval?: number; subjectFilter?: string } = {}
): Promise<FullMessage> {
  const { timeout = 30_000, interval = 2_000, subjectFilter } = options;
  const encoded = encodeURIComponent(address);
  const headers = { "X-API-Key": process.env.UCT_API_KEY! };
  const deadline = Date.now() + timeout;

  while (Date.now() < deadline) {
    let messages: MessageSummary[];

    try {
      const res = await fetch(`${BASE}/mailboxes/${encoded}/messages`, { headers });
      if (!res.ok) throw new Error(`HTTP ${res.status}`);
      messages = await res.json();
    } catch (err) {
      console.warn(`[waitForEmail] poll error: ${err}. Retrying...`);
      await sleep(interval);
      continue;
    }

    const match = subjectFilter
      ? messages.find((m) => m.subject.toLowerCase().includes(subjectFilter.toLowerCase()))
      : messages[0];

    if (match) {
      const res = await fetch(`${BASE}/mailboxes/${encoded}/messages/${match.id}`, { headers });
      if (!res.ok) throw new Error(`Failed to fetch message body: ${res.status}`);
      return res.json();
    }

    await sleep(interval);
  }

  throw new Error(`[waitForEmail] No email arrived at ${address} within ${timeout}ms`);
}

Key design: network errors are retried, not propagated. "No message" means keep polling — not failure. Use subjectFilter when your app sends multiple emails per registration.

Step 4 — Parse the Email (OTP and Link Extraction)

export function extractOTP(message: FullMessage): string {
  const sources = [message.body_text, message.body_html].filter(Boolean) as string[];
  for (const content of sources) {
    const match = content.match(/\b(\d{6})\b/);
    if (match) return match[1];
  }
  throw new Error("OTP not found in email body");
}

export function extractVerificationLink(message: FullMessage, urlPrefix: string): string {
  const pattern = new RegExp(`${escapeRegex(urlPrefix)}[^\\s"'<>]+`);
  const sources = [message.body_text, message.body_html].filter(Boolean) as string[];
  for (const content of sources) {
    const match = content.match(pattern);
    if (match) return match[0];
  }
  throw new Error(`Verification link not found (prefix: ${urlPrefix})`);
}

Prefer body_text for numeric OTP extraction — it's cleaner than parsing HTML.

Step 5 — Complete the Auth Flow

OTP flow:

const email = await waitForEmail(inbox.address, inbox.session_token, {
  subjectFilter: "Your verification code",
});
const otp = extractOTP(email);

await page.fill('[name="otp"]', otp);
await page.click('[type="submit"]');
await expect(page.locator('[data-testid="dashboard"]')).toBeVisible();

Verification link flow:

const link = extractVerificationLink(email, `${process.env.DEMO_BASE_URL}/verify`);
await page.goto(link);
await expect(page.locator("h1")).toHaveText("Account verified");

Full Working Example

// tests/auth/register.spec.ts
import { test, expect } from "@playwright/test";

const BASE = "https://uncorreotemporal.com/api/v1";
const API_HEADERS = {
  "Content-Type": "application/json",
  "X-API-Key": process.env.UCT_API_KEY!,
};

async function createInbox(ttlMinutes = 15) {
  const res = await fetch(`${BASE}/mailboxes`, {
    method: "POST",
    headers: API_HEADERS,
    body: JSON.stringify({ ttl_minutes: ttlMinutes }),
  });
  if (!res.ok) throw new Error(`createInbox failed: ${res.status}`);
  return res.json() as Promise<{ address: string; session_token: string }>;
}

async function waitForEmail(address: string, timeout = 30_000, subjectFilter?: string) {
  const encoded = encodeURIComponent(address);
  const deadline = Date.now() + timeout;
  while (Date.now() < deadline) {
    const res = await fetch(`${BASE}/mailboxes/${encoded}/messages`, { headers: API_HEADERS });
    if (!res.ok) { await sleep(2_000); continue; }
    const messages: Array<{ id: string; subject: string }> = await res.json();
    const hit = subjectFilter
      ? messages.find((m) => m.subject.toLowerCase().includes(subjectFilter.toLowerCase()))
      : messages[0];
    if (hit) {
      const full = await fetch(`${BASE}/mailboxes/${encoded}/messages/${hit.id}`, { headers: API_HEADERS });
      return full.json() as Promise<{ body_text: string | null; body_html: string | null }>;
    }
    await sleep(2_000);
  }
  throw new Error(`No email at ${address} after ${timeout}ms`);
}

async function deleteInbox(address: string) {
  const encoded = encodeURIComponent(address);
  await fetch(`${BASE}/mailboxes/${encoded}`, { method: "DELETE", headers: API_HEADERS });
}

function extractOTP(body_text: string | null, body_html: string | null): string {
  for (const content of [body_text, body_html]) {
    if (!content) continue;
    const m = content.match(/\b(\d{6})\b/);
    if (m) return m[1];
  }
  throw new Error("OTP not found in email");
}

function sleep(ms: number) { return new Promise<void>((r) => setTimeout(r, ms)); }

test("full signup -> OTP -> dashboard", async ({ page }) => {
  const inbox = await createInbox();
  try {
    await page.goto("/register");
    await page.fill('[name="email"]', inbox.address);
    await page.click('[type="submit"]');
    await page.waitForURL("**/confirm**");

    const email = await waitForEmail(inbox.address, 30_000, "OTP");
    const otp = extractOTP(email.body_text, email.body_html);

    await page.fill('[name="email"]', inbox.address);
    await page.fill('[name="otp"]', otp);
    await page.click('[type="submit"]');

    await expect(page.locator("h1")).toHaveText("Welcome to the dashboard");
  } finally {
    await deleteInbox(inbox.address);
  }
});

CI/CD Integration

# .github/workflows/e2e.yml
name: E2E Tests
on:
  push:
    branches: [main, staging]
  pull_request:

jobs:
  playwright:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    env:
      UCT_API_KEY: ${{ secrets.UCT_API_KEY }}
      DEMO_BASE_URL: ${{ vars.STAGING_URL }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with: { node-version: 20, cache: npm }
      - run: npm ci
      - run: npx playwright install --with-deps chromium
      - run: npx playwright test
      - uses: actions/upload-artifact@v4
        if: failure()
        with:
          name: playwright-report
          path: playwright-report/

Each CI run creates isolated inboxes. Parallel matrix jobs need zero coordination — each leg creates its own inbox.

Best Practices

Use subjectFilter when your app sends multiple emails per registration (welcome + confirmation). messages[0] on a two-message inbox is a race condition.

Distinguish network errors from missing emails. A 503 from the API should retry silently. An empty inbox after 30 seconds is a real failure.

Set ttl_minutes=5 in CI. Shorter TTLs reduce orphaned inbox accumulation if a test crashes before finally runs.

One inbox per test(), not per describe(). Two tests sharing an inbox re-introduce shared-state problems.

Add jitter for high-concurrency pipelines:

await sleep(Math.random() * 200);
const inbox = await createInbox();

Log the inbox address on failure:

test.info().annotations.push({ type: "inbox", description: inbox.address });

Why Real Email Testing Matters

Bug	Mock catches	Real inbox catches
`sendEmail()` not called	Yes	Yes
Email template render failure	No	Yes
Broken verification link format	No	Yes
Wrong base URL for the environment	No	Yes
SMTP credential expiry	No	Yes
OTP in email doesn't match backend	No	Yes

The most painful production bugs are in the second column. Testing against real SMTP delivery does not require standing up your own mail server. Programmable Temporary Email Infrastructure — what uncorreotemporal.com provides — is designed exactly for this.

Try It

The complete working examples — including the demo Express app the Playwright tests use as the application under test — are in the temporary-email-api-examples repository on GitHub.

Get your API key and start testing real email flows at https://uncorreotemporal.com

Why AI Agents in Google Colab Need Real Email Infrastructure

Francisco Perez — Mon, 23 Mar 2026 19:30:36 +0000

Capable but Incomplete

AI agents have made a remarkable leap in the past two years. They can write working code, call external APIs, browse the web, parse documents, orchestrate multi-step tasks, and operate in rich environments like Google Colab that give them access to real compute. By many measures, they look like fully capable automation systems.

But there is a class of workflow they consistently fail to complete: anything that requires interacting with email.

Not because the model is not smart enough. Not because the tooling is too complex. But because the infrastructure does not exist for them to use. Email — real email, with real SMTP, real delivery, real inboxes — has never been designed with agents in mind. And until it is, agents that seem capable of handling real-world tasks will keep running into the same invisible wall.

This is not a product gap. It is an infrastructure gap. That distinction matters more than most people realize.

The Illusion of Capability

Spend an hour with a capable agent in Google Colab and you will be impressed. It can install packages, generate and run code, query REST APIs, parse JSON, scrape pages, and feed results back into an LLM for reasoning. The loop feels tight and capable.

Then try to build an agent that registers for an external service and verifies the account.

The agent can fill out the form. It can submit the registration. It can handle the HTTP response. But then the service sends a confirmation email — and the agent has no inbox to receive it. The workflow terminates at the one point where email is a hard dependency, not an optional channel.

This is not an edge case. Account signup is one of the most common real-world automations. Password reset flows require email. OTP delivery relies on email. Subscription confirmations, trial activations, approval chains — nearly every SaaS product uses email as a critical control point at some stage of its workflow. An agent that cannot interact with email is an agent that can demo well but cannot complete real work.

The capability agents actually have is impressive. What is missing is the infrastructure layer that would let them use it.

Why Email Matters More Than It Seems

There is a tendency to treat email as a legacy communication channel — something old-fashioned that will eventually be replaced by APIs and webhooks. In practice, email is one of the most durable primitives on the internet.

It is the backbone of identity verification. You cannot register for a serious service without providing an email address that you control. That verification step is not a technical formality — it is a trust anchor. The email address is proof that you own an inbox, and owning an inbox is how the web has decided to define a real person — or a real entity — since the 1990s.

This means email is inherently tied to identity, authentication, and access control. It is the channel through which OTPs arrive, through which password resets are authorized, through which trial activations are gated. It is asynchronous by nature — the service sends a message and waits for you to act on it, without any guarantee of when you will respond. That async nature is exactly what makes it hard to handle programmatically.

Receiving email is not like calling an API. There is no request-response cycle you can initiate. You cannot await inbox.get(). An email arrives when the sender decides to send it, on infrastructure you do not control, and you need to be ready to receive it. For a human with an email client, this is trivial. For an agent, it requires infrastructure that does not come for free.

That infrastructure — real SMTP ingestion, programmatic inbox access, event-driven delivery — is not difficult to build. But it has not been built for agents. Most existing temporary email services were designed for humans who need a throwaway address in a browser. They are not designed to be called from Python, integrated into an agent loop, or managed programmatically. They are interfaces, not infrastructure.

The Current Broken Approaches

When developers hit this wall, they tend to reach for one of several workarounds. None of them hold up at scale.

Shared inboxes. Some teams use a real email account — Gmail, Outlook, a company address — as the target for all test emails. This works for one developer running one test at a time. The moment two CI workers run in parallel, messages from different runs land in the same inbox and cross-contaminate. There is no isolation. You cannot reliably know which message belongs to which run.

Mocks. You skip the real email flow and simulate it in your test. The agent pretends to receive an OTP. The test passes. What you have actually tested is your code's ability to handle a mocked input — not whether the real email delivery system works, not whether the OTP format matches what the actual service sends, not whether your regex correctly extracts from the real email body. Mocks are useful for unit testing logic. They are liabilities for integration testing real workflows.

Scraping Gmail. OAuth flows, scope management, token refresh, filtering out production emails from test emails — this is a non-trivial integration that has to be rebuilt for every new project. It requires a real account tied to a real identity. It creates noise in a production inbox. It is fragile to Gmail UI changes and quota limits. It is not composable into an agent framework.

Manual steps. The most honest broken approach: the human operator stops the agent, reads the email, copies the OTP, and resumes the workflow. This is not automation. This is automation with a gap that someone has to fill in by hand.

The reason these approaches persist is not that developers are unaware of their limitations. It is that there is no off-the-shelf infrastructure that solves the problem properly. So teams use what is available and work around the gaps.

Infrastructure vs Tooling

This distinction is the core of the problem.

A tool solves a specific, bounded task. "Read an email" is a tool. You reach for it when you need it, it does one thing, and you move on. A tool mindset works well when the surface is simple and the frequency is low. It does not scale.

Infrastructure is what you build when the tool is not enough. Infrastructure means:

Isolated inboxes: each agent session, each test run, each parallel worker gets its own inbox with no shared state
Lifecycle management: inboxes are created before they are needed, expire when they should, and are deleted when the workflow is done — without manual cleanup
Async handling: the system knows how to wait for an email to arrive without blocking the agent loop or polling in a tight loop
Programmatic access: every operation — create, read, delete — is available via an API that code can call, not a UI that a human clicks

The gap is not a missing feature. It is a missing layer. Consumer disposable email services were designed for humans. They solve the throwaway address problem for a person with a browser. The architectural intent was never to serve machines, so the design does not accommodate them: no authentication, no isolation, no API, no event delivery.

Building email as infrastructure — designed from the start for programmatic access by agents and automation systems — is a different project than adding an API to a browser tool. It requires a different mental model: the inbox is a resource, like a database row or a queue. It has an owner, a lifecycle, a quota, and an access control layer. Agents are first-class consumers of that resource, not afterthoughts.

This is where the category needs to go. Not "temporary email service" but "programmable email infrastructure." The difference is architectural, not cosmetic.

The Rise of MCP and Tool-Based Agents

The Model Context Protocol is changing how agents are built. Instead of hard-coding every capability into an agent framework, MCP standardizes how tools are defined, discovered, and called. An MCP-compatible agent can dynamically acquire new capabilities — search, code execution, database access, browser control — by connecting to MCP servers that expose those capabilities as typed tools.

The implication is significant: capabilities are becoming modular and externalized. Agent architectures are decoupling from specific implementations. A capability is a server you connect to, not code you write.

This is good for quality. When capabilities are externalized, they can be maintained by the teams with the deepest expertise in each domain. The team that builds the search tool knows more about search than the agent developer does. The team that builds the code execution environment knows more about sandboxing. This specialization produces better tools.

Email should follow the same pattern. If an agent needs email, it should be able to connect to an MCP server that exposes email as a typed, callable tool — without the agent developer needing to understand SMTP, Redis pub/sub, or inbox lifecycle management. The capability is available; the implementation is encapsulated.

When email is treated as a first-class tool in the MCP ecosystem, agent builders stop re-solving the infrastructure problem in every project. They connect to the server and get email. The work shifts to what actually matters: what the agent does with the email after it arrives.

What Real Email Infrastructure Looks Like

The characteristics are not exotic. They follow directly from what programmatic access requires.

Real SMTP ingestion. Not simulation. Not a service that pretends to accept email. A real mail transfer agent with real domain configuration, MX records, and compliance with RFC standards. Email that arrives is actual email — delivered via the same path as production traffic, subject to the same validation.

API-first access. Every operation is available as a documented HTTP endpoint. Creating an inbox is a POST. Reading messages is a GET. Deleting an inbox is a DELETE. The API is the interface; the UI is optional.

Ephemeral inboxes. Each inbox has a configurable TTL and expires deterministically. The agent does not need to worry about cleanup — the infrastructure handles it. Failure cases are covered automatically.

Isolation per task. Each agent session, each test run, each parallel workflow gets its own inbox. API key scoping ensures that one agent cannot accidentally read another's messages. There is no shared state to corrupt.

Event-driven delivery. When a message arrives, connected clients are notified immediately — no polling, no race conditions, no guessing at delivery latency. The agent opens a subscription before triggering the external workflow, then receives the event when the email arrives.

These are not aspirational features. They are the minimum viable set for infrastructure that agents can actually rely on.

What This Enables

When email works as infrastructure — not a workaround, not a mock — the scope of what agents can do expands considerably.

Autonomous account creation. An agent can register for external services, receive the confirmation email, extract the verification link, and complete the signup without any human in the loop. The entire flow, end-to-end.

Full CI/CD email testing. A test pipeline can provision a fresh inbox per job, trigger a real registration flow against a staging environment, receive the real email, verify the real OTP, and delete the inbox on completion. No shared inboxes, no mocks, no flaky tests caused by isolation failures.

AI agents completing real workflows. The category of task expands from "things that don't require email" to "anything that requires email" — which, in practice, means most of the real web. Signup flows, password resets, account verification, transactional approval chains, subscription activations.

Automation beyond APIs. Some systems do not have APIs. They have email. Legacy enterprise workflows, approval chains, notification systems built around email as the integration point. Agents that can interact with email can participate in these workflows without requiring modernization of the systems they interact with.

The unlock here is not incremental. It is categorical. Email is not one more tool in a long list. It is a gate that blocks access to a large fraction of the real web for agents that do not have it.

One Example of This Infrastructure

uncorreotemporal.com is a programmable temporary email infrastructure built around these principles. It runs a real SMTP server, stores messages with full RFC 2822 bytes alongside parsed fields, delivers events via WebSocket over Redis pub/sub, and exposes everything through a REST API and a native MCP server.

Inboxes are created per API call, scoped to API keys, and expire on configurable TTLs. The MCP server exposes five typed tools that agents can call without any email infrastructure knowledge. The whole thing can be used from Google Colab with a single pip install and an API key.

It is not the only way to solve the infrastructure problem. But it is an example of what the infrastructure layer looks like when it is built for agents rather than adapted from a browser tool.

The Agents That Will Matter

There is a useful frame for thinking about where agent capabilities are heading.

The first generation of capable agents impressed people because they could handle reasoning tasks: summarization, code generation, question answering. These are tasks that agents can complete entirely within their context window.

The second generation is defined by tool use. Agents that can call APIs, browse the web, execute code, and interact with external systems. Google Colab is a powerful environment for this — real Python, real compute, real libraries, real network access.

The third generation will be defined by infrastructure. Agents that do not just call tools, but participate in real-world systems as first-class actors. Systems that have state, history, async interactions, and long-lived workflows. Email is one of the first such systems.

Agents will not just call APIs. They will receive emails, respond to confirmation flows, participate in async approval chains, and interact with systems that were never designed with an API in mind. The infrastructure layer that makes this possible is not a feature request — it is the next constraint to solve.

Email is one of the first layers that needs to be solved properly. Not wrapped, not mocked, not scraped. Built as infrastructure, exposed as a tool, and made available to every agent that needs it.

The agents that will actually matter in production are the ones that can navigate the real web — including the parts that run on email.

uncorreotemporal.com

Add Email Capabilities to AI Agents in Google Colab

Francisco Perez — Sun, 22 Mar 2026 16:47:18 +0000

The Capability Gap

Modern AI agents have impressive reach. They can browse the web, execute code, call external APIs, read documents, and invoke dozens of tools. Frameworks like LangChain, CrewAI, and AutoGen ship with built-in tool wrappers for search engines, calculators, code interpreters, and databases.

But there is a capability they are almost universally missing: interacting with email.

This is not a minor gap. Email is the control plane for most external services. Account registration, one-time passwords, verification links, subscription confirmations, approval chains — nearly every system that matters uses email as a coordination mechanism. An agent that cannot interact with email can navigate to the "Check your inbox" page and stop there.

The usual workaround is to skip email entirely, mock the verification step, or require a human in the loop. These are all admissions of failure.

What agents actually need is email as a callable capability — not a workaround, not a one-off script, but a reusable component the agent can invoke whenever the task requires it.

From Script to Capability

There is a natural progression in how developers add email to their automation flows:

Stage 1 — One-off script. You write a quick function that creates an inbox, polls for a message, and extracts a code. It works. You ship it. Six weeks later a different agent needs email for a different flow, so you copy the function and modify it slightly. The codebase now has three slightly different implementations of the same thing. They diverge. One breaks.

Stage 2 — Shared utility. You extract the logic into a shared module. But it is still a flat collection of functions with no coherent interface. The caller needs to know which fields to pass, which headers to use, and how errors propagate. The abstraction is partial.

Stage 3 — Capability component. Email is encapsulated as a class with a clear interface. The agent does not know or care about HTTP calls, session tokens, or polling intervals. It calls create_inbox(), wait_for_email(), or extract_otp() — and gets back structured results. The implementation can be swapped or upgraded without touching the agent.

Stage 3 is what we want. Email becomes infrastructure that the agent consumes through a well-defined contract.

Designing the Interface

A minimal email capability interface for an agent needs to answer four questions:

Question	Operation
Where should I receive mail?	`create_inbox(ttl_minutes)`
Did any mail arrive?	`wait_for_email(address, timeout)`
What is the latest message?	`get_latest_message(address)`
What is the verification code?	`extract_otp(message)`

Let's define these contracts before writing any implementation:

# Contract: what the agent expects

def create_inbox(ttl_minutes: int = 10) -> Inbox:
    """
    Returns an Inbox with:
        .address  — e.g. mango-panda-42@uncorreotemporal.com
        .token    — session token for subsequent operations
        .expires_at — datetime string
    """

def wait_for_email(inbox: Inbox, timeout: int = 30) -> Message:
    """
    Blocks until an email arrives or timeout is reached.
    Returns a Message with:
        .id, .from_address, .subject, .body_text, .body_html
    Raises TimeoutError if no email arrives in time.
    """

def get_latest_message(inbox: Inbox) -> Message | None:
    """
    Returns the most recent message, or None if inbox is empty.
    Non-blocking.
    """

def extract_otp(message: Message, digits: int = 6) -> str:
    """
    Parses the message body for a numeric OTP code.
    Returns the code as a string.
    Raises ValueError if no code is found.
    """

Clean inputs, typed outputs, one job per function. The agent calls these without knowing anything about the underlying transport.

Building the EmailCapability Class

Here is a production-ready implementation against uncorreotemporal.com. This is the same class you would use in a Google Colab notebook, a pytest fixture, or a LangChain tool wrapper.

import re
import time
import requests
from dataclasses import dataclass

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


@dataclass
class Inbox:
    address: str
    token: str
    expires_at: str


@dataclass
class Message:
    id: str
    from_address: str
    subject: str | None
    body_text: str | None
    body_html: str | None
    received_at: str
    is_read: bool


class EmailCapability:
    """
    Reusable email capability for AI agents.
    Wraps the uncorreotemporal.com API with a clean, agent-friendly interface.
    """

    def __init__(self, api_key: str | None = None):
        self._headers = (
            {"Authorization": f"Bearer {api_key}"} if api_key else {}
        )

    def create_inbox(self, ttl_minutes: int = 10) -> Inbox:
        """Create an ephemeral inbox. Returns address + session token."""
        resp = requests.post(
            f"{BASE_URL}/mailboxes",
            headers=self._headers,
            params={"ttl_minutes": ttl_minutes},
        )
        resp.raise_for_status()
        data = resp.json()
        return Inbox(
            address=data["address"],
            token=data.get("session_token", ""),
            expires_at=data["expires_at"],
        )

    def wait_for_email(
        self,
        inbox: Inbox,
        timeout: int = 30,
        poll_interval: float = 1.5,
    ) -> Message:
        """Block until a message arrives. Raises TimeoutError on timeout."""
        headers = self._auth_headers(inbox.token)
        deadline = time.time() + timeout

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

            if messages:
                return self._fetch_message(inbox, messages[0]["id"])

            time.sleep(poll_interval)

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

    def get_latest_message(self, inbox: Inbox) -> Message | None:
        """Non-blocking: return the most recent message, or None."""
        headers = self._auth_headers(inbox.token)
        resp = requests.get(
            f"{BASE_URL}/mailboxes/{inbox.address}/messages",
            headers=headers,
        )
        resp.raise_for_status()
        messages = resp.json()
        if not messages:
            return None
        return self._fetch_message(inbox, messages[0]["id"])

    def extract_otp(self, message: Message, digits: int = 6) -> str:
        """Extract a numeric OTP code from a message body."""
        body = message.body_text or ""
        pattern = rf"(?<!\d)(\d{{{digits}}})(?!\d)"
        match = re.search(pattern, body)
        if not match:
            raise ValueError(
                f"No {digits}-digit OTP found in message from {message.from_address}"
            )
        return match.group(1)

    def _auth_headers(self, session_token: str) -> dict:
        if self._headers:
            return self._headers
        return {"X-Session-Token": session_token}

    def _fetch_message(self, inbox: Inbox, message_id: str) -> Message:
        headers = self._auth_headers(inbox.token)
        resp = requests.get(
            f"{BASE_URL}/mailboxes/{inbox.address}/messages/{message_id}",
            headers=headers,
        )
        resp.raise_for_status()
        data = resp.json()
        return Message(
            id=data["id"],
            from_address=data["from_address"],
            subject=data.get("subject"),
            body_text=data.get("body_text"),
            body_html=data.get("body_html"),
            received_at=data["received_at"],
            is_read=data["is_read"],
        )

No additional dependencies — requests is pre-installed in every Colab runtime.

Using It in an Agent Loop

This is where the abstraction earns its value. Here is a simple agent loop running in Google Colab:

email = EmailCapability()  # anonymous mode; pass api_key= for higher quotas

# Step 1: Agent decides it needs an inbox
print("[agent] Task requires email. Creating inbox...")
inbox = email.create_inbox(ttl_minutes=10)
print(f"[agent] Inbox ready: {inbox.address}")

# Step 2: Agent performs the external action
def register_with_service(email_address: str) -> None:
    print(f"[agent] Registering at external service with: {email_address}")
    # requests.post("https://yourapp.com/signup", json={"email": email_address})

register_with_service(inbox.address)

# Step 3: Agent waits for the email capability to deliver the message
print("[agent] Waiting for verification email...")
try:
    message = email.wait_for_email(inbox, timeout=30)
    print(f"[agent] Email received from: {message.from_address}")
except TimeoutError:
    print("[agent] No email arrived. Aborting.")
    raise

# Step 4: Agent extracts the OTP and continues
otp = email.extract_otp(message)
print(f"[agent] OTP extracted: {otp}")
print("[agent] Task complete.")

The loop expresses intent: "I need to receive an email and extract a code." The capability handles the mechanism.

For flows requiring multiple emails:

first  = email.wait_for_email(inbox, timeout=30)   # confirmation email
second = email.wait_for_email(inbox, timeout=30)   # welcome email
receipt_link = re.search(r"https://\S+", second.body_text or "")

Before and After

Without the capability:

resp = requests.post("https://api.uncorreotemporal.com/api/v1/mailboxes", params={"ttl_minutes": 10})
data = resp.json()
address = data["address"]
token = data["session_token"]

deadline = time.time() + 30
found = None
while time.time() < deadline:
    r = requests.get(f"https://api.uncorreotemporal.com/api/v1/mailboxes/{address}/messages",
                     headers={"X-Session-Token": token})
    msgs = r.json()
    if msgs:
        full = requests.get(f"https://api.uncorreotemporal.com/api/v1/mailboxes/{address}/messages/{msgs[0]['id']}",
                            headers={"X-Session-Token": token})
        found = full.json()
        break
    time.sleep(1.5)

m = re.search(r"\b(\d{6})\b", found.get("body_text", ""))
otp = m.group(1)

With the capability:

inbox = email.create_inbox(ttl_minutes=10)
register_with_service(inbox.address)
message = email.wait_for_email(inbox, timeout=30)
otp = email.extract_otp(message)

Four lines. No duplication. Swappable at any layer.

The Natural Connection to MCP

Once you have a well-defined interface, the next step is exposing it as an MCP tool so any LLM agent can call it natively:

EmailCapability method	MCP Tool name
`create_inbox(ttl_minutes)`	`create_mailbox`
`wait_for_email(inbox, timeout)`	`get_messages(address)`
`get_latest_message(inbox)`	`get_messages(address, limit=1)`
read full body	`read_message(address, message_id)`

UnCorreoTemporal ships a native MCP server. Configure it once:

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

The model calls create_mailbox, get_messages, and read_message as first-class tool invocations — the same way it calls a web search or code interpreter.

Why This Architecture Supports It

Ephemeral inboxes by default. Each inbox carries an expires_at timestamp and is soft-deleted automatically. No teardown logic needed.

Structured message fields. The API returns body_text and body_html as pre-decoded Python strings. The internal core/parser.py handles RFC 2822 parsing, quoted-printable decoding, and base64 extraction.

Real SMTP delivery. Inboxes are not simulated — aiosmtpd in development, AWS SES in production. When wait_for_email returns, the message is there.

Async delivery events. The system publishes to Redis pub/sub (mailbox:{address}) the instant a message is stored. The WebSocket endpoint forwards these in milliseconds. wait_for_email can be upgraded from polling to push with no interface change.

Real Use Cases

Autonomous account provisioning — agent registers, confirms email, continues workflow
QA test suites — each test case gets an isolated inbox, parallel runs never collide
Onboarding automation — agent handles each transactional email at the right step
LLM research pipelines — model registers for service access, processes confirmation, cleans up

The agent loop is always the same: create inbox, perform action, wait for email, extract signal, continue.

Conclusion

The missing piece in most agent capability stacks is not a new model feature — it is a clean abstraction over infrastructure that has existed for decades.

Email is not a niche integration. It is a coordination primitive that most real systems depend on. Building agents that treat it as a first-class capability — rather than bolting on one-off scripts that break and diverge — is the difference between fragile prototypes and systems that hold up.

The EmailCapability class above is a starting point. The interface is small by design. It composes cleanly with agent loops, pytest fixtures, LangChain tools, and MCP servers.

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

Automate OTP Flows in Google Colab with AI Agents

Francisco Perez — Sat, 21 Mar 2026 21:26:39 +0000

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, 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):
    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

Filter by subject when the flow sends more than one email:

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

OTP pattern variations

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

# 8-digit code
extract_otp(body, pattern=r'\b\d{8}\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

import html, 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 for quick experiments: no local setup, shareable notebooks, free compute. The code above runs as-is in a Colab cell. For AI agent workflows — LangChain, CrewAI, 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

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

Isolated inboxes. Each call to POST /api/v1/mailboxes creates a new address. Messages are only accessible via the inbox's session_token. Parallel test runs cannot interfere.

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.

Get Started

The API requires no upfront setup for anonymous inboxes — just POST to create one and start receiving email. Free tier covers 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 for 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.

Using Google Colab MCP with a Temporary Email API

Francisco Perez — Fri, 20 Mar 2026 16:42:52 +0000

The Gap in the MCP Tool Ecosystem

The Model Context Protocol is doing something important: it is turning external capabilities into first-class tools that AI agents can discover and invoke without custom integration code. Google Colab MCP takes this further — your agent can now execute Python code in a real Colab runtime as a tool call. Search, web browsing, code execution, structured data retrieval: these are all becoming standardized tool surfaces.

But there is a gap that almost every real-world automation workflow hits eventually.

Agents still cannot reliably interact with email.

This is not an oversight. Email is structurally awkward to expose as a tool. It requires live SMTP infrastructure, async delivery timing, MIME parsing, and some mechanism to match a received message to a specific agent session. None of that is obvious to implement, and existing solutions — shared inboxes, Gmail IMAP, mocked flows — break under real conditions.

This article shows how to close that gap. You will see how a temporary email API exposes real inbox capabilities as MCP tools, how to call those tools from Google Colab using the Python MCP client, and why the MCP approach is meaningfully different from calling the REST API directly.

What MCP Actually Changes

The difference between calling a REST API and using an MCP tool is not just syntax. It is a change in who is responsible for knowing what is available.

With a REST API, the agent (or the human building the agent) has to know the endpoint, the method, the request shape, and the response schema before writing a single line of code. The knowledge is baked into the integration layer — usually as custom glue code, a wrapper function, or a prompt that tells the LLM what to call.

With MCP, the agent calls tools/list and gets back a structured description of every available capability: names, descriptions, input schemas, and output shapes. The agent can reason about which tools to use without any prior instruction. An agent runtime like Claude Desktop, AutoGen, or LangGraph can consume an MCP server it has never seen before and immediately know what it can do.

This matters for agent design in a concrete way: it removes the coupling between the agent's reasoning and the specific API surface of each service. You do not need to update your prompt or your wrapper code when the underlying service changes. You add a server to the configuration, and the capability becomes available.

For email specifically, this means the agent does not need to know anything about SMTP, MIME, or polling intervals. It calls create_signup_inbox, gets an address, uses it, and calls wait_for_verification_email. The infrastructure knowledge stays in the server.

The Missing Tool: Email

Email shows up as a dependency in a surprisingly large fraction of real-world automation:

Account signups that require a confirmation link before the account activates
Two-factor authentication flows where the second factor is an OTP sent to the registered address
Subscription confirmations that gate access to an API key or resource
Password resets that block account recovery until the user clicks a link

Each of these is a hard stop for an autonomous agent. The agent can complete every other step of the workflow. Then it hits "check your email" and either stalls waiting for human input or fails entirely.

The workarounds that teams reach for do not hold up under scrutiny:

Shared inboxes (qa-team@company.com) introduce race conditions immediately. Ten parallel test runs create ten verification emails into one inbox. There is no clean way to correlate which code belongs to which session.

Gmail IMAP requires OAuth2 credentials, refresh token logic, and ongoing credential management. App passwords get blocked by workspace policies. IMAP polling adds unpredictable latency. And you are parsing raw RFC 2822 MIME with all its multipart structure on your end — it breaks every few months.

Mocking the email step means your agent never exercises the real flow. When you need the agent to interact with a third-party service that controls its own email delivery, there is nothing to mock.

Manual intervention defeats the purpose. An agent that pauses for a human to check an inbox is not automating the flow; it is just moving the manual step around.

What agents actually need is a programmable inbox: created on demand, accepting real SMTP traffic, queryable via a structured interface, and gone when the job is done.

The MCP Email Tools

UnCorreoTemporal exposes exactly this infrastructure as an MCP server. The server runs over streamable-HTTP at https://uncorreotemporal.com/mcp and authenticates callers with a Bearer token (Authorization: Bearer uct_your_key_here).

It exposes six tools. Here they are with their actual input schemas, taken directly from the server implementation:

create_signup_inbox

Input:  service_name (str), ttl_minutes (int, optional)
Output: {inbox_id, email, expires_at, service_name}

Creates a live temporary inbox. The email field is the SMTP address — human-readable, like coral-tiger-17@uncorreotemporal.com. The inbox is live immediately and expires automatically at expires_at.

wait_for_verification_email

Input:  inbox_id (str), timeout_seconds (int=90), poll_interval_seconds (int=3),
        subject_contains (str, optional), from_contains (str, optional)
Output: {message_id, status, ...} or {status: "timeout"}

Blocks until a matching email arrives or the timeout is exceeded. The subject_contains and from_contains filters let you wait for a specific sender or subject line without reading everything.

get_latest_email

Input:  inbox_id (str), mark_as_read (bool=False)
Output: full message object with body_text, body_html, from_address, subject, received_at

Retrieves the most recent message in the inbox, with full body content. The server handles MIME parsing — body_text and body_html are already decoded strings.

extract_otp_code

Input:  message_text (str, optional), inbox_id (str, optional), message_id (str, optional),
        otp_length_min (int=4), otp_length_max (int=8)
Output: {otp_code, candidates}

Extracts a numeric OTP from message text. You can pass message_text directly or let the tool fetch the message itself via inbox_id + message_id.

extract_verification_link

Input:  message_text (str, optional), inbox_id (str, optional), message_id (str, optional),
        preferred_domains (list[str], optional)
Output: {verification_link, candidates}

Extracts the most likely verification URL from message text. Use preferred_domains to bias toward a specific service's domain when multiple links are present.

complete_signup_flow

Input:  service_name (str), timeout_seconds (int=90), ttl_minutes (int, optional),
        subject_contains (str, optional), from_contains (str, optional),
        preferred_domains (list[str], optional)
Output: {status, email, verification_link, otp_code, verification_message, ...}

An end-to-end composite tool: creates the inbox, waits for the email, extracts both the verification link and the OTP in a single call. Returns status: "success" when at least one of the two was found, status: "partial_success" if the email arrived but neither could be extracted, or status: "timeout" if nothing came through.

This is the key tool for agents. Instead of orchestrating five separate calls, the agent calls one tool and gets back everything it needs to complete a verification flow.

Using Email as a Tool in Colab

Google Colab gives you a Python runtime that is already connected to the internet and has pip available. The mcp Python package supports streamable-HTTP transport, which means you can connect to the UnCorreoTemporal MCP server directly from a Colab cell.

Install the client:

# Cell 1
!pip install mcp httpx -q

Connect to the server:

# Cell 2
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

UCT_API_KEY = "uct_your_key_here"  # get one at uncorreotemporal.com
MCP_URL = "https://uncorreotemporal.com/mcp"

async def list_tools():
    async with streamablehttp_client(
        url=MCP_URL,
        headers={"Authorization": f"Bearer {UCT_API_KEY}"}
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            return [t.name for t in tools.tools]

tools = asyncio.run(list_tools())
print(tools)
# ['create_signup_inbox', 'wait_for_verification_email', 'get_latest_email',
#  'extract_otp_code', 'extract_verification_link', 'complete_signup_flow']

The agent discovers what is available without any hardcoded knowledge. Now use the composite tool:

# Cell 3 — create an inbox ready for a signup flow
import json

async def create_inbox(service_name: str):
    async with streamablehttp_client(
        url=MCP_URL,
        headers={"Authorization": f"Bearer {UCT_API_KEY}"}
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool(
                "create_signup_inbox",
                {"service_name": service_name, "ttl_minutes": 15}
            )
            return json.loads(result.content[0].text)

inbox = asyncio.run(create_inbox("GitHub"))
print(f"Inbox address: {inbox['email']}")
print(f"Expires at:    {inbox['expires_at']}")
# Inbox address: coral-tiger-17@uncorreotemporal.com
# Expires at:    2026-03-20T10:45:00Z

Use the address to register with the target service, then retrieve everything in one call:

# Cell 4 — after triggering the signup email, call complete_signup_flow

async def run_complete_flow(service_name: str):
    async with streamablehttp_client(
        url=MCP_URL,
        headers={"Authorization": f"Bearer {UCT_API_KEY}"}
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # One call: creates inbox, waits for email, extracts link + OTP
            result = await session.call_tool(
                "complete_signup_flow",
                {
                    "service_name": "GitHub",
                    "timeout_seconds": 90,
                    "ttl_minutes": 15,
                }
            )
            return json.loads(result.content[0].text)

flow = asyncio.run(run_complete_flow("GitHub"))
print(f"Status:  {flow['status']}")
print(f"Email:   {flow['email']}")
print(f"Link:    {flow['verification_link']}")
print(f"OTP:     {flow['otp_code']}")
# Status:  success
# Email:   coral-tiger-17@uncorreotemporal.com
# Link:    https://github.com/users/confirm-email?token=eyJh...
# OTP:     None  (GitHub uses links, not OTPs)

The complete_signup_flow tool returns both verification_link and otp_code — the caller does not need to know which one the service uses. If it is an OTP-based service, otp_code will be populated. If it sends a link, verification_link will be populated. The agent handles whichever is non-null.

MCP vs REST: The Design Difference

Here is the same flow written against the REST API directly:

# REST approach
import requests, re, time

BASE = "https://uncorreotemporal.com/api/v1"
HEADERS = {"Authorization": "Bearer uct_your_key_here"}

# Step 1: create inbox
resp = requests.post(f"{BASE}/mailboxes", params={"ttl_minutes": 15}, headers=HEADERS)
data = resp.json()
address = data["address"]
session_token = data["session_token"]

# Step 2: poll for messages (manual loop)
msg_headers = {"X-Session-Token": session_token}
deadline = time.time() + 90
message = None

while time.time() < deadline:
    msgs = requests.get(
        f"{BASE}/mailboxes/{address}/messages",
        headers=msg_headers
    ).json()
    if msgs:
        full = requests.get(
            f"{BASE}/mailboxes/{address}/messages/{msgs[0]['id']}",
            headers=msg_headers
        ).json()
        message = full
        break
    time.sleep(3)

# Step 3: extract link manually
if message:
    match = re.search(r"https?://[^\s]+verify[^\s]+", message["body_text"])
    link = match.group(0) if match else None

This works, and it is what a Colab notebook using raw requests looks like. But notice what the developer had to build:

A polling loop with timeout management
Session token tracking per inbox
Manual regex extraction for links and OTPs
Knowledge of the API endpoint structure

Compare to the MCP approach:

# MCP approach — same outcome, zero glue code
result = await session.call_tool(
    "complete_signup_flow",
    {"service_name": "GitHub", "timeout_seconds": 90, "ttl_minutes": 15}
)
flow = json.loads(result.content[0].text)
link = flow["verification_link"]

The MCP approach is not just shorter. It is structurally different in a way that matters at scale:

	REST	MCP
Tool discovery	Manual (docs, prompts)	Automatic (tools/list)
Polling	Developer implements	Server implements
Extraction logic	Developer implements regex	Tool handles OTP + link extraction
Session management	Developer tracks tokens	Server manages per-call auth
Composability	One-off integration	Reusable across all MCP runtimes
Agent integration	Custom per-framework	Standard across Claude, AutoGen, LangGraph

The REST API is the right choice when you are building a custom pipeline where you control every step. MCP is the right choice when the consumer is an agent that needs to reason about what it can do and orchestrate the steps dynamically.

Architecture: What Makes This Work

The tools describe what agents can do. The architecture explains why they are reliable.

When any service sends a verification email to a UnCorreoTemporal address, here is what happens:

External sender → AWS SES (spam/virus filtering)
                         ↓ SNS webhook
              POST /api/v1/ses/inbound
                         ↓
              core/delivery.deliver_raw_email()
               ┌──────────┴──────────┐
               ↓                     ↓
         PostgreSQL             Redis pub/sub
         (message stored)    (channel: mailbox:{address})
                                      ↓
                             WebSocket push to browser
                             /ws/inbox/{address}

Key properties:

Real SMTP delivery. In production, emails travel through AWS SES. There is no simulation, no test mode — any sender on the internet can deliver to a UnCorreoTemporal address, and it works. The wait_for_verification_email tool is waiting for messages that actually arrived via SMTP, not mocked locally.

Parsed at ingest. deliver_raw_email() parses the RFC 2822 message, extracts body_text and body_html, stores everything in PostgreSQL. By the time a tool call reads the message, the raw MIME has already been processed. The agent gets clean strings.

Ephemeral by design. Every inbox has an expires_at timestamp. A background worker soft-deletes expired mailboxes automatically. The agent creates an inbox, uses it, and never needs to clean up — it just stops calling it.

Parallel-safe. Each inbox is isolated. 50 parallel agent sessions creating 50 inboxes never share state. There are no race conditions on the inbox side.

The MCP server itself is stateless per request. Each tool call creates its own database connection, performs the operation, and closes. The server holds no inbox state between calls — everything lives in PostgreSQL.

Why This Matters for Autonomous Agents

The steady-state for AI agents is increasing autonomy over longer-horizon tasks. An agent that can write code, search the web, call APIs, and execute logic in Colab is already impressive. But it hits a ceiling the moment it needs to interact with a service that uses email as a control point.

That ceiling is structural, not technical. The agent is not missing intelligence — it is missing infrastructure. Give it a programmable inbox as a tool, and the ceiling disappears:

Autonomous account provisioning: The agent registers on external services, confirms the email, continues the workflow — no human in the loop.
CI/CD email testing: Every build creates fresh inboxes, triggers signup flows against the production login system, asserts verification emails arrived, tears down automatically.
Parallel agent runs: 20 sessions, 20 inboxes, zero shared state. Each agent operates in complete isolation.
Research workflows: An agent registers for access to a dataset, receives and processes the confirmation, downloads the resource, and the inbox expires when the job is done.

The common thread is that email is not optional in these flows. It is the coordination mechanism that external services use to verify identity and gate access. Agents that cannot handle it are excluded from a large class of real-world tasks.

The MCP interface makes this accessible without infrastructure overhead. You point your agent at the server, it discovers the tools, and email becomes a capability — not a special case.

Conclusion

Google Colab MCP and similar execution environments are expanding what agents can do. But code execution is only one piece of real-world automation. Email is another, and it has been the missing piece for longer than it should have been.

The UnCorreoTemporal MCP server gives agents six tools for handling email flows: create an inbox, wait for a message, read its content, extract an OTP, extract a verification link, or do all of it in one call with complete_signup_flow. The server runs over streamable-HTTP and works with any MCP-compatible runtime — including a Colab cell with the mcp Python library installed.

The infrastructure is real: SMTP delivery via AWS SES, PostgreSQL-backed message storage, Redis pub/sub for real-time events, ephemeral inboxes that expire automatically. There is no simulation. The emails that arrive are emails that actually traveled across the internet.

If you are building agents that need to interact with real-world services, the inbox is one call_tool away. Start at uncorreotemporal.com.

Give Your AI Agent a Real Email Inbox in Google Colab

Francisco Perez — Thu, 19 Mar 2026 19:13:17 +0000

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:

An email address it controls, created on demand
Real SMTP delivery — not simulated, not mocked
A programmatic way to read incoming messages
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"

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

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")

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}")

# 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)

# 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")

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"]

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": "..."}

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" }
    }
  }
}

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.

End-to-End Testing with Disposable Email and Playwright

Francisco Perez — Wed, 18 Mar 2026 23:32:19 +0000

The Problem Nobody Wants to Talk About

Most E2E test suites cover the happy path: fill a form, click a button, assert a result. But the moment you need to test a signup flow that ends with an email — a verification link, an OTP, a welcome message — things break down fast.

The flow is deceptively simple in production: user submits email → system sends message → user clicks link → access granted. In a test environment, that middle step is a black box. Your Playwright test has no way to receive email natively. And end-to-end testing email is one of those problems that teams defer until a critical bug slips past QA.

The common workarounds:

Mocking the email sender — your tests pass, but you've never verified that the real SMTP path works.
A shared test inbox — fragile, pollutes state across test runs, requires cleanup discipline nobody enforces.
Hardcoded OTP bypass codes — shipping test backdoors into production code.
Internal Mailhog/Mailpit setups — one more service to maintain, and they don't test real SMTP delivery.

Each of these introduces a different kind of lie into your test suite. You're testing the logic, not the system.

Why Traditional Approaches Fail

Mocking doesn't validate delivery

When you mock your email sender (replacing SMTP with a no-op or a spy), you verify that your application called the send function. You don't verify that the email was actually generated, that the template rendered correctly, or that the SMTP configuration is valid.

Real failures — DNS misconfiguration, SPF rejections, broken HTML templates — only surface when you actually send mail. Mocking hides all of this.

Shared inboxes make tests non-deterministic

Multiple tests running in parallel against the same inbox create race conditions. Test A registers, test B registers, both wait for email — and the first email that arrives gets claimed by whichever poll runs first. Flaky tests are the result.

Maintenance overhead compounds

Internal email servers (Mailhog, Mailpit, smtp4dev) need to be deployed, versioned, and kept in sync with your test infrastructure. They have their own APIs, different from production. Your test helpers diverge from real-world behavior. Over time, they become a source of friction rather than a solution.

The Better Approach: Disposable Email + Real Infrastructure

The right model is: one temporary inbox per test, provisioned via API, receiving real SMTP delivery, automatically expiring after the test completes.

This is identical to how production users interact with your system — they have an email address, your app sends to it, they receive it. The only difference is that the inbox is created programmatically, scoped to one test run, and deleted (or allowed to expire) afterward.

A temporary email API lets you:

Create an isolated inbox with a single HTTP call
Poll or stream for incoming messages
Extract OTP codes and verification links from message bodies
Clean up after each test without side effects

No shared state. No mocking. No bypass codes.

Architecture Overview

The infrastructure behind this involves several components working in sequence.

Inbox creation happens via a REST API call. The API generates a unique address in the format <adjective>-<noun>-<number>@<domain> (e.g., crisp-falcon-77@uncorreotemporal.com), stores it in PostgreSQL with an expiration timestamp, and returns the address to the caller.

Email ingestion uses an aiosmtpd-based SMTP server running alongside the API. When mail arrives for a known domain, the handler parses the raw RFC 2822 bytes, extracts headers and body (plaintext and HTML), stores the message in the database, and publishes a new_message event to a Redis pub/sub channel keyed by inbox address. Invalid recipients are silently accepted — returning 250 regardless prevents retry storms from external senders.

Retrieval supports two patterns: polling via REST (GET /api/v1/mailboxes/{address}/messages) and real-time push via WebSocket (WS /ws/inbox/{address}?api_key=<key>). The WebSocket stream emits {"event": "new_message", "message_id": "<uuid>"} when email arrives. The full message body — body_text, body_html, attachments metadata — is fetched separately.

Expiration runs as a background task inside the API process, executing every 60 seconds. Inboxes where expires_at <= now() are soft-deleted. Messages are retained for the configured retention period; the inbox simply stops accepting new mail.

Step-by-Step: Testing with Playwright

Let's walk through a complete Playwright email testing flow using the REST API.

Create a Temporary Inbox

Before your test starts, provision a dedicated inbox:

const API_KEY = process.env.UCT_API_KEY; // uct_<token>
const BASE_URL = "https://uncorreotemporal.com/api/v1";

async function createInbox(ttlMinutes = 15) {
  const res = await fetch(`${BASE_URL}/mailboxes`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-API-Key": API_KEY,
    },
    body: JSON.stringify({ ttl_minutes: ttlMinutes }),
  });
  const data = await res.json();
  return data; // { address, expires_at }
}

The returned address is what you feed into your signup form. No hardcoded emails.

Trigger the Signup Flow with Playwright

test("signup email verification", async ({ page }) => {
  const { address } = await createInbox();

  await page.goto("https://yourapp.com/signup");
  await page.fill('[name="email"]', address);
  await page.fill('[name="password"]', "Test1234!");
  await page.click('[type="submit"]');

  await page.waitForSelector(".signup-success");
  // Now wait for the verification email
});

At this point, your application has sent a verification email to the temporary inbox. Real SMTP delivery, real message in the queue.

Wait for the Email to Arrive

Use a polling helper with a timeout:

async function waitForEmail(address, { timeout = 30000, interval = 2000 } = {}) {
  const deadline = Date.now() + timeout;

  while (Date.now() < deadline) {
    const res = await fetch(`${BASE_URL}/mailboxes/${address}/messages`, {
      headers: { "X-API-Key": API_KEY },
    });
    const { messages } = await res.json();

    if (messages.length > 0) {
      const msgRes = await fetch(
        `${BASE_URL}/mailboxes/${address}/messages/${messages[0].id}`,
        { headers: { "X-API-Key": API_KEY } }
      );
      return msgRes.json();
    }

    await new Promise((r) => setTimeout(r, interval));
  }

  throw new Error(`No email arrived within ${timeout}ms`);
}

Extract the OTP or Verification Link

function extractVerificationLink(message) {
  const body = message.body_text || "";
  const match = body.match(/https:\/\/yourapp\.com\/verify\?token=[a-zA-Z0-9_-]+/);
  if (!match) throw new Error("Verification link not found in email body");
  return match[0];
}

function extractOTP(message) {
  const body = message.body_text || "";
  const match = body.match(/\b(\d{6})\b/);
  if (!match) throw new Error("OTP not found in email body");
  return match[1];
}

Complete the Verification Flow

  const email = await waitForEmail(address);
  const verificationLink = extractVerificationLink(email);

  await page.goto(verificationLink);
  await expect(page.locator(".verified-badge")).toBeVisible();
});

Full End-to-End Script

import { test, expect } from "@playwright/test";

const API_KEY = process.env.UCT_API_KEY;
const BASE = "https://uncorreotemporal.com/api/v1";

async function apiFetch(path, opts = {}) {
  return fetch(`${BASE}${path}`, {
    ...opts,
    headers: { "X-API-Key": API_KEY, "Content-Type": "application/json", ...opts.headers },
  }).then((r) => r.json());
}

async function createInbox() {
  return apiFetch("/mailboxes", { method: "POST", body: JSON.stringify({ ttl_minutes: 15 }) });
}

async function waitForEmail(address, timeout = 30000) {
  const deadline = Date.now() + timeout;
  while (Date.now() < deadline) {
    const { messages } = await apiFetch(`/mailboxes/${address}/messages`);
    if (messages?.length > 0) return apiFetch(`/mailboxes/${address}/messages/${messages[0].id}`);
    await new Promise((r) => setTimeout(r, 2000));
  }
  throw new Error("Email timeout");
}

async function deleteInbox(address) {
  await fetch(`${BASE}/mailboxes/${address}`, { method: "DELETE", headers: { "X-API-Key": API_KEY } });
}

test("full signup + email verification", async ({ page }) => {
  const { address } = await createInbox();

  try {
    await page.goto("https://yourapp.com/signup");
    await page.fill('[name="email"]', address);
    await page.fill('[name="password"]', "Secure1234!");
    await page.click('[type="submit"]');
    await page.waitForSelector(".check-your-email");

    const email = await waitForEmail(address);
    const [link] = email.body_text.match(/https:\/\/yourapp\.com\/verify\?token=\S+/) || [];
    if (!link) throw new Error("No verification link");

    await page.goto(link);
    await expect(page.locator("h1")).toHaveText("Account verified");
  } finally {
    await deleteInbox(address);
  }
});

CI/CD Integration

This pattern fits naturally into any CI pipeline. Set UCT_API_KEY as a secret, and your tests run identically in local, staging, and CI environments.

# .github/workflows/e2e.yml
- name: Run E2E tests
  env:
    UCT_API_KEY: ${{ secrets.UCT_API_KEY }}
  run: npx playwright test

Key benefits for QA teams:

No shared infrastructure to provision — the email backend is external, always available.
Parallel-safe — each test creates its own inbox; no inbox is shared across test runs.
Test reports include real email content — when a test fails, you can inspect the actual message that arrived (or didn't).

Best Practices

Set aggressive timeouts in CI. Transactional email delivery in a test environment should be fast (under 5s). Set your polling timeout to 20–30 seconds and treat any test that takes longer as a potential infrastructure issue, not a Playwright issue.

Retry on network errors, not on missing email. Distinguish between "the API call failed" (retry) and "no email arrived" (fail the test). Swallowing network errors in your wait loop masks real problems.

Create one inbox per test, not one per suite. Sharing an inbox across multiple tests in a suite reintroduces the shared-state problem. The inbox creation API call is fast — there's no meaningful performance tradeoff.

Clean up explicitly. Even though inboxes expire automatically, DELETE /api/v1/mailboxes/{address} in a finally block keeps your account tidy and reduces noise if you're monitoring active inbox counts.

Store the inbox address in test context. If you use Playwright fixtures, create the inbox in a fixture and attach the address to the test context. This makes it easy to log the address when a test fails, so you can inspect the inbox state during debugging.

Conclusion

Email flows are a first-class part of most user journeys — and they deserve first-class test coverage. Mocking the email layer trades realism for convenience, and eventually that tradeoff costs you. A real email lands in a real inbox, and your test suite should be able to verify that.

The approach described here — temporary inbox provisioned per test, real SMTP delivery, polling or WebSocket retrieval, explicit cleanup — is production-grade and CI-ready. It requires no additional infrastructure on your end and no changes to your application code.

If you want to try this approach, you can use a temporary email API like uncorreotemporal.com to get started.

Deploying an Email Infrastructure on AWS

Francisco Perez — Tue, 17 Mar 2026 03:25:51 +0000

How we built a production-ready, programmable temporary email platform using FastAPI, aiosmtpd, Redis, and Terraform.

1. Introduction

Receiving emails programmatically is deceptively hard. Sending email is a solved problem — you call an API, provide credentials, and a third-party handles everything. Receiving email is a different challenge entirely.

To receive email, you need a listening SMTP server, an MX DNS record pointing at it, a strategy for spam filtering, a mechanism to parse RFC 2822 messages reliably, a way to route incoming messages to the right user or process, and a real-time notification layer so that waiting clients know when something arrived. And all of that has to survive production traffic without a single SMTP connection timing out.

This article walks through the full architecture of uncorreotemporal.com, a programmable temporary email platform. The platform lets developers and automation systems create disposable inboxes, receive real emails, and access them via a REST API, WebSocket stream, or MCP server (for AI agent workflows). Every claim here is grounded in the actual codebase.

2. The Problem with Email Infrastructure

Before diving into the implementation, it is worth naming the specific challenges that make email ingestion infrastructure hard to get right.

SMTP is a stateful, connection-oriented protocol. Unlike an HTTP request that fails fast, an SMTP session involves a multi-step handshake (EHLO, MAIL FROM, RCPT TO, DATA). If anything in that chain fails or is too slow, the sending server retries — often for hours or days. Your ingestion server must respond correctly and quickly, or legitimate senders will conclude your domain is broken.

Deliverability and reception are different concerns. SPF and DKIM records affect whether outbound email is trusted. For inbound email, the critical record is your MX record, and the critical decision is who handles your SMTP traffic — a real server that speaks the full protocol, or a managed relay like AWS SES.

Real-time processing is expected. Users of temporary email systems expect messages to appear instantly. Polling a database every few seconds is not acceptable. This drives the need for a pub/sub layer that can push events to waiting WebSocket clients the moment a message is written to storage.

Inbox systems are write-heavy with unpredictable bursts. A campaign sending 10,000 test emails, or a CI/CD pipeline spinning up hundreds of parallel verification flows, can produce sudden spikes that need to be absorbed cleanly.

3. High-Level Architecture

The system is composed of four layers:

[External Sender]
       │
       ▼
[AWS SES Inbound] ──── SNS Topic ──── HTTPS Webhook
       │                                     │
       │                           [FastAPI API Server]
       │                                     │
       └───────────────────────────► [PostgreSQL (RDS)]
                                             │
                                         [Redis]
                                             │
                                     [WebSocket Clients]

In development, a local aiosmtpd SMTP server replaces AWS SES. In production, SES handles the SMTP session, runs spam and virus checks, and forwards messages to an SNS topic which triggers a webhook on the FastAPI server. This design decouples SMTP availability (a managed AWS concern) from application availability (our concern).

The FastAPI application is a single async process running on a t3.medium EC2 instance behind an Application Load Balancer. PostgreSQL (RDS) provides durable storage. Redis provides the pub/sub channel that bridges email delivery to WebSocket clients.

4. AWS Infrastructure Design

All infrastructure is defined with Terraform, organized into modules: vpc, ec2, rds, redis, alb, acm, ses, route53, and ecr.

VPC Layout

The VPC spans two availability zones with public and private subnets. A NAT gateway per AZ provides high-availability outbound access for private instances. The EC2 instance hosting the application lives in a public subnet (accessible via the ALB), while RDS and ElastiCache live in private subnets.

# terraform/modules/vpc/main.tf (abbreviated)
resource "aws_vpc" "main" {
  cidr_block           = "10.0.0.0/16"
  enable_dns_hostnames = true
}

EC2 for Application Hosting

A single t3.medium EC2 instance (Ubuntu 22.04) runs the full Docker Compose stack — FastAPI API and Nginx frontend. An Elastic IP is assigned so the address is stable across instance stops and restarts. Security groups allow inbound traffic only from the ALB on port 8000 and from trusted CIDR blocks for SSH.

The application is deployed as a Docker image pulled from ECR:

uncorreotemporal-api   → FastAPI + Uvicorn (port 8000, internal)
uncorreotemporal-front → Nginx + Angular SPA (port 80, external)

Docker networking puts both containers on a shared bridge (uncorreotemporal_net). Nginx inside the front container handles TLS termination (via the ALB), then proxies API traffic to api:8000.

RDS for Persistence

The Terraform rds module provisions a db.t3.small PostgreSQL 15 instance with Multi-AZ enabled, 50 GB of encrypted gp3 storage, 7-day automated backups, and deletion protection. The database is not exposed to the internet — only the EC2 security group has access on port 5432.

The application uses asyncpg as the PostgreSQL driver, wrapped by SQLAlchemy's async ORM:

# db/session.py
engine = create_async_engine(
    settings.async_database_url,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession)

pool_pre_ping=True ensures stale connections (common after RDS failovers) are detected and recycled before being handed to a request handler.

Redis for Real-Time Events

In production, ElastiCache runs a replication group with two cache.t3.small nodes, Multi-AZ failover enabled, and TLS enforced. The application connects using rediss:// when redis_url starts with that scheme:

# core/redis_client.py
_client = redis.Redis.from_url(
    settings.redis_url,   # rediss:// in prod
    decode_responses=True,
)

Redis is used exclusively for pub/sub — it is not a job queue or a cache. Every time a message is delivered to a mailbox, a single publish call fires on a per-mailbox channel. This design is intentionally lightweight.

Application Load Balancer and HTTPS

The ALB module provisions a public-facing load balancer with an HTTP listener (port 80) that issues a 301 redirect to HTTPS, and an HTTPS listener (port 443) that terminates TLS using an ACM certificate and forwards to the EC2 instance on port 8000. The health check hits GET /health on the FastAPI app.

The ALB provides TLS offload, so the application never needs to manage certificates. ACM handles automatic renewal via Route 53 DNS validation.

Route 53 and DNS Records

The route53 module creates the hosted zone and all records needed:

Record	Type	Purpose
`uncorreotemporal.com`	A (alias)	Points to the ALB
`uncorreotemporal.com`	MX	Routes inbound email to SES
`uncorreotemporal.com`	TXT	SPF policy
`mail.uncorreotemporal.com`	MX/TXT	Custom MAIL FROM for SES
`_domainkey`	CNAME	DKIM tokens from SES

The MX record pointing to SES inbound endpoints is what makes the domain receivable. Without it, external senders cannot locate the mail server.

SES and Email Deliverability

AWS SES handles the full inbound SMTP session. The ses Terraform module creates a domain identity, requests DKIM tokens, configures a custom MAIL FROM subdomain, creates an SNS topic and receipt rule. The SNS topic has an HTTPS subscription pointing at the FastAPI webhook endpoint.

When the application server boots and processes a SubscriptionConfirmation message from SNS, it calls the confirmation URL, activating the subscription. This handshake must succeed before any inbound email flows through.

5. Email Ingestion Pipeline

Once DNS, SES, and SNS are configured, every email sent to @uncorreotemporal.com follows this path:

External sender → SES SMTP endpoint → spam/virus check
    → SNS topic → HTTPS POST /api/v1/ses/inbound
    → core/delivery.deliver_raw_email()
    → Mailbox lookup + quota check
    → Parse RFC 2822 message
    → INSERT into messages table
    → Redis PUBLISH mailbox:{address}
    → WebSocket clients receive notification

The SES Webhook Handler

# api/routers/ses_inbound.py
@router.post("/api/v1/ses/inbound")
async def ses_inbound(request: Request, db: AsyncSession = Depends(get_db)):
    body = await request.json()
    msg_type = body.get("Type")

    if msg_type == "SubscriptionConfirmation":
        async with httpx.AsyncClient() as client:
            await client.get(body["SubscribeURL"])
        return {"status": "confirmed"}

    if msg_type == "Notification":
        payload = json.loads(body["Message"])
        raw_email = base64.b64decode(payload["content"])
        recipient = payload["receipt"]["recipients"][0]
        await deliver_raw_email(raw_email, recipient.lower(), db)

    return {"status": "ok"}

The endpoint always returns 200 OK regardless of internal errors. If it returned a 5xx, SNS would retry for up to 23 days, creating a storm of duplicate delivery attempts.

The Delivery Core

The deliver_raw_email function in core/delivery.py is the single entry point for both the SES webhook and the development SMTP handler:

async def deliver_raw_email(raw: bytes, address: str, db: AsyncSession) -> bool:
    mailbox = await get_active_mailbox(address, db)
    if not mailbox:
        return False

    plan = await get_user_active_plan(mailbox.owner, db)
    if not await check_message_quota(mailbox, plan, db):
        return False

    parsed = parse_email(raw)  # core/parser.py — RFC 2822

    message = Message(
        mailbox_id=mailbox.id,
        from_address=parsed.from_address,
        subject=parsed.subject,
        body_text=parsed.text,
        body_html=parsed.html,
        raw_email=raw,
        attachments=parsed.attachments,
    )
    db.add(message)
    await db.commit()

    await redis_client.publish(f"mailbox:{address}", json.dumps({
        "event": "new_message",
        "message_id": str(message.id),
    }))
    return True

The raw RFC 2822 bytes are stored alongside the parsed fields. This makes it possible to re-parse messages if the extraction logic improves, without touching the source data.

Development: aiosmtpd

In development, the smtp Docker service runs an aiosmtpd server bound to port 25:

# smtp/handler.py
class MailHandler:
    async def handle_DATA(self, server, session, envelope):
        raw = envelope.content
        for rcpt in envelope.rcpt_tos:
            if rcpt.lower().endswith("@uncorreotemporal.com"):
                await deliver_raw_email(raw, rcpt.lower(), db)
        return "250 Message accepted for delivery"

6. Handling Real-Time Email Events

The WebSocket endpoint subscribes to the Redis channel for the requested mailbox. Two concurrent async tasks run for each connected client:

# ws/inbox.py
@router.websocket("/ws/inbox/{address}")
async def websocket_inbox(websocket: WebSocket, address: str, ...):
    await websocket.accept()
    pubsub = await redis_client.subscribe(f"mailbox:{address}")

    async def send_loop():
        async for message in pubsub.listen():
            if message["type"] == "message":
                await websocket.send_text(message["data"])

    async def ping_loop():
        while True:
            await asyncio.sleep(30)
            await websocket.send_json({"event": "ping"})

    tasks = [
        asyncio.create_task(send_loop()),
        asyncio.create_task(ping_loop()),
    ]
    try:
        await asyncio.gather(*tasks)
    except WebSocketDisconnect:
        for t in tasks:
            t.cancel()
        await pubsub.unsubscribe(f"mailbox:{address}")

The ping loop prevents proxy and load balancer timeouts on idle connections. When deliver_raw_email calls redis.publish, the message arrives in send_loop within milliseconds. The client receives only a message ID — it then fetches full content from the REST API. This keeps the WebSocket channel thin and stateless.

7. Security Considerations

API Key Hashing

API keys are never stored in plaintext. On creation, the server generates a random key in the format uct_<32 URL-safe base64 chars>, stores its SHA-256 hash, and returns the raw key once to the user:

# models/api_key.py
raw_key = "uct_" + secrets.token_urlsafe(32)
key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
key_prefix = raw_key[:12]  # for UI display only

Anonymous Sessions

Anonymous mailboxes are protected by a 32-byte session token returned once at creation. Each anonymous mailbox is its own authorization boundary — no shared credentials.

SNS Signature Verification

The SES inbound webhook validates the SNS message signature using RSA-SHA1 before processing the payload. This prevents arbitrary HTTP requests from injecting fake emails.

Abuse Prevention Through Quotas

Plan-based quotas (max_mailboxes, max_messages_per_mailbox, max_ttl_minutes) are enforced in the delivery path and at mailbox creation time. Anonymous users receive the most restrictive limits.

8. Infrastructure as Code with Terraform

The entire AWS footprint is defined in Terraform, split into discrete modules:

terraform/
├── main.tf
├── variables.tf
├── outputs.tf
└── modules/
    ├── vpc/
    ├── ec2/
    ├── rds/
    ├── redis/
    ├── alb/
    ├── acm/
    ├── ses/
    ├── route53/
    ├── ecr/
    └── iam/

Terraform state is stored in an S3 backend. One non-obvious decision: the ses module does not create the SNS HTTPS subscription automatically — it creates the topic and outputs its ARN. The subscription is confirmed at runtime when the application first boots and handles the SubscriptionConfirmation request from SNS.

9. Scaling the System

Background Workers and the Single-Worker Constraint

The mailbox expiry background task runs inside the FastAPI process as an asyncio task:

# core/expiry.py — runs every 60 seconds
async def run_expiry_loop():
    while True:
        await asyncio.sleep(60)
        async with AsyncSessionLocal() as db:
            expired = await db.execute(
                select(Mailbox).where(
                    Mailbox.expires_at <= datetime.utcnow(),
                    Mailbox.is_active == True,
                )
            )
            for mailbox in expired.scalars():
                mailbox.is_active = False
            await db.commit()

This works correctly only with uvicorn --workers 1. With multiple workers, every process runs the expiry loop independently causing duplicate writes. A dedicated worker process or cron job is the right solution for horizontal scaling.

Database Indexing

The messages table has a composite index on (mailbox_id, received_at). The mailboxes table has indexes on (expires_at, is_active) and (owner_id, is_active). These were added proactively based on known access patterns.

Redis Throughput

Each message delivery issues one PUBLISH command; each WebSocket connection issues one SUBSCRIBE. At high scale, channels can be sharded by mailbox address across a Redis Cluster.

10. Lessons Learned

Delegate SMTP, own the processing. Running a production SMTP server that survives IP reputation issues is a full-time job. AWS SES handles the protocol layer; a webhook handles the processing.

Store raw email bytes. Parsing RFC 2822 has real failure modes. Storing original bytes alongside parsed fields means parsing bugs are recoverable without data loss.

Make the delivery path the single entry point. Both the development SMTP handler and the production SES webhook call the same deliver_raw_email function.

Subscriptions as the single source of truth. Migration 0005 removed plan_id from the users table. The effective plan is now always derived at query time from the subscriptions table — eliminating an entire class of data consistency bugs.

Async all the way down. FastAPI, asyncpg, aioredis, and aiosmtpd all run on the same event loop. A blocking call in one coroutine stalls all connected WebSocket clients.

11. Conclusion

Programmable email infrastructure is genuinely useful for CI/CD testing (real inbox per test run), automation workflows, and AI agents (the MCP server exposes create/read/delete as tools an agent can invoke without human intervention).

The architecture described here — FastAPI email backend, async ingestion pipeline, Redis pub/sub, Terraform infrastructure on AWS — is production-deployed and handling real traffic. Visit uncorreotemporal.com to try the API.

Automating Email Verification Flows with MCP and Claude

Francisco Perez — Sun, 15 Mar 2026 16:25:50 +0000

Introduction

Most LLM agents can call REST APIs, browse the web, write and run code, and coordinate multi-step tasks. But there is one failure mode that surfaces reliably in real-world automation: email verification.

The pattern shows up everywhere. A service sends a six-digit OTP before granting access. A SaaS requires clicking a confirmation link before an account activates. A CI/CD job provisions a test user and then stalls because it has nowhere to receive the token. In each case, the automation is fully capable of handling every step of the workflow except the one that goes through an inbox.

This article is a concrete walkthrough of how Claude — using the Model Context Protocol and the uncorreotemporal.com MCP server — can complete these flows end to end. The focus is on the mechanics: what tool calls Claude makes, how it polls for incoming messages, how it extracts a verification link or OTP from email body text, and what guard rails keep the system from becoming a liability.

The Problem with Email Verification Automation

Email verification is designed to slow down machines. The implicit assumption behind every "check your inbox" prompt is that a human is on the other end of the email address — that access is being granted to a person with a real mailbox, not a script with a temporary address.

That assumption breaks clean automation in several ways:

No programmatic access by default. Receiving email requires SMTP infrastructure: a domain with MX records, a server on port 25, and some mechanism to query delivered messages. None of that comes out of the box when you spin up a test environment.

Real inboxes accumulate state. Using a shared QA mailbox like qa@company.com means parallel test runs write into the same inbox simultaneously. There is no clean way to correlate an OTP from a specific test run when ten jobs are executing at once.

OAuth-based inbox access is heavyweight. Reading Gmail programmatically requires an OAuth 2.0 flow, scoped permissions, token refresh logic, and ongoing credential management. For a CI job that runs every commit, the maintenance cost is disproportionate to the task.

Agents have no "inbox" concept. LLM agents are built around function calls and HTTP. There is no standard way to tell an agent "wait here until a specific email arrives." The agent has to know both that email is happening and how to interact with it — and that knowledge is not baked into any standard framework.

The result is that email verification is the last manual step in workflows that are otherwise fully automated. Someone has to check the inbox.

Making Email Programmable

Temporary email infrastructure flips the model. Instead of an inbox that exists independently and happens to receive email, you get an inbox created on demand, with a fixed lifetime, accessible via a structured API.

The uncorreotemporal.com stack implements this with real SMTP delivery. In production, email arrives via AWS SES, which delivers to an SNS topic that POSTs to a webhook at /api/v1/ses/inbound. The webhook validates the SNS signature, runs the SES spam/virus verdicts, parses the raw RFC 2822 message, and calls deliver_raw_email() — which inserts a Message row into PostgreSQL and publishes a Redis event to mailbox:{address}.

The result is a live inbox that can receive actual SMTP traffic from any sender on the internet, queryable via REST, and linked to a user account via an API key.

For agents specifically, the MCP server is the right entry point. Not because REST is insufficient, but because MCP lets Claude discover and invoke these capabilities without any custom integration code — just a server configuration and an API key.

MCP Servers and AI Tooling

The Model Context Protocol defines a standard interface for giving AI agents access to external tools. An MCP server exposes a named set of tools with typed input schemas. Any MCP-compatible runtime — Claude Desktop, Cursor, AutoGen, LangGraph — can call tools/list to discover what is available and tools/call to invoke them, with no adapter code per framework.

The uncorreotemporal MCP server runs as a subprocess over stdio. Its configuration in claude_desktop_config.json looks like this:

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

The server exposes five tools: create_mailbox, list_mailboxes, get_messages, read_message, and delete_mailbox. Each maps directly to a database operation — the MCP layer is thin on purpose. There is no HTTP hop between the MCP server and the database; mcp/server.py imports and calls SQLAlchemy model operations directly, with its own AsyncSessionLocal() per call.

Authentication happens once. On the first tool call, _get_user_id() hashes UCT_API_KEY with SHA-256, queries the ApiKey table, resolves the associated User, and caches _authenticated_user_id in module memory. Every subsequent call in the same server process skips the auth lookup entirely.

Creating Inboxes from an AI Agent

When Claude needs an inbox, it calls create_mailbox. The call is minimal:

{
  "name": "create_mailbox",
  "arguments": { "ttl_minutes": 15 }
}

The server resolves the user's active plan from their subscription table (no plan_id FK on the users row — plan is derived from the active subscription, fallback to free). The TTL is clamped to [1, plan.max_ttl_minutes], an expires_at timestamp is calculated, and the address is generated using generate_friendly_address() — an adjective-noun-number@domain format that produces human-readable names like coral-tiger-17@uncorreotemporal.com.

The mailbox is tagged owner_type=mcp in the database — structurally identical to an API-created mailbox, but distinguishable in audit logs. The tool returns:

{
  "address": "coral-tiger-17@uncorreotemporal.com",
  "expires_at": "2026-03-15T10:45:00Z",
  "ttl_minutes": 15
}

The address is live immediately. Any SMTP sender can deliver to it.

Detecting Incoming Emails

The system has two delivery notification mechanisms. Browser clients use WebSocket, subscribing to a Redis pub/sub channel keyed mailbox:{address}. When deliver_raw_email() inserts a new message, it publishes {"event": "new_message", "message_id": "<uuid>"} to that channel. The WebSocket client receives the push event within milliseconds.

The MCP tools do not expose the WebSocket channel. Agents poll.

Polling with get_messages is intentionally lightweight. The tool returns only message metadata — id, from_address, subject, received_at, is_read, has_attachments — without body content. This keeps each poll response small. A typical agent loop looks like this:

# Agent polling loop (pseudocode)
for attempt in range(12):           # max ~60 seconds
    result = mcp.call("get_messages", address=address, limit=5)
    unread = [m for m in result["messages"] if not m["is_read"]]

    if unread:
        break

    time.sleep(5)
else:
    raise TimeoutError("No verification email arrived within 60 seconds")

Claude naturally writes this kind of loop when it understands the task: create an inbox, trigger the external flow, then poll until a message with a relevant subject arrives. The is_read flag on each message lets the agent filter to genuinely new arrivals without re-reading messages it has already processed.

The received_at DESC ordering on the query means the most recent message is always first — no reverse scanning needed.

Extracting Verification Links and OTP Codes

read_message fetches the full message content and marks it is_read=True:

{
  "name": "read_message",
  "arguments": {
    "address": "coral-tiger-17@uncorreotemporal.com",
    "message_id": "3f8a1c2b-9e14-4a77-b832-d47ac1f30512"
  }
}

The response includes both body_text and body_html, parsed from the raw RFC 2822 bytes stored at ingest time. For verification flows, body_text is usually sufficient — it strips HTML tags and gives the agent a clean string to reason over.

For link-based verification, Claude extracts the URL directly from body_text:

body_text: "Welcome! To confirm your account, click the link below:
https://service.com/verify?token=eyJhbGciOiJIUzI1NiJ9.abc123

This link expires in 24 hours."

Claude identifies the URL in plain text and navigates to it. No regex is required in the agent code itself — the LLM parses the text naturally. For more structured extraction in deterministic pipelines, a regex like r"https?://[^\s]+" on body_text is reliable.

For OTP codes, the pattern is slightly different. Services typically send a message with a prominent numeric code:

body_text: "Your verification code is: 847392

Do not share this code with anyone."

Claude reads the body, identifies the six-digit code, and uses it in the next step of the workflow. In a programmatic context, r"\b\d{6}\b" captures most OTP formats. For codes with different lengths or formats (XXXX-YYYY, alphanumeric), the agent's reasoning handles the variation without code changes.

The body always contains both representations — body_text for easy parsing, body_html for cases where the link is in an anchor tag and not visible in the plaintext version.

A Fully Automated Verification Flow

Here is the complete sequence, annotated with the actual MCP tool calls Claude makes:

Step 1 — Create an inbox with a short TTL

→ create_mailbox({"ttl_minutes": 10})
← {"address": "coral-tiger-17@uncorreotemporal.com", "expires_at": "...", "ttl_minutes": 10}

Step 2 — Use the address during signup

Claude fills the email field in a registration form with coral-tiger-17@uncorreotemporal.com. The exact mechanism depends on the agent's browser tool — Playwright, Selenium, or a browser-use MCP tool. The point is that the address is real and Claude owns the inbox.

Step 3 — Wait for the verification email

→ get_messages({"address": "coral-tiger-17@uncorreotemporal.com", "limit": 5})
← {"messages": []}

→ get_messages({"address": "coral-tiger-17@uncorreotemporal.com", "limit": 5})
← {
    "messages": [{
      "id": "3f8a1c2b-...",
      "from_address": "noreply@service.com",
      "subject": "Confirm your account",
      "received_at": "2026-03-15T10:35:44Z",
      "is_read": false,
      "has_attachments": false
    }]
  }

Step 4 — Read the message and extract the token

→ read_message({"address": "coral-tiger-17@uncorreotemporal.com", "message_id": "3f8a1c2b-..."})
← {
    "from_address": "noreply@service.com",
    "subject": "Confirm your account",
    "body_text": "Click to confirm: https://service.com/verify?token=abc123\n\nExpires in 1 hour.",
    "body_html": "<p>Click <a href=\"https://service.com/verify?token=abc123\">here</a>...</p>",
    "attachments": []
  }

Claude extracts https://service.com/verify?token=abc123 from body_text.

Step 5 — Complete the verification

Claude navigates to the extracted URL (or submits the OTP in a form). The account is now verified.

Step 6 — Clean up

→ delete_mailbox({"address": "coral-tiger-17@uncorreotemporal.com"})
← {"success": true, "address": "coral-tiger-17@uncorreotemporal.com"}

The entire flow — inbox creation to account verification — runs in under 30 seconds with no human in the loop.

Why AI Agents Need Real Email Infrastructure

The alternative to a live SMTP inbox is interception. Some test setups run a local SMTP server that captures outbound messages before they leave the network. This works for controlled environments but breaks for any flow involving a third-party service, because those services send to a real domain and expect actual delivery.

Agents working with real-world systems — creating accounts on external services, automating procurement workflows, testing SaaS integrations, running compliance checks — need inboxes that receive genuine SMTP traffic. A fake inbox that captures messages internally is only useful if you control both the sender and the recipient.

There is also a composability argument. An agent that uses a temporary inbox via MCP requires no special infrastructure knowledge. It calls create_mailbox, gets a live address, submits it to an external service, polls get_messages until the verification email arrives, reads body_text via read_message, extracts the token, completes the flow, and cleans up with delete_mailbox.

The five-tool surface is intentionally small. There is no tool for raw SMTP access, no streaming subscription, no attachment download — just the operations an agent needs to complete a verification workflow. Complexity stays in the infrastructure layer, not in the agent's reasoning loop.

Conclusion

Email verification is a blocking dependency in a large class of real-world automation tasks. The underlying problem is structural: receiving email requires infrastructure, and most agent runtimes have no standard way to access it.

The approach described here — a real SMTP backend exposed as MCP tools — resolves this without requiring agents to understand anything about email delivery. Claude calls create_mailbox, gets a live address, submits it to an external service, polls get_messages until the verification email arrives, reads body_text via read_message, extracts the token, completes the flow, and cleans up with delete_mailbox.

As agent runtimes mature and autonomous workflows become more common, email needs to be a first-class tool — not a special case requiring a human fallback. The infrastructure to make that real exists today. The full stack is at uncorreotemporal.com.

Temporary Email for Selenium Automation

Francisco Perez — Sat, 14 Mar 2026 15:07:41 +0000

Every Selenium test that touches a signup form eventually hits the same wall: after the form is submitted, the app sends a confirmation email, and the test has no way to read it. The browser is waiting. Your test has no inbox. The flow is stuck.

This is not an edge case. Registration, password reset, magic link login, and OTP verification are among the most common flows in any web application — and among the most routinely skipped in automated test suites. This article explains why email testing breaks in Selenium, what the common workarounds cost you, and how to replace them with a programmable temporary inbox API that makes the entire flow testable in under 50 lines of Python.

Why Email Testing Breaks in Selenium Automation

Selenium gives you a browser. It can fill forms, click buttons, and assert on page content. What it cannot do is open a mail client, find an email, and extract a link from it. That gap has to be filled externally — and the common solutions all introduce friction that eventually breaks pipelines or degrades test quality.

Hardcoded inboxes. Teams often maintain a qa@company.com address and configure their staging environment to send all test emails there. This works until you run tests in parallel. Two jobs submit the signup form at the same time, and both wait for confirmation. The inbox receives both emails. Each job picks up the wrong one. Intermittent failures start appearing on Fridays and nobody can reproduce them locally.

Gmail test accounts. Some teams provision a real Gmail account and use IMAP to read emails in tests. Gmail IMAP requires OAuth2 tokens, app passwords, or "less secure app" settings — all of which expire, get blocked by workspace policies, or break when Google changes their authentication requirements. IMAP polling adds 5–30 seconds of latency. You also end up parsing raw RFC 2822 MIME messages, which is not what you want inside a Selenium test.

Mocked email systems. Mocking sendgrid.send() or patching the email provider at the service layer tells you that your code called the send function — not that the email is deliverable, not that the template renders correctly, not that the confirmation link is well-formed, not that the OTP matches what the backend expects. The test passes. The production signup flow is broken. You find out from a user.

Staging SMTP with no retrieval API. Some teams configure a local SMTP sink like MailHog or Mailtrap on staging. These work for manual inspection but become awkward in CI. MailHog has no reliable message filtering API. Mailtrap shared inboxes cause the same cross-contamination problem as hardcoded addresses. Per-test inbox isolation on Mailtrap requires a paid plan and SDK integration.

The root problem is the same in every case: the test cannot programmatically create an inbox, use that address during the Selenium session, and then retrieve the resulting email through a clean HTTP interface.

The Better Approach: Temporary Inbox APIs

A programmable temporary email API changes the model. Instead of pre-existing shared inboxes, you create a fresh inbox at the start of each test. That inbox has:

A unique @uncorreotemporal.com address ready to receive real SMTP email
A session token that authenticates all subsequent operations
A REST endpoint to list and read messages
A WebSocket stream that emits an event the moment mail arrives
An automatic expiration — no cleanup required

The inbox exists for the duration of the test and is gone after. No shared state. No cross-contamination. No authentication overhead beyond a single HTTP call.

This is not a browser-based disposable email service. It is an API-first infrastructure designed for use in code — test fixtures, CI pipelines, and automation scripts. You interact with it entirely over HTTP. Selenium never needs to know it exists.

Example Workflow with Selenium

The pattern is consistent regardless of what the application under test does with the email:

Create an inbox via the API. You get back an address and a session token.
Run the Selenium session, entering the temporary address wherever the signup form asks for email.
Poll or subscribe for incoming messages. Wait up to 30 seconds.
Retrieve the full message body using the message ID returned by the list endpoint.
Extract the verification link or OTP from body_text or body_html using a regex.
Complete the flow — either by navigating Selenium to the confirmation URL or by submitting the OTP in the browser.

Each of these steps maps to one or two HTTP calls. Selenium handles the browser. requests handles the inbox API. They never need to share state beyond the email address string.

Python Example

The following example tests a signup flow end-to-end: creating a real inbox, filling a registration form with Selenium, waiting for the confirmation email, extracting the verification link, and asserting the account is active.

import re
import time
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options

BASE_URL = "https://uncorreotemporal.com"
APP_URL = "https://your-app-staging.example.com"


# ── Step 1: Create a temporary inbox ─────────────────────────────────────────

def create_inbox(ttl_minutes: int = 10) -> tuple[str, str]:
    """
    Create a temporary inbox via the API.
    Returns (address, session_token).
    Anonymous use: no API key required.
    """
    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"]


# ── Step 2: Poll the inbox for a new message ──────────────────────────────────

def wait_for_message(
    address: str,
    session_token: str,
    timeout: int = 30,
    poll_interval: float = 2.0,
) -> dict:
    """
    Poll the message list until at least one message arrives.
    Returns the first message summary (id, subject, from_address, etc.).
    Raises TimeoutError if nothing arrives within timeout seconds.
    """
    headers = {"X-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]  # ordered by received_at desc
        time.sleep(poll_interval)

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


# ── Step 3: Fetch the full message body ───────────────────────────────────────

def get_full_message(address: str, session_token: str, message_id: str) -> dict:
    """
    Retrieve the full message content including body_text and body_html.
    Automatically marks the message as read.
    """
    headers = {"X-Session-Token": session_token}
    resp = requests.get(
        f"{BASE_URL}/api/v1/mailboxes/{address}/messages/{message_id}",
        headers=headers,
    )
    resp.raise_for_status()
    return resp.json()


# ── Step 4: Extract the confirmation link ────────────────────────────────────

def extract_confirmation_link(body_text: str | None, body_html: str | None) -> str:
    """
    Extract the confirmation URL from the email body.
    Tries body_text first (simpler to parse), falls back to body_html.
    Adjust the pattern to match your application's confirmation URL format.
    """
    pattern = rf"{re.escape(APP_URL)}/confirm\?token=[A-Za-z0-9._-]+"

    for content in [body_text, body_html]:
        if content:
            match = re.search(pattern, content)
            if match:
                return match.group(0)

    raise ValueError("Confirmation link not found in email body")


# ── Main test ─────────────────────────────────────────────────────────────────

def test_signup_and_email_confirmation():
    # 1. Create a fresh, isolated inbox for this test run
    address, session_token = create_inbox(ttl_minutes=10)
    print(f"Test inbox: {address}")

    # 2. Open the browser and fill the signup form
    options = Options()
    options.add_argument("--headless")
    options.add_argument("--no-sandbox")
    options.add_argument("--disable-dev-shm-usage")

    driver = webdriver.Chrome(options=options)
    try:
        driver.get(f"{APP_URL}/signup")

        # Enter the temporary inbox address where the form asks for email
        driver.find_element(By.NAME, "email").send_keys(address)
        driver.find_element(By.NAME, "password").send_keys("TestPassword1!")
        driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()

        # Assert the app shows a "check your email" page
        assert "check your email" in driver.page_source.lower(), \
            "Expected confirmation prompt after signup"

        # 3. Wait for the confirmation email to arrive
        summary = wait_for_message(address, session_token, timeout=30)
        assert "confirm" in summary["subject"].lower(), \
            f"Unexpected subject: {summary['subject']}"

        # 4. Retrieve the full message
        message = get_full_message(address, session_token, summary["id"])

        # 5. Extract the confirmation link from the body
        confirm_url = extract_confirmation_link(
            message["body_text"],
            message["body_html"],
        )

        # 6. Navigate Selenium to the confirmation link
        driver.get(confirm_url)

        # Assert the account is now active
        assert "dashboard" in driver.current_url or \
               "verified" in driver.page_source.lower(), \
               "Email confirmation did not complete successfully"

        print("Test passed.")

    finally:
        driver.quit()


if __name__ == "__main__":
    test_signup_and_email_confirmation()

A few notes on this implementation:

body_text over body_html for extraction. The API returns both fields pre-decoded — multipart MIME, quoted-printable encoding, and base64 are all handled server-side. You work with plain Python strings. For regex matching, body_text is less noisy than HTML, but both are available.

wait_for_message is synchronous. For headless Selenium in a single-threaded test, polling with time.sleep is the straightforward choice. If you want lower-latency delivery notification without polling overhead, the WebSocket endpoint wss://uncorreotemporal.com/ws/inbox/{address}?token={session_token} emits {"event": "new_message", "message_id": "..."} the instant the message is stored. Using asyncio + websockets removes the polling entirely, but adds async complexity that is rarely necessary in a standard Selenium suite.

The address format is human-readable: adjective-noun-number@uncorreotemporal.com. It reads cleanly in test logs and screenshots, which helps when debugging failures.

Advantages for CI/CD Pipelines

Once you replace shared inboxes with per-test temporary inboxes, several problems in CI disappear without any further changes.

No shared state between test workers. Pytest-xdist workers, GitHub Actions matrix jobs, or parallel Jenkins stages each create their own inbox independently. There is no coordination layer, no locking, no message-routing logic. Each job creates one inbox, uses it, and lets it expire.

Reproducible environments. Every test run starts with the same precondition: an empty inbox. There are no leftover messages from previous runs. You do not need a cleanup job or a teardown step to ensure isolation — the TTL handles it.

Zero credentials to manage. For anonymous use, POST /api/v1/mailboxes requires no authentication header. The session token in the response is the only credential, and it is scoped to a single inbox. You do not need to provision IMAP credentials, rotate API keys, or configure OAuth clients in CI secrets.

Easy to add to an existing pytest suite. Wrapping inbox creation in a function-scoped fixture takes five lines:

import pytest
import requests

BASE_URL = "https://uncorreotemporal.com"

@pytest.fixture
def temp_inbox():
    resp = requests.post(f"{BASE_URL}/api/v1/mailboxes", params={"ttl_minutes": 10})
    resp.raise_for_status()
    data = resp.json()
    yield data["address"], data["session_token"]
    # No explicit teardown needed — inbox expires automatically

Any Selenium test that needs email can declare def test_something(driver, temp_inbox) and receive a fresh inbox address and token without any setup boilerplate. The same fixture works in Playwright — just swap out the Selenium driver.

Integration with GitHub Actions is straightforward. No special service containers or SMTP routing are required:

- name: Run E2E tests
  run: pytest tests/e2e/ -v --timeout=60
  env:
    APP_URL: ${{ vars.STAGING_URL }}

For authenticated access (higher TTL limits, multiple active mailboxes), add an API key secret and pass it as a Bearer token:

headers = {"Authorization": "Bearer ${{ secrets.UCT_API_KEY }}"}
resp = requests.post(f"{BASE_URL}/api/v1/mailboxes", headers=headers)

The Infrastructure Behind It

uncorreotemporal.com is a programmable temporary email infrastructure that allows developers to create and manage inboxes via API. The backend is a FastAPI application with an async PostgreSQL database, Redis pub/sub for real-time event propagation, and an aiosmtpd-based SMTP server for local email ingestion. Inboxes expire automatically via a background worker — no external cron jobs needed.

The WebSocket endpoint (/ws/inbox/{address}) subscribes to the Redis channel mailbox:{address} and forwards events to connected clients in real time. The REST API (/api/v1/mailboxes and /api/v1/mailboxes/{address}/messages) provides the polling interface for environments where WebSockets are not practical.

It is designed to be used in code, not in a browser — which is what makes it a practical fit for Selenium test suites.

Conclusion

Email confirmation flows fail in automated testing because the browser can fill a form but cannot read an inbox. The standard workarounds — shared test addresses, Gmail IMAP, mocked senders, staging SMTP sinks — each introduce a different form of fragility: cross-test contamination, credential management, incomplete coverage, or flaky polling.

A temporary inbox API solves the problem at the right level. Create a real inbox via HTTP before each test. Pass the address to Selenium. Wait for the email. Extract the link or OTP. Complete the flow. The entire pattern fits in a pytest fixture and runs in any CI environment without additional infrastructure.

The result is that email confirmation tests become as reliable as your other Selenium tests — isolated, deterministic, and runnable in parallel without coordination.

Building Developer Tools That Scale: Lessons from Email Infrastructure

Francisco Perez — Fri, 13 Mar 2026 20:34:56 +0000

There is a certain category of developer tool that looks, from the outside, like it should take a weekend to build. A temporary email service is one of them. Create an address, receive mail, expire it after N minutes. The interface is trivial. The implementation is not.

Building uncorreotemporal.com — a programmable temporary email infrastructure for automation, CI pipelines, and AI agents — required solving a set of engineering problems that most "simple" developer tools eventually surface: concurrent ingestion, time-bounded data, real-time event propagation, and API contracts that hold up under machine load. This article is about those problems and how we solved them.

The Illusion of Simple Developer Tools

The most dangerous developer tools to build are the ones where the interface is clean. A queue with three methods. An email API with two endpoints. An auth service with one decision. The surface area looks small, which makes it easy to underestimate the engine behind it.

Consider what email APIs, queues, authentication services, and testing tools have in common: they are all infrastructure primitives. They do not do business logic — they are the layer below business logic. And infrastructure primitives fail in ways that are invisible until they do not: message loss, split-brain states, TTL races, partial writes, replay ambiguity.

When you build a temporary email service for developers, you are not building a toy. You are building infrastructure that test suites depend on to create isolated inboxes, receive confirmation emails, and clean up reliably after every run. If it drops a message or delivers to an expired inbox, a CI job fails with a confusing, non-deterministic error. If it has an unpredictable API, automation scripts break when the contract shifts. The simplicity of the interface is a contract with your users — maintaining it requires solving complexity underneath.

Real Email Infrastructure Is Hard

SMTP is a 40-year-old protocol that was designed for delivery, not for programmable use. Building on top of it means accepting its characteristics: messages are raw bytes, headers are often inconsistent, multipart structure is implicit, and the protocol itself has no concept of a "user" or an "inbox" — only a recipient address and a message.

The ingestion layer has two distinct modes. In development and local testing, aiosmtpd runs as a proper async SMTP server:

class MailHandler:
    async def handle_DATA(self, server, session, envelope) -> str:
        raw: bytes = envelope.content

        for rcpt in envelope.rcpt_tos:
            address = rcpt.lower().strip()
            if not address.endswith(f"@{settings.domain}"):
                logger.debug("Dominio no reconocido: %s — ignorado", address)
                continue
            await deliver_raw_email(raw, address)

        return "250 Message accepted"

The handler always returns 250 Message accepted. Rejection is silent and internal — a deliberate design decision. SMTP's response codes communicate to the sending server, not to the end user. Returning a 5xx for an unknown address would tell spammers which addresses exist.

In production, the ingestion path is entirely different: AWS SES receives the mail, publishes an SNS notification, and calls a webhook at POST /api/v1/ses/inbound. Both paths converge at core/delivery.py.

Parsing raw RFC 2822 bytes into structured data is where the real complexity lives. The implementation uses Python's stdlib email module with policy.default — the modern, standards-compliant policy introduced in Python 3.6:

def parse_email(raw: bytes) -> ParsedEmail:
    msg = email.message_from_bytes(raw, policy=policy.default)
    if msg.is_multipart():
        for part in msg.walk():
            content_type = part.get_content_type()
            disposition = str(part.get("content-disposition", ""))
            if "attachment" in disposition:
                attachments.append(ParsedAttachment(
                    filename=part.get_filename() or "attachment",
                    content_type=content_type,
                    size=len(part.get_payload(decode=True) or b""),
                ))
            elif content_type == "text/plain" and body_text is None:
                body_text = part.get_content()
            elif content_type == "text/html" and body_html is None:
                body_html = part.get_content()

A critical architectural choice: attachment binary payloads are not stored separately. Only metadata (filename, content type, size, content-id) is persisted as JSONB. The complete raw_email bytes are stored alongside. This means the original message is always re-parseable without duplication — and it simplifies the schema at the cost of making attachment retrieval slightly more expensive.

Designing for Concurrency from Day One

The entire stack is async: FastAPI with SQLAlchemy async sessions, aiosmtpd for SMTP, aioredis for pub/sub, and asyncio tasks for background work. This was not an afterthought — it was a prerequisite for any form of meaningful throughput.

The delivery pipeline (core/delivery.py) illustrates the async design. It is a six-step sequence where each step is an awaited call: look up the mailbox, load the plan, check quota, parse the email, insert the message, publish to Redis. No blocking I/O, no thread-per-connection overhead:

async def deliver_raw_email(raw: bytes, address: str) -> bool:
    async with AsyncSessionLocal() as db:
        return await _save_to_mailbox(raw, address, db)

The function returns bool — True if the message was saved, False if it was silently rejected. This return value is consumed by the SMTP handler and the SES webhook alike, keeping both callers simple.

The fan-out from message delivery to real-time notification follows a pub/sub pattern over Redis:

# After DB commit:
payload = json.dumps({"event": "new_message", "message_id": str(message.id)})
await redis.publish(f"mailbox:{address}", payload)

The publish happens after the database commit. If Redis is unreachable, the exception is caught and logged — the message is already durable in PostgreSQL. Redis is the notification layer, not the source of truth. This ordering matters: message loss in Redis is recoverable (the client can poll); message loss in the DB is not.

Time-Based Data Is a Hidden Challenge

Expiring data is deceptively hard. The naive implementation — DELETE FROM mailboxes WHERE expires_at < now() — conflates two separate concerns: enforcement and storage reclamation. The production implementation separates them.

Expiration is a state transition, not a deletion. The background worker runs as an asyncio task co-hosted with the API process and fires every 60 seconds:

async def _expire_mailboxes() -> int:
    now = datetime.now(timezone.utc)
    async with AsyncSessionLocal() as db:
        result = await db.execute(
            update(Mailbox)
            .where(
                Mailbox.expires_at <= now,
                Mailbox.is_active == True,
            )
            .values(is_active=False)
        )
        await db.commit()
        return result.rowcount

One SQL statement. No Python-level iteration. No loading rows into memory. The entire sweep is a single round-trip to PostgreSQL, bounded in cost by the number of newly expired inboxes since the last cycle.

The sleep mechanism is worth examining. Rather than await asyncio.sleep(60), the worker uses:

await asyncio.wait_for(stop_event.wait(), timeout=interval_seconds)

A sleeping coroutine cannot be interrupted without task cancellation. An event wait can be signaled immediately. When the FastAPI lifespan shuts down, stop_expiry_task() sets the event and the worker exits within milliseconds rather than waiting up to 60 seconds for a sleep to expire.

The subtler challenge is the race between the 60-second polling window and message delivery. Between an inbox's expires_at timestamp and when the background worker next runs, is_active may still be True even though the inbox has expired. The delivery layer closes this window with a double-check:

result = await db.execute(
    select(Mailbox).where(
        Mailbox.address == address,
        Mailbox.is_active == True,
        Mailbox.expires_at > now,   # critical: not just is_active
    )
)

The background worker's state transition is an optimization that keeps list queries and quota checks efficient. But it is not the primary enforcement mechanism. Delivery correctness does not depend on the worker having run.

API Design for Automation

Developer tools built for automation require a different API design philosophy than consumer products. The contract must be machine-readable, predictable, and composable with CI pipelines and agent loops.

The mailbox creation endpoint is designed for automation-first use:

POST /api/v1/mailboxes?ttl_minutes=30

No request body — TTL is a query parameter because it is always optional and has a plan-dependent default. The response is flat and explicit:

{
  "address": "swift-river-42@uncorreotemporal.com",
  "expires_at": "2026-03-13T15:30:00+00:00",
  "session_token": "uct_xxxxx"
}

expires_at is always an ISO 8601 timestamp with UTC offset — not a duration, not a relative time. A test harness receives this and can schedule its own assertion: if datetime.now(tz=utc) > expires_at, the inbox is gone. No ambiguity about timezone, no state to track.

Plan limits are enforced at the API layer before any resource is created:

effective_ttl = min(ttl_minutes or plan.default_ttl_minutes, plan.max_ttl_minutes)

A free plan user requesting a 48-hour inbox gets a 60-minute inbox, silently capped. The expires_at in the response reflects the actual cap. Free caps: 60 minutes, 1 mailbox, 10 messages. Pro: 24 hours, 20 mailboxes, 500 messages.

The messages API separates listing from retrieval:

GET    /api/v1/mailboxes/{address}/messages        → list (metadata only, no body)
GET    /api/v1/mailboxes/{address}/messages/{id}   → full message (marks read)
DELETE /api/v1/mailboxes/{address}/messages/{id}   → delete

The list endpoint never returns body content — only id, from_address, subject, received_at, is_read, has_attachments. A CI test polling for an OTP email reads the list, finds the matching subject, fetches only that message. The list stays lightweight regardless of how many large emails have been received.

Real-Time Systems Change the Game

Polling is the wrong abstraction for event-driven workflows. A test suite that polls GET /messages every second is burning API quota and introducing latency that scales with polling interval. WebSockets change the model entirely.

The WebSocket endpoint at WS /ws/inbox/{address} accepts connections authenticated via either a session token or an API key. Authentication happens before the connection is accepted — an unauthorized client receives a 1008 Policy Violation close frame, not a connected stream followed by a permission error:

authorized = await _authenticate(address, token, api_key)
if not authorized:
    await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
    return
await websocket.accept()

Once connected, the endpoint subscribes to the Redis pub/sub channel for that inbox and runs two concurrent asyncio tasks:

send_task = asyncio.create_task(_send_loop())   # forwards Redis messages to WS
ping_task = asyncio.create_task(_ping_loop())   # 30s keepalive ping

done, pending = await asyncio.wait(
    [send_task, ping_task],
    return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
    task.cancel()

The client receives two event types:

{"event": "new_message", "message_id": "550e8400-e29b-41d4-a716-446655440000"}
{"event": "ping"}

On new_message, the client calls GET /api/v1/mailboxes/{address}/messages/{id} to retrieve the full body. The WebSocket is a notification channel only — it carries the signal, not the payload. For AI agents and CI pipelines: open a WebSocket before triggering the flow that sends the email, wait for the new_message event, fetch the body, extract the OTP. Zero polling. Sub-second latency from email arrival to agent response.

Infrastructure Lessons Learned

Async is the default, not an optimization. Every I/O operation — database queries, Redis operations, SMTP handling, WebSocket messaging — is non-blocking. This is the design that makes a single-process deployment serve concurrent WebSocket connections, handle incoming SMTP sessions, and run background expiration sweeps without threading complexity.

Design for failure at every layer. The Redis publish in the delivery pipeline is wrapped in a try/except that logs the error and returns True. The message is already in PostgreSQL. Redis failure does not become delivery failure. The expiry loop wraps its core operation in a try/except so that a transient database error does not kill the loop. Infrastructure components fail. The question is whether the failure is isolated or cascading.

Separate enforcement from cleanup. The expiry model distinguishes between is_active = False (enforcement, near-real-time) and physical deletion (cleanup, deferred). This separation means the delivery layer can enforce TTL without depending on a garbage collector having run.

Single source of truth for state. Plan limits and TTL caps live in the database, not in application configuration. A free plan with max_ttl_minutes = 60 is a row in the plans table. Changing it requires no deployment.

Dual ingestion path, shared logic. The aiosmtpd handler and the SES webhook share core/delivery.py entirely. The SMTP handler's handle_DATA method is three lines of routing logic; everything else is shared. Local development tests exactly the same delivery pipeline as production SES ingestion.

The Hidden Complexity Behind "Simple" Products

The products that developers call "simple" are the ones where the interface succeeded. Stripe's charge API is simple. S3's put/get is simple. The simplicity is the product of enormous engineering effort spent hiding complexity from the caller.

A temporary email service that works looks like this from the outside: create inbox, receive mail, read message, inbox expires. Four operations. What it requires underneath: a domain registered with MX records pointing at your infrastructure, an async SMTP ingestion layer with domain validation and silent rejection, an RFC 2822 parser that handles malformed multipart messages without crashing, a quota enforcement system tied to a plan hierarchy, a background expiration worker with correct race condition handling, a real-time notification system over Redis pub/sub with WebSocket fan-out, and an API designed for machine consumption at every endpoint.

None of these are visible in the API contract. That invisibility is the job.

The engineering trap is believing that because the interface is simple, the implementation can be simple too. It cannot. The interface's simplicity is the result of pushing complexity inward — into the infrastructure, into the error handling, into the data model. When you skip that work, the complexity leaks out into the caller's code.

Conclusion

Building developer tools that scale comes down to a few durable engineering principles.

Make async the baseline, not the upgrade path. If your infrastructure needs to handle concurrent connections, background workers, and real-time events — and any non-trivial tool will — an async framework is the correct starting point, not a future refactor.

Design data lifecycle explicitly. Time-bounded data requires a model that separates enforcement from cleanup, and an enforcement mechanism that does not depend on cleanup having run. Race conditions in time-based systems are not edge cases — they are the default condition.

Build APIs for machines first. Predictable response shapes, explicit timestamps, flat structures, and quiet enforcement of limits are properties that make automation reliable. An API designed for a human UI can often be driven by a machine; an API designed for a machine can always be used by a human.

Isolate failure. Every I/O operation has failure modes. The question is whether those failures propagate or are contained. Redis unavailability should not cause message loss. A transient database error should not kill a background worker. Design the blast radius before the failure happens.

The measure of infrastructure engineering is not what happens under ideal conditions — it is what happens when things go wrong, and whether those failures are invisible to the people building on top of you. That is what "simple" actually means.

The full system is running at https://uncorreotemporal.com. Anonymous inboxes need no signup: POST /api/v1/mailboxes?ttl_minutes=5 creates a live inbox with a five-minute TTL and returns a WebSocket-ready address. It is as simple as it looks.