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.
Top comments (0)