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:

• 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.

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. [e-mail]
    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), api (authenticated user with API key), and mcp (user accessing via the MCP server).

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]

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

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

        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. When either task completes the other is cancelled.

---

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.

API key format: uct_<32-char-url-safe-base64>. Only the SHA-256 hash is stored — 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 [e-mail]
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}")
# → [e-mail]

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

---

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

{
  "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. No HTTP client code, no auth logic, no parsing boilerplate.

Integration patterns

Autonomous QA agents. Each test run gets a fresh inbox. Parallel test suites run with isolated inboxes — no coordination needed.

Research agents. Register for multiple services, benchmark onboarding flows, test email deliverability — all without a real email account.

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

Multi-agent systems. An orchestrator provisions a dedicated inbox per sub-agent, routing external signals without shared state.

---

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

A few natural extensions:

• Webhook forwarding: Push new_message events to an HTTP callback without a persistent WebSocket connection.
• Content filtering: Search messages by sender or subject at the API level.
• Attachment extraction: Binary content retrieval to unlock document-processing workflows.
• Domain customization: Custom domains for enterprise and integration testing scenarios.

---

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 (aiosmtpd in development, AWS SES in production), WebSocket-based real-time delivery via Redis pub/sub, and an MCP server that exposes email as a native tool for LLM agents. Inboxes are isolated, plan-gated, and automatically expired.

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

→ uncorreotemporal.com