PythonでメールからOTPの自動取得

サインアップフローが確認メールを送るたびに、ボタンパイプラインの処理が停止する—コピーし、そのコードをどこかに入れるまで待ちます。開けるためのインストゥルメントはデバッグ者、テストパイプライン、サインアップ自動化スクリプトがあり、AIアシスタントで動作させるためのアカウント確認が必要です。この問題にはボタンを介さない方法があります。オーバーラップが存在するように見えるメールからパスワード再取得用のOTPを自動的に取得することで、さまざまな障害と不確実性により、より容易に達成できます。 リアルなメールはスパムや共有フォルダによる汚染を防ぎ、Gmail IMAP APIでは制限時間を有し、OAuthで接続が難しくなるため、プログラム化アクセスは維持するのが難しいです。これによって一貫性のないスパンを生み出します。 ベストな方法となるのはリソース間のより複雑で脆弱なものを作ります。代替として、リソースの一つである代替アカウント管理プラットフォームを使用することで解決策が見つかりました。外部SMTP送信者 │ ▼ AWS SES (Spam/virusフィルター) │ ▼ SNSウェブhooks POST /api/v1/ses/inbound │▼ core/delivery.deliver_raw_email() │ ┌────┴────┐ 組織とテストが利用可能な一時的なメールアドレス @uncorreotemporal.comへの完全なプログラマブルのメールインクルームを提供する UnCorreoTemporal は理想的なプラットフォームです。それはドメインとSMTP実体とSMTPサーバーで異なるものが必要でいません。そのようなインフラストラクチャーが一つだけであるため、テストやデバッグでのプログラム化操作を効率的に行うことができます。このようにして送信可能なメッセージはすべて受信済みトレイに自動的に取り込まれます。また、リアルタイム接続とRFC 2882のメール体内容とHTML本文はスクリプトを使用せず、標準インポートシステムを使用します。これは、ウェブソケットとPOSTリクエストだけで行われるため高速化を可能にしました。 リソース間で個別オーガナイズすると、コードが混乱するかもしれませんが、本番環境ではそれが助ける可能性があります。外部アカウント送信者 │ ▼ AWS SES (スパム/悪用フィルター) │▼ SNSウェブhooks POST /api/v1/ses/inbound │→ core/delivery.deliver_raw_email() │← core/delivery.py データポストを読み取る、処理するメールインクルームを作成します。その結果、データベースとローカル aioSMTPSサーバーにはメッセージが直接取り込まれます。これは実際のSMTP通信を使って送信できるためです。AWS SES APIからのウェブhooksを使用して、この情報を送受信できます。それ以外にも、テスト環境ではテスト用IMAPサーバーを立ち上げてSMTPデータを処理することで、同じ機能を持っていました。 デザインの一部として、WebSocket経由で個々の受信メールについて通知することが予定されており、リソース間での動作が困難になると予想されます。これら3つの要素の整合性は非常に複雑であり、それぞれのメッセージをリアルタイムに送受信できるため、これは代替アカウントの管理プラットフォームとは異なる一意のシステムになります。 これらの情報により、個々のメール通信に対する正確な同期が可能になります。それだけでなく、これからのアプリケーションは各トレイに対する特定のスキャンや分析を行うように自動化できます

Automating OTP Retrieval from Emails in Python Every time a signup flow sends a verification email, an automation pipeline stalls — waiting for a human to open their inbox, copy a six-digit code, and paste it somewhere. For developers, this bottleneck shows up in: E2E test suites that exercise account registration flows CI/CD pipelines that provision test accounts on every build Signup automation scripts that need to complete email verification AI agents that create accounts autonomously and need to confirm them The problem isn't just inconvenience. Real email addresses accumulate spam, shared inboxes cause test pollution, and Gmail's IMAP API has rate limits and OAuth friction that make programmatic access painful to maintain. You end up with flaky hacks — sleeping for five seconds and hoping the email arrived. There's a better way. Using a real inbox for automated OTP retrieval introduces several failure modes: Shared inbox pollution. If your test suite uses qa@yourcompany.com, parallel test runs receive emails from multiple jobs into the same inbox, making it impossible to reliably match an OTP to a specific test run. Gmail IMAP fragility. Gmail's IMAP access requires OAuth2 tokens, app passwords, or Less Secure App settings — all of which require manual setup, expire, and get blocked by workspace policies. Polling via IMAP also adds 5–30 seconds of latency. Regex-on-raw-MIME hell. If you do get IMAP working, you're parsing RFC 2822 messages with base64 parts, quoted-printable encoding, and multipart boundaries. One encoding change in the sender's template breaks your parser. Rate limits and delivery delays. Cloud email providers throttle connections, and deliverability adds unpredictable latency. A test that needs to run in under 30 seconds becomes a 90-second flake. A programmable temporary email infrastructure solves all of this by giving you: On-demand inbox creation via a REST API — one per test, discarded afterwards Real SMTP reception — actual email protocols, not simulation Instant message retrieval — read the body milliseconds after delivery No authentication overhead — anonymous inboxes with a session token, or API key access for CI UnCorreoTemporal is exactly this kind of infrastructure. It provides temporary email addresses at @uncorreotemporal.com, accepts real SMTP delivery, stores messages in a PostgreSQL database, and exposes them through a REST API and a WebSocket stream. You get a full programmable email inbox in a single HTTP call. Understanding how the system works helps you use it correctly in production automation. 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) (mailbox:{address}) │ ▼ WebSocket /ws/inbox/{address} {"event": "new_message", "message_id": "..."} The deliver_raw_email() function in core/delivery.py is the shared ingestion point — used by both the AWS SES webhook (api/routers/ses_inbound.py) and the local aiosmtpd-based SMTP server (smtp/handler.py). Once an email is stored, it publishes a Redis event on the channel mailbox:{address}, which the WebSocket endpoint in ws/inbox.py forwards to any connected clients. Messages are stored with parsed body_text and body_html fields extracted via Python's standard email library, plus the full RFC 2822 bytes for re-parsing. You work directly with decoded strings — no MIME handling on your end. The simplest call creates an anonymous inbox with no authentication required. The response includes a session_token that authenticates all subsequent operations on that mailbox. import requests BASE_URL = "https://api.uncorreotemporal.com/api/v1" 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(address) # mango-panda-42@uncorreotemporal.com For CI/CD where you have an API key, use Bearer token auth instead: headers = {"Authorization": "Bearer uct_your_api_key_here"} resp = requests.post(f"{BASE_URL}/mailboxes", headers=headers) The address format (adjective-noun-number@uncorreotemporal.com) is human-readable and collision-resistant. Inboxes expire automatically at expires_at — no cleanup code required. After triggering your signup flow, poll the messages endpoint until an email arrives. import time def wait_for_email( address: str, session_token: str, timeout: int = 30, poll_interval: float = 1.5, ) -> dict: headers = {"X-Session-Token": session_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(poll_interval) raise TimeoutError(f"No email received within {timeout}s") Real-time alternative: Connect to the WebSocket endpoint for push-based delivery notification: import asyncio, json import websockets async def wait_for_email_ws(address: str, session_token: str) -> str: url = f"wss://api.uncorreotemporal.com/ws/inbox/{address}?token={session_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"] OTP codes in verification emails are almost always 6 or 8 isolated digits. A targeted regex on body_text is reliable and fast: import re def extract_otp(body: str) -> str: match = re.search(r"(? 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"] def trigger_signup(email: str) -> None: requests.post("https://yourapp.com/api/signup", json={"email": email}) def wait_for_message(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("Email not received in time") def extract_otp(body: str) -> str: match = re.search(r"(?\d)(\d{6})(?!\d)", body) if not match: raise ValueError("No OTP found") return match.group(1) if __name__ == "__main__": address, token = create_inbox(ttl_minutes=5) print(f"Inbox: {address}") trigger_signup(email=address) message = wait_for_message(address, token, timeout=30) otp = extract_otp(message["body_text"]) print(f"OTP: {otp}") import pytest, requests BASE_URL = "https://api.uncorreotemporal.com/api/v1" @pytest.fixture() def temp_inbox(): resp = requests.post(f"{BASE_URL}/mailboxes", params={"ttl_minutes": 10}) resp.raise_for_status() data = resp.json() yield data["address"], data["session_token"] def test_signup_sends_otp(temp_inbox, page): address, token = temp_inbox page.goto("https://yourapp.com/signup") page.fill('[name="email"]', address) page.click('[type="submit"]') message = wait_for_message(address, token, timeout=30) otp = extract_otp(message["body_text"]) page.fill('[name="otp"]', otp) page.click('[type="submit"]') page.wait_for_url("**/dashboard") - name: Run E2E tests env: UCT_API_KEY: ${{ secrets.UCT_API_KEY }} run: pytest tests/e2e/ -v UnCorreoTemporal ships an MCP server that exposes inbox management directly as tools callable by Claude Desktop and other MCP-compatible agents: { "mcpServers": { "uncorreotemporal": { "command": "python", "args": ["-m", "uncorreotemporal.mcp.server"], "env": { "UCT_API_KEY": "uct_your_key" } } } } With this configured, an agent can create a mailbox, complete a registration flow, retrieve the verification email, extract the OTP, and proceed — all without human intervention. Automated OTP retrieval from email doesn't have to be fragile. The combination of on-demand inbox creation, real SMTP delivery, and a clean REST API removes every layer of indirection that makes traditional email testing painful. UnCorreoTemporal is built for this use case — async FastAPI, PostgreSQL, Redis-driven WebSocket push, and AWS SES. The API surface is small enough to integrate in an afternoon. Try the API at uncorreotemporal.com. Your first inbox is one HTTP call away. Originally published at uncorreotemporal.com