#!/usr/bin/env python3
"""
Email Agent — Programmable Email Inboxes for E2E Testing
==========================================================
Provides AI agents with temporary email inboxes for end-to-end testing.
Supports MailSlurp, AgentMail, and Inbucket (self-hosted) backends
via a unified async interface.
Usage:
provider = get_email_provider()
inbox = await provider.create_inbox()
email = await wait_for_email(provider, inbox.id, subject_contains="Verify")
link = extract_verification_link(email.html)
Run self-test:
python testing/email_agent.py
"""
from __future__ import annotations
import asyncio
import logging
import os
import random
import re
import time
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from html.parser import HTMLParser
from typing import List, Optional
from urllib.parse import urlparse
import httpx
logger = logging.getLogger("email_agent")
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s — %(message)s")
# ---------------------------------------------------------------------------
# Data models
# ---------------------------------------------------------------------------
@dataclass
class Inbox:
"""A temporary email inbox."""
id: str
email_address: str
provider: str
created_at: Optional[str] = None
extra: dict = field(default_factory=dict)
@dataclass
class Email:
"""An email message received in an inbox."""
id: str
inbox_id: str
subject: str
from_address: str
to: List[str]
received_at: Optional[str]
html: Optional[str] = None
text: Optional[str] = None
extra: dict = field(default_factory=dict)
# ---------------------------------------------------------------------------
# Abstract interface
# ---------------------------------------------------------------------------
class EmailProvider(ABC):
"""Abstract base class for email inbox providers."""
@abstractmethod
async def create_inbox(self, name: Optional[str] = None) -> Inbox:
"""Create a new temporary inbox and return it."""
...
@abstractmethod
async def get_emails(self, inbox_id: str) -> List[Email]:
"""Return all emails currently in the inbox (summary, no full body)."""
...
@abstractmethod
async def get_email_content(self, inbox_id: str, email_id: str) -> Email:
"""Fetch full content (HTML + text body) for a single email."""
...
@abstractmethod
async def extract_links(self, inbox_id: str, email_id: str) -> List[str]:
"""Return all hyperlinks found in the email HTML body."""
...
@abstractmethod
async def delete_inbox(self, inbox_id: str) -> bool:
"""Delete the inbox and all its messages. Returns True on success."""
...
def create_human_email(self, domain: Optional[str] = None) -> str:
"""Generate a realistic-looking human email address for the provider domain."""
first_names = [
"james", "emma", "oliver", "ava", "william", "sophia", "noah",
"isabella", "liam", "mia", "ethan", "charlotte", "mason",
"amelia", "jacob", "harper", "michael", "evelyn", "jack", "abigail",
]
last_names = [
"smith", "johnson", "williams", "jones", "brown", "davis", "miller",
"wilson", "moore", "taylor", "anderson", "thomas", "jackson",
"white", "harris", "martin", "thompson", "garcia", "martinez", "robinson",
]
first = random.choice(first_names)
last = random.choice(last_names)
suffix = random.randint(10, 99)
separators = [f"{first}.{last}{suffix}", f"{first}{last}{suffix}",
f"{first}_{last}{suffix}", f"{first[0]}{last}{suffix}"]
local = random.choice(separators)
target_domain = domain or "gmail.com"
return f"{local}@{target_domain}"
# ---------------------------------------------------------------------------
# Link extraction helper (no external HTML parser needed)
# ---------------------------------------------------------------------------
class _LinkExtractor(HTMLParser):
"""Minimal HTML parser that collects all href values."""
def __init__(self) -> None:
super().__init__()
self.links: List[str] = []
def handle_starttag(self, tag: str, attrs: list) -> None:
if tag == "a":
for attr, value in attrs:
if attr == "href" and value:
self.links.append(value)
def _extract_links_from_html(html: str) -> List[str]:
parser = _LinkExtractor()
parser.feed(html)
return parser.links
# ---------------------------------------------------------------------------
# MailSlurp provider
# ---------------------------------------------------------------------------
class MailSlurpProvider(EmailProvider):
"""
MailSlurp (https://mailslurp.com) — cloud temporary email service.
Requires MAILSLURP_API_KEY env var.
"""
BASE_URL = "https://api.mailslurp.com"
def __init__(self, api_key: Optional[str] = None) -> None:
self.api_key = api_key or os.environ["MAILSLURP_API_KEY"]
self._headers = {
"x-api-key": self.api_key,
"Content-Type": "application/json",
}
def _client(self) -> httpx.AsyncClient:
return httpx.AsyncClient(headers=self._headers, timeout=30)
async def create_inbox(self, name: Optional[str] = None) -> Inbox:
payload: dict = {}
if name:
payload["name"] = name
async with self._client() as client:
resp = await client.post(f"{self.BASE_URL}/createInbox", json=payload)
resp.raise_for_status()
data = resp.json()
logger.info(f"[MailSlurp] Created inbox: {data['emailAddress']}")
return Inbox(
id=data["id"],
email_address=data["emailAddress"],
provider="mailslurp",
created_at=data.get("createdAt"),
)
async def get_emails(self, inbox_id: str) -> List[Email]:
async with self._client() as client:
resp = await client.get(
f"{self.BASE_URL}/getEmailsForInbox",
params={"inboxId": inbox_id, "sort": "ASC"},
)
resp.raise_for_status()
items = resp.json()
return [
Email(
id=item["id"],
inbox_id=inbox_id,
subject=item.get("subject", ""),
from_address=item.get("from", ""),
to=item.get("to", []),
received_at=item.get("createdAt"),
)
for item in items
]
async def get_email_content(self, inbox_id: str, email_id: str) -> Email:
async with self._client() as client:
resp = await client.get(f"{self.BASE_URL}/emails/{email_id}")
resp.raise_for_status()
data = resp.json()
return Email(
id=data["id"],
inbox_id=inbox_id,
subject=data.get("subject", ""),
from_address=data.get("from", ""),
to=data.get("to", []),
received_at=data.get("createdAt"),
html=data.get("body"),
text=data.get("bodyMD5") and data.get("body"), # MailSlurp body is HTML
)
async def extract_links(self, inbox_id: str, email_id: str) -> List[str]:
email = await self.get_email_content(inbox_id, email_id)
if not email.html:
return []
return _extract_links_from_html(email.html)
async def delete_inbox(self, inbox_id: str) -> bool:
async with self._client() as client:
resp = await client.delete(f"{self.BASE_URL}/inboxes/{inbox_id}")
logger.info(f"[MailSlurp] Deleted inbox {inbox_id}: {resp.status_code}")
return resp.status_code in (200, 204)
def create_human_email(self, domain: Optional[str] = None) -> str:
return super().create_human_email(domain or "mailslurp.com")
# ---------------------------------------------------------------------------
# AgentMail provider
# ---------------------------------------------------------------------------
class AgentMailProvider(EmailProvider):
"""
AgentMail (https://agentmail.to) — programmable inboxes for AI agents.
Requires AGENTMAIL_API_KEY env var.
"""
BASE_URL = "https://api.agentmail.to/v0"
def __init__(self, api_key: Optional[str] = None) -> None:
self.api_key = api_key or os.environ["AGENTMAIL_API_KEY"]
self._headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
def _client(self) -> httpx.AsyncClient:
return httpx.AsyncClient(headers=self._headers, timeout=30)
async def create_inbox(self, name: Optional[str] = None) -> Inbox:
payload: dict = {}
if name:
payload["username"] = name.lower().replace(" ", "_")
async with self._client() as client:
resp = await client.post(f"{self.BASE_URL}/inboxes", json=payload)
resp.raise_for_status()
data = resp.json()
logger.info(f"[AgentMail] Created inbox: {data['address']}")
return Inbox(
id=data["inbox_id"],
email_address=data["address"],
provider="agentmail",
created_at=data.get("created_at"),
)
async def get_emails(self, inbox_id: str) -> List[Email]:
async with self._client() as client:
resp = await client.get(f"{self.BASE_URL}/inboxes/{inbox_id}/messages")
resp.raise_for_status()
data = resp.json()
messages = data.get("messages", data) if isinstance(data, dict) else data
return [
Email(
id=msg["message_id"],
inbox_id=inbox_id,
subject=msg.get("subject", ""),
from_address=msg.get("from", ""),
to=msg.get("to", []) if isinstance(msg.get("to"), list) else [msg.get("to", "")],
received_at=msg.get("received_at"),
)
for msg in messages
]
async def get_email_content(self, inbox_id: str, email_id: str) -> Email:
async with self._client() as client:
resp = await client.get(
f"{self.BASE_URL}/inboxes/{inbox_id}/messages/{email_id}"
)
resp.raise_for_status()
data = resp.json()
to_field = data.get("to", [])
if isinstance(to_field, str):
to_field = [to_field]
return Email(
id=data["message_id"],
inbox_id=inbox_id,
subject=data.get("subject", ""),
from_address=data.get("from", ""),
to=to_field,
received_at=data.get("received_at"),
html=data.get("html"),
text=data.get("text"),
)
async def extract_links(self, inbox_id: str, email_id: str) -> List[str]:
email = await self.get_email_content(inbox_id, email_id)
if not email.html:
return []
return _extract_links_from_html(email.html)
async def delete_inbox(self, inbox_id: str) -> bool:
async with self._client() as client:
resp = await client.delete(f"{self.BASE_URL}/inboxes/{inbox_id}")
logger.info(f"[AgentMail] Deleted inbox {inbox_id}: {resp.status_code}")
return resp.status_code in (200, 204)
def create_human_email(self, domain: Optional[str] = None) -> str:
return super().create_human_email(domain or "agentmail.to")
# ---------------------------------------------------------------------------
# Inbucket provider (self-hosted)
# ---------------------------------------------------------------------------
class InbucketProvider(EmailProvider):
"""
Inbucket (https://inbucket.org) — self-hosted disposable email server.
Connects to Inbucket REST API.
Env vars: INBUCKET_URL (default http://localhost:9000)
"""
def __init__(self, base_url: Optional[str] = None) -> None:
raw = base_url or os.environ.get("INBUCKET_URL", "http://localhost:9000")
self.base_url = raw.rstrip("/")
self._domain = urlparse(self.base_url).hostname or "localhost"
def _client(self) -> httpx.AsyncClient:
return httpx.AsyncClient(timeout=15)
def _make_inbox_address(self, name: str) -> str:
return f"{name}@{self._domain}"
async def create_inbox(self, name: Optional[str] = None) -> Inbox:
# Inbucket is implicit — any mailbox name is valid without pre-creation.
# We generate a unique local part and the inbox is ready immediately.
local = name or f"agent_{uuid.uuid4().hex[:12]}"
email_address = self._make_inbox_address(local)
logger.info(f"[Inbucket] Using inbox: {email_address}")
return Inbox(
id=local, # Inbucket uses mailbox name as ID
email_address=email_address,
provider="inbucket",
)
async def get_emails(self, inbox_id: str) -> List[Email]:
async with self._client() as client:
resp = await client.get(f"{self.base_url}/api/v1/mailbox/{inbox_id}")
if resp.status_code == 404:
return []
resp.raise_for_status()
items = resp.json() or []
return [
Email(
id=item["id"],
inbox_id=inbox_id,
subject=item.get("subject", ""),
from_address=item.get("from", ""),
to=[item.get("mailbox", "")],
received_at=item.get("date"),
)
for item in items
]
async def get_email_content(self, inbox_id: str, email_id: str) -> Email:
async with self._client() as client:
resp = await client.get(
f"{self.base_url}/api/v1/mailbox/{inbox_id}/{email_id}"
)
resp.raise_for_status()
data = resp.json()
body = data.get("body", {})
return Email(
id=email_id,
inbox_id=inbox_id,
subject=data.get("subject", ""),
from_address=data.get("from", ""),
to=[data.get("mailbox", "")],
received_at=data.get("date"),
html=body.get("html"),
text=body.get("text"),
)
async def extract_links(self, inbox_id: str, email_id: str) -> List[str]:
email = await self.get_email_content(inbox_id, email_id)
if not email.html:
return []
return _extract_links_from_html(email.html)
async def delete_inbox(self, inbox_id: str) -> bool:
async with self._client() as client:
resp = await client.delete(f"{self.base_url}/api/v1/mailbox/{inbox_id}")
logger.info(f"[Inbucket] Deleted mailbox {inbox_id}: {resp.status_code}")
return resp.status_code in (200, 204)
def create_human_email(self, domain: Optional[str] = None) -> str:
return super().create_human_email(domain or self._domain)
# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------
def get_email_provider(provider_name: Optional[str] = None) -> EmailProvider:
"""
Return an EmailProvider instance.
Auto-detection order (when provider_name is None):
1. MAILSLURP_API_KEY → MailSlurpProvider
2. AGENTMAIL_API_KEY → AgentMailProvider
3. INBUCKET_URL → InbucketProvider
4. fallback → InbucketProvider (localhost:9000)
Args:
provider_name: One of "mailslurp", "agentmail", "inbucket".
If None, auto-detects from environment.
"""
name = (provider_name or "").lower().strip()
if name == "mailslurp" or (not name and os.environ.get("MAILSLURP_API_KEY")):
logger.info("EmailProvider: MailSlurp")
return MailSlurpProvider()
if name == "agentmail" or (not name and os.environ.get("AGENTMAIL_API_KEY")):
logger.info("EmailProvider: AgentMail")
return AgentMailProvider()
if name == "inbucket" or not name:
url = os.environ.get("INBUCKET_URL", "http://localhost:9000")
logger.info(f"EmailProvider: Inbucket ({url})")
return InbucketProvider(base_url=url)
raise ValueError(
f"Unknown provider: {provider_name!r}. "
"Choose from: mailslurp, agentmail, inbucket"
)
# ---------------------------------------------------------------------------
# Helper functions
# ---------------------------------------------------------------------------
async def wait_for_email(
provider: EmailProvider,
inbox_id: str,
subject_contains: Optional[str] = None,
timeout: int = 60,
poll_interval: float = 3.0,
) -> Email:
"""
Poll an inbox until a matching email arrives or timeout elapses.
Args:
provider: An EmailProvider instance.
inbox_id: The inbox to poll.
subject_contains: Optional substring to match against email subjects
(case-insensitive). If None, returns the first email.
timeout: Maximum seconds to wait before raising TimeoutError.
poll_interval: Seconds between polls.
Returns:
The first Email matching the filter.
Raises:
TimeoutError: If no matching email arrives within timeout.
"""
deadline = time.monotonic() + timeout
attempt = 0
while time.monotonic() < deadline:
attempt += 1
emails = await provider.get_emails(inbox_id)
for summary in emails:
if subject_contains is None:
# Fetch full content and return immediately
return await provider.get_email_content(inbox_id, summary.id)
if subject_contains.lower() in (summary.subject or "").lower():
return await provider.get_email_content(inbox_id, summary.id)
remaining = deadline - time.monotonic()
logger.debug(
f"wait_for_email attempt {attempt}: no match yet "
f"({len(emails)} emails, {remaining:.0f}s remaining)"
)
await asyncio.sleep(min(poll_interval, remaining))
raise TimeoutError(
f"No email matching subject_contains={subject_contains!r} "
f"arrived in inbox {inbox_id} within {timeout}s"
)
def extract_verification_link(email_html: str) -> Optional[str]:
"""
Parse email HTML and return the first verification/confirmation URL found.
Looks for links whose text or href contains keywords like:
verify, confirm, activate, reset, click here, get started, etc.
Args:
email_html: Raw HTML string of the email body.
Returns:
The first matching URL, or None if not found.
"""
VERIFICATION_KEYWORDS = re.compile(
r"verif|confirm|activat|reset|click.here|get.started|validate|unsubscribe",
re.IGNORECASE,
)
URL_PATTERN = re.compile(
r'href=["\']([^"\']+)["\']',
re.IGNORECASE,
)
# Strategy 1: find tags whose href contains keyword patterns
anchor_pattern = re.compile(
r']*href=["\']([^"\']+)["\'][^>]*>(.*?)',
re.IGNORECASE | re.DOTALL,
)
for match in anchor_pattern.finditer(email_html):
href, link_text = match.group(1), match.group(2)
if VERIFICATION_KEYWORDS.search(href) or VERIFICATION_KEYWORDS.search(link_text):
url = href.strip()
if url.startswith("http"):
return url
# Strategy 2: return first http URL that looks like a verification link
for href_match in URL_PATTERN.finditer(email_html):
href = href_match.group(1).strip()
if href.startswith("http") and VERIFICATION_KEYWORDS.search(href):
return href
# Strategy 3: return first http link (fallback)
for href_match in URL_PATTERN.finditer(email_html):
href = href_match.group(1).strip()
if href.startswith("http"):
return href
return None
def create_human_email(domain: str = "gmail.com") -> str:
"""
Generate a realistic human-style email address on the given domain.
Examples:
create_human_email() → "emma.johnson45@gmail.com"
create_human_email("yahoo.com") → "jsmith72@yahoo.com"
This is a standalone helper; individual providers also expose
this via EmailProvider.create_human_email() using their own domain.
"""
provider = _NullProvider()
return provider.create_human_email(domain)
class _NullProvider(EmailProvider):
"""Internal stub to reuse create_human_email without a real provider."""
async def create_inbox(self, name=None): # type: ignore[override]
raise NotImplementedError
async def get_emails(self, inbox_id): # type: ignore[override]
raise NotImplementedError
async def get_email_content(self, inbox_id, email_id): # type: ignore[override]
raise NotImplementedError
async def extract_links(self, inbox_id, email_id): # type: ignore[override]
raise NotImplementedError
async def delete_inbox(self, inbox_id): # type: ignore[override]
raise NotImplementedError
# ---------------------------------------------------------------------------
# Self-test (runs when module is executed directly)
# ---------------------------------------------------------------------------
async def _run_self_test() -> None:
print("\n=== Email Agent Self-Test ===\n")
passed = 0
failed = 0
# --- Test: create_human_email standalone ---
print("Test 1: create_human_email()")
for domain in ["gmail.com", "yahoo.com", "mailslurp.com"]:
addr = create_human_email(domain)
assert "@" in addr and domain in addr, f"Bad address: {addr}"
assert addr.split("@")[0].replace(".", "").replace("_", "").isalnum(), \
f"Non-alphanum local part: {addr}"
print(f" {addr}")
print(" PASSED\n")
passed += 1
# --- Test: _extract_links_from_html ---
print("Test 2: _extract_links_from_html()")
sample_html = """
Verify your email
Home
Contact
"""
links = _extract_links_from_html(sample_html)
assert len(links) == 3, f"Expected 3 links, got {len(links)}"
assert "https://example.com/verify?token=abc123" in links
print(f" Links found: {links}")
print(" PASSED\n")
passed += 1
# --- Test: extract_verification_link ---
print("Test 3: extract_verification_link()")
verification_html = """
Click below to verify your account:
Verify your email
Back to site
"""
link = extract_verification_link(verification_html)
assert link == "https://app.example.com/verify?token=xyz789&user=42", \
f"Wrong link: {link}"
print(f" Extracted: {link}")
print(" PASSED\n")
passed += 1
# --- Test: extract_verification_link fallback ---
print("Test 4: extract_verification_link() fallback (no keyword in text)")
plain_html = """
Click here
"""
link2 = extract_verification_link(plain_html)
assert link2 == "https://accounts.example.com/confirm/abc", f"Wrong: {link2}"
print(f" Extracted: {link2}")
print(" PASSED\n")
passed += 1
# --- Test: extract_verification_link returns None on empty ---
print("Test 5: extract_verification_link() returns None for no links")
result = extract_verification_link("No links here")
assert result is None, f"Expected None, got: {result}"
print(" Result: None")
print(" PASSED\n")
passed += 1
# --- Test: get_email_provider auto-detection ---
print("Test 6: get_email_provider() auto-detection")
# Clear env vars to force Inbucket fallback
for var in ("MAILSLURP_API_KEY", "AGENTMAIL_API_KEY"):
os.environ.pop(var, None)
provider = get_email_provider()
assert isinstance(provider, InbucketProvider), \
f"Expected InbucketProvider, got {type(provider)}"
print(f" Auto-detected: {type(provider).__name__} (no API keys set → Inbucket)")
print(" PASSED\n")
passed += 1
# --- Test: get_email_provider explicit name ---
print("Test 7: get_email_provider('inbucket')")
p = get_email_provider("inbucket")
assert isinstance(p, InbucketProvider)
print(f" Got: {type(p).__name__}")
print(" PASSED\n")
passed += 1
# --- Test: provider create_human_email uses own domain ---
print("Test 8: Provider.create_human_email() domain")
inbucket_p = InbucketProvider(base_url="http://mail.internal:9000")
addr = inbucket_p.create_human_email()
assert "mail.internal" in addr, f"Expected mail.internal domain: {addr}"
print(f" Inbucket address: {addr}")
mailslurp_p = MailSlurpProvider.__new__(MailSlurpProvider)
mailslurp_p.api_key = "dummy"
mailslurp_p._headers = {}
ms_addr = mailslurp_p.create_human_email()
assert "mailslurp.com" in ms_addr, f"Expected mailslurp.com: {ms_addr}"
print(f" MailSlurp address: {ms_addr}")
print(" PASSED\n")
passed += 1
# --- Test: wait_for_email raises TimeoutError on mock ---
print("Test 9: wait_for_email() raises TimeoutError when inbox empty")
class _EmptyProvider(InbucketProvider):
def __init__(self):
self.base_url = "http://localhost:9000"
self._domain = "localhost"
async def get_emails(self, inbox_id):
return []
async def get_email_content(self, inbox_id, email_id):
raise NotImplementedError
empty = _EmptyProvider()
try:
await wait_for_email(empty, "test_inbox", timeout=2, poll_interval=0.5)
print(" FAILED — should have raised TimeoutError")
failed += 1
except TimeoutError as e:
print(f" Got TimeoutError as expected: {e}")
print(" PASSED\n")
passed += 1
# --- Test: wait_for_email returns email on match ---
print("Test 10: wait_for_email() returns matching email")
sample_email = Email(
id="e001",
inbox_id="box1",
subject="Please verify your account",
from_address="noreply@app.com",
to=["agent@test.com"],
received_at="2026-02-27T00:00:00Z",
html="Verify",
)
class _PopulatedProvider(InbucketProvider):
def __init__(self):
self.base_url = "http://localhost:9000"
self._domain = "localhost"
async def get_emails(self, inbox_id):
return [sample_email]
async def get_email_content(self, inbox_id, email_id):
return sample_email
pop = _PopulatedProvider()
result_email = await wait_for_email(
pop, "box1", subject_contains="verify", timeout=5
)
assert result_email.id == "e001"
assert "verify" in result_email.subject.lower()
print(f" Received: {result_email.subject}")
print(" PASSED\n")
passed += 1
# --- Test: InbucketProvider live ping (optional) ---
print("Test 11: InbucketProvider connectivity (skipped if not running)")
inbucket_url = os.environ.get("INBUCKET_URL", "http://localhost:9000")
try:
async with httpx.AsyncClient(timeout=3) as client:
resp = await client.get(f"{inbucket_url}/api/v1/status")
if resp.status_code == 200:
print(f" Inbucket LIVE at {inbucket_url}: {resp.json()}")
print(" PASSED\n")
passed += 1
else:
print(f" Inbucket responded {resp.status_code} — skipped")
except Exception as exc:
print(f" Inbucket not reachable ({exc}) — skipped (not a failure)\n")
# --- Summary ---
print(f"{'=' * 40}")
print(f"Results: {passed} passed, {failed} failed")
if failed:
print("SOME TESTS FAILED")
else:
print("ALL TESTS PASSED")
print(f"{'=' * 40}\n")
if __name__ == "__main__":
asyncio.run(_run_self_test())