Giving Autonomous Agents Access to Email

1. The Problem: Why AI Agents Cannot Use Email

Most autonomous agents are built around HTTP. They call APIs, parse JSON, render HTML. They can browse the web, fill out forms, even write and run code. But when a workflow requires receiving an email — to complete a signup, retrieve an OTP, or interact with an email-based workflow — most agent architectures hit a wall.

The reasons are structural:

• Email requires SMTP infrastructure. You cannot receive email without a domain, an MX record, and a server listening on port 25. Standing that up for a test or a research experiment is expensive and slow.
• Inbox polling is not trivial. Reading from Gmail or Outlook programmatically requires OAuth flows, scope approvals, and token management — none of which are agent-friendly.
• Test emails pollute real inboxes. If you are testing a registration flow or an OTP retrieval workflow, you do not want noise in your production mailbox.
• Email confirmation flows break automation. Many web services require clicking a link in a verification email before an account becomes active. Agents have no good way to intercept that email on demand.
• Receiving email programmatically is non-trivial. There is no "email equivalent of fetch()." Email delivery is push-based, asynchronous, and requires infrastructure to receive.

For human users, temporary email services solve the throwaway inbox problem. But they are built for browsers, not APIs. They offer no programmatic access, no webhooks, no structured responses — nothing an agent can work with.

---

2. Why Email Is Still a Critical Interface

Despite the dominance of REST APIs and webhooks, email remains a load-bearing interface for identity and verification across the modern web. A surprisingly large fraction of real-world automation pipelines runs through it:

• Account registration with email confirmation
• One-time password (OTP) delivery
• Magic link authentication
• Password reset workflows
• Transactional notifications with actionable content
• B2B trial activation and onboarding sequences

For an autonomous agent to operate in the real world — registering for services, verifying accounts, or testing signup flows end-to-end — it needs email access. Not a mocked version. A real inbox that receives real SMTP traffic.

---

3. Architecture of a Programmable Email Infrastructure

uncorreotemporal.com is a programmable temporary email service built with agents and developers in mind. The stack is designed for programmatic access from the ground up.

Core components:

┌─────────────────────────────────────────────────────────┐
│                  SMTP Server (aiosmtpd)                  │
│              Listens on port 25 for inbound mail         │
└────────────────────────┬────────────────────────────────┘
                         │ deliver_raw_email()
                         ▼
┌─────────────────────────────────────────────────────────┐
│              Delivery Pipeline (core/delivery.py)        │
│   1. Lookup active mailbox                               │
│   2. Check plan quota                                    │
│   3. Parse RFC 2822 email                                │
│   4. Insert Message into PostgreSQL                      │
│   5. Publish event to Redis pub/sub                      │
└────────────┬────────────────────────┬───────────────────┘
             │                        │
             ▼                        ▼
┌────────────────────┐   ┌────────────────────────────────┐
│   PostgreSQL       │   │   Redis (mailbox:{address})    │
│   - mailboxes      │   │   {"event":"new_message",      │
│   - messages       │   │    "message_id": "uuid"}       │
│   - api_keys       │   └────────────┬───────────────────┘
└────────────────────┘                │
                                      ▼
                         ┌────────────────────────────────┐
                         │   WebSocket /ws/inbox/{address}│
                         │   Forwards events to client    │
                         └────────────────────────────────┘

The system runs on Python 3.12 + FastAPI, with aiosmtpd handling SMTP in development and AWS SES (via SNS webhooks) in production. PostgreSQL is the primary store; Redis is the message bus between ingestion and real-time consumers.

Application lifecycle

The FastAPI app uses a lifespan context manager to start background tasks at startup:

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    start_expiry_task(interval_seconds=60)
    yield
    await stop_expiry_task()
    await close_redis()

The expiry worker runs every 60 seconds and soft-deletes mailboxes where expires_at <= now(). Because this is a single asyncio task, the API must run with --workers 1.

Data models

A Mailbox tracks ownership, expiry, and activation state:

class Mailbox(Base):
    __tablename__ = "mailboxes"

    id: Mapped[uuid.UUID]
    address: Mapped[str]              # e.g. mango-panda-42@uncorreotemporal.com
    expires_at: Mapped[datetime]
    owner_type: Mapped[OwnerType]     # "anonymous" | "api" | "mcp"
    owner_id: Mapped[uuid.UUID | None]
    session_token: Mapped[str | None] # for anonymous access
    is_active: Mapped[bool]

Three ownership modes exist: anonymous (session token only, no account needed), api (authenticated user with API key), and mcp (user accessing via the MCP server). The owner_type field is tagged for audit purposes — all three ultimately enforce the same quota rules.

A Message stores the complete raw email alongside parsed fields:

class Message(Base):
    __tablename__ = "messages"

    id: Mapped[uuid.UUID]
    mailbox_id: Mapped[uuid.UUID]
    from_address: Mapped[str]
    subject: Mapped[str | None]
    body_text: Mapped[str | None]
    body_html: Mapped[str | None]
    raw_email: Mapped[bytes]               # complete RFC 2822, always stored
    attachments: Mapped[list[dict] | None] # JSONB metadata only
    received_at: Mapped[datetime]
    is_read: Mapped[bool]

Attachment binary content stays in raw_email. JSONB stores only metadata — filename, content type, size, content-id. This means you can re-parse the original bytes with a different parser later without any data loss.

Shared delivery pipeline

Both SMTP and SES paths call the same function:

async def deliver_raw_email(raw: bytes, address: str) -> bool:
    async with AsyncSessionLocal() as db:
        mailbox = await find_active_mailbox(address, db)
        if not mailbox:
            return False                          # silent reject

        plan = await get_plan(mailbox, db)
        if not await check_message_quota(mailbox, plan, db):
            return False                          # quota exceeded

        parsed = parse_email(raw)
        message = Message(
            mailbox_id=mailbox.id,
            from_address=parsed.from_address,
            subject=parsed.subject,
            body_text=parsed.body_text,
            body_html=parsed.body_html,
            raw_email=raw,
            attachments=[a.to_dict() for a in parsed.attachments],
            is_read=False,
        )
        db.add(message)
        await db.commit()

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

After persisting the message, it publishes to mailbox:{address} on Redis. Any WebSocket client subscribed to that channel receives the event within milliseconds.

Real-time delivery via WebSocket

@router.websocket("/ws/inbox/{address}")
async def websocket_inbox(websocket: WebSocket, address: str, api_key: str | None = None):
    if not await _authenticate(address, api_key):
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    await websocket.accept()
    pubsub = redis.pubsub()
    await pubsub.subscribe(f"mailbox:{address}")

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

    async def _ping_loop():
        while True:
            await asyncio.sleep(30)
            await websocket.send_text(json.dumps({"event": "ping"}))

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

Two concurrent tasks run per connection: _send_loop forwards Redis events, _ping_loop sends keepalives every 30 seconds to prevent proxy timeouts. When either task completes the other is cancelled and the pubsub connection is cleaned up.

---

4. How Autonomous Agents Use Temporary Inboxes

With this infrastructure in place, an agent gains the following primitives:

Action	Endpoint
Create a disposable inbox	POST /api/v1/mailboxes
List messages (metadata)	GET /api/v1/mailboxes/{address}/messages
Read full message + mark read	GET /api/v1/mailboxes/{address}/messages/{id}
Watch inbox in real time	WS /ws/inbox/{address}
Delete inbox	DELETE /api/v1/mailboxes/{address}

Each inbox is ephemeral, isolated, and scoped to the agent's API key. No shared state, no cross-contamination between runs.

Plan-based quotas govern every mailbox:

• max_mailboxes — how many active inboxes a user can hold
• max_messages_per_mailbox — -1 means unlimited
• max_ttl_minutes — upper bound on inbox lifetime
• api_access / mcp_access — feature flags per plan tier

API key format: uct_<32-char-url-safe-base64>. Only the SHA-256 hash is stored in the database — the raw key is shown exactly once at creation time.

---

5. Example Workflow: Agent Completes Email Verification

Consider an autonomous QA agent testing a new user registration flow:

1. Create a temporary inbox → receives mango-panda-42@uncorreotemporal.com
2. Open a WebSocket connection to watch for incoming mail
3. Submit the registration form with the temporary address
4. Receive the confirmation email via WebSocket event
5. Read the email body to extract the verification link or OTP
6. Follow the link to complete registration
7. Delete the inbox to clean up

This loop closes in seconds. The agent never touches a real email account, never requires OAuth, and leaves no trace.

---

6. Code Examples Using the API

Create an inbox

import httpx

API_BASE = "https://uncorreotemporal.com"
HEADERS = {"Authorization": "Bearer uct_your_api_key_here"}

resp = httpx.post(
    f"{API_BASE}/api/v1/mailboxes",
    headers=HEADERS,
    params={"ttl_minutes": 30},
)
resp.raise_for_status()
inbox = resp.json()
address = inbox["address"]
print(f"Inbox: {address}")
# → mango-panda-42@uncorreotemporal.com

Poll for an email

import time

def wait_for_email(address: str, timeout: int = 60) -> dict | None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        resp = httpx.get(
            f"{API_BASE}/api/v1/mailboxes/{address}/messages",
            headers=HEADERS,
        )
        messages = resp.json()
        if messages:
            return messages[0]
        time.sleep(3)
    return None

message_summary = wait_for_email(address)

Read the full message and extract an OTP

import re

msg = httpx.get(
    f"{API_BASE}/api/v1/mailboxes/{address}/messages/{message_summary['id']}",
    headers=HEADERS,
).json()

body = msg.get("body_text", "") or ""
match = re.search(r"\b(\d{6})\b", body)
if match:
    otp = match.group(1)
    print(f"OTP: {otp}")

Watch with WebSocket for real-time delivery

import asyncio
import websockets
import json

async def watch_inbox(address: str, api_key: str):
    uri = f"wss://uncorreotemporal.com/ws/inbox/{address}?api_key={api_key}"
    async with websockets.connect(uri) as ws:
        async for raw in ws:
            event = json.loads(raw)
            if event.get("event") == "new_message":
                print(f"New message: {event['message_id']}")
                return event["message_id"]

asyncio.run(watch_inbox(address, "uct_your_api_key_here"))

The WebSocket approach is preferable for latency-sensitive workflows. The server publishes {"event": "new_message", "message_id": "uuid"} to the Redis channel mailbox:{address} immediately after persisting each message — the WebSocket handler forwards it to all connected subscribers without polling.

---

7. Using Email as a Tool in Agent Architectures

MCP Server (Model Context Protocol)

The system ships a native MCP server at mcp/server.py that exposes five tools directly to LLM agents via stdio:

Tool	Description
create_mailbox	Create a new temporary inbox with optional TTL
list_mailboxes	List all active inboxes for the authenticated user
get_messages	List messages in an inbox (metadata only, up to 100)
read_message	Fetch full message body and mark as read
delete_mailbox	Soft-delete an inbox

For agents running in Claude Desktop or any MCP-compatible runtime, this is a zero-boilerplate integration:

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

The MCP server reads UCT_API_KEY from the environment at startup, validates it once, and exposes all five tools. The agent calls create_mailbox(), get_messages(), and read_message() as native tool calls — no HTTP client code, no auth logic, no parsing boilerplate.

Integration patterns

Autonomous QA agents. Each test run gets a fresh inbox. No manual cleanup. Parallel test suites run with isolated inboxes — no coordination mechanism required, no collision.

Research agents. Agents that register for multiple services to gather data, benchmark onboarding flows, or test email deliverability — all without a real email account exposed.

Agentic RPA. Automation agents that interact with legacy systems built around email-based workflows: invoice delivery, approval chains, alert digests.

Multi-agent systems. An orchestrator can provision a dedicated inbox per sub-agent, routing external signals to specific workers without any shared state.

---

8. Future: Email as a First-Class Agent Tool

The pattern established here — ephemeral, API-first, real-infrastructure — suggests a direction for how agent tooling around communication channels should evolve.

A few natural extensions:

• Webhook forwarding: Push new_message events to an agent's HTTP callback without requiring a persistent WebSocket connection.
• Content filtering: Search messages by sender or subject at the API level, rather than fetching all and filtering client-side.
• Attachment extraction: Attachment metadata is already stored (filename, content-type, size); direct content retrieval would unlock document-processing workflows.
• Domain customization: Allowing agents to receive mail on custom domains opens enterprise and integration testing scenarios.

The infrastructure is already capable. The surface area for agent-specific ergonomics is where the interesting work is.

---

Summary

uncorreotemporal.com is programmable temporary email infrastructure for developers and autonomous agents. It provides a REST API for inbox creation and message retrieval, a real SMTP layer that receives actual email (aiosmtpd in development, AWS SES in production), WebSocket-based real-time delivery notifications backed by Redis pub/sub, and a Model Context Protocol server that exposes email as a native tool for LLM agents. Inboxes are isolated, plan-gated, and automatically expired — everything an agent needs to participate in email-based workflows without managing email infrastructure.

If you are building autonomous agents that need to interact with the real web, email is no longer the missing piece.

→ uncorreotemporal.com