|
#!/usr/bin/env python3 |
|
""" |
|
Mail.tm API demo: create inbox, send "loremipsum", read message back. |
|
|
|
Mail.tm (https://docs.mail.tm/) is a free disposable-email API for *receiving* |
|
mail. It does not expose an endpoint to send email (POST /messages returns 405). |
|
|
|
This script uses the Mail.tm REST API for: |
|
1. Creating a temporary address |
|
2. Polling the inbox |
|
3. Fetching full message content |
|
|
|
Outbound delivery is handled separately via HTTP (Brevo) or SMTP env vars, |
|
because Mail.tm cannot send mail on your behalf. |
|
|
|
Requirements: |
|
Python 3.10+ |
|
|
|
Optional outbound setup (pick one): |
|
- Brevo (free): BREVO_API_KEY + BREVO_SENDER_EMAIL |
|
- Generic SMTP: SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASS |
|
- Manual: pass --manual and send "loremipsum" yourself from any mail client |
|
|
|
Usage: |
|
python scripts/mailtm_test.py |
|
python scripts/mailtm_test.py --manual |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import json |
|
import os |
|
import secrets |
|
import smtplib |
|
import sys |
|
import time |
|
import urllib.error |
|
import urllib.request |
|
from email.mime.text import MIMEText |
|
from typing import Any |
|
|
|
MAILTM_BASE_URL = "https://api.mail.tm" |
|
BREVO_SEND_URL = "https://api.brevo.com/v3/smtp/email" |
|
DEFAULT_SUBJECT = "Mail.tm test" |
|
DEFAULT_BODY = "loremipsum" |
|
POLL_INTERVAL_SECONDS = 2 |
|
POLL_TIMEOUT_SECONDS = 120 |
|
|
|
|
|
class MailTmError(RuntimeError): |
|
pass |
|
|
|
|
|
class MailTmClient: |
|
def __init__(self, base_url: str = MAILTM_BASE_URL) -> None: |
|
self.base_url = base_url.rstrip("/") |
|
self.token: str | None = None |
|
self.account_id: str | None = None |
|
self.address: str | None = None |
|
|
|
def _request( |
|
self, |
|
method: str, |
|
path: str, |
|
*, |
|
payload: dict[str, Any] | None = None, |
|
auth: bool = False, |
|
) -> Any: |
|
headers = { |
|
"Accept": "application/ld+json", |
|
"Content-Type": "application/json", |
|
} |
|
if auth: |
|
if not self.token: |
|
raise MailTmError("Not authenticated. Call login() first.") |
|
headers["Authorization"] = f"Bearer {self.token}" |
|
|
|
body = json.dumps(payload).encode("utf-8") if payload is not None else None |
|
request = urllib.request.Request( |
|
f"{self.base_url}{path}", |
|
data=body, |
|
headers=headers, |
|
method=method, |
|
) |
|
|
|
try: |
|
with urllib.request.urlopen(request) as response: |
|
raw = response.read() |
|
if not raw: |
|
return None |
|
return json.loads(raw) |
|
except urllib.error.HTTPError as exc: |
|
detail = exc.read().decode("utf-8", errors="replace") |
|
raise MailTmError(f"{method} {path} failed ({exc.code}): {detail}") from exc |
|
|
|
def get_domain(self) -> str: |
|
data = self._request("GET", "/domains") |
|
members = data.get("hydra:member", []) |
|
if not members: |
|
raise MailTmError("No active Mail.tm domains available.") |
|
return members[0]["domain"] |
|
|
|
def create_account(self, password: str | None = None) -> str: |
|
domain = self.get_domain() |
|
local_part = f"test{secrets.token_hex(4)}" |
|
password = password or secrets.token_urlsafe(16) |
|
address = f"{local_part}@{domain}" |
|
|
|
account = self._request( |
|
"POST", |
|
"/accounts", |
|
payload={"address": address, "password": password}, |
|
) |
|
self.account_id = account["id"] |
|
self.address = address |
|
self._password = password |
|
return address |
|
|
|
def login(self, address: str | None = None, password: str | None = None) -> str: |
|
address = address or self.address |
|
password = password or getattr(self, "_password", None) |
|
if not address or not password: |
|
raise MailTmError("Missing address/password for login().") |
|
|
|
token_response = self._request( |
|
"POST", |
|
"/token", |
|
payload={"address": address, "password": password}, |
|
) |
|
self.token = token_response["token"] |
|
self.address = address |
|
return self.token |
|
|
|
def list_messages(self) -> list[dict[str, Any]]: |
|
data = self._request("GET", "/messages", auth=True) |
|
return data.get("hydra:member", []) |
|
|
|
def get_message(self, message_id: str) -> dict[str, Any]: |
|
return self._request("GET", f"/messages/{message_id}", auth=True) |
|
|
|
def wait_for_message(self, timeout_seconds: int = POLL_TIMEOUT_SECONDS) -> dict[str, Any]: |
|
deadline = time.time() + timeout_seconds |
|
while time.time() < deadline: |
|
messages = self.list_messages() |
|
if messages: |
|
return self.get_message(messages[0]["id"]) |
|
time.sleep(POLL_INTERVAL_SECONDS) |
|
raise TimeoutError(f"No message received within {timeout_seconds}s.") |
|
|
|
|
|
def send_via_brevo(to_address: str, subject: str, body: str) -> None: |
|
api_key = os.environ.get("BREVO_API_KEY") |
|
sender_email = os.environ.get("BREVO_SENDER_EMAIL") |
|
if not api_key or not sender_email: |
|
raise MailTmError( |
|
"Set BREVO_API_KEY and BREVO_SENDER_EMAIL to send via Brevo's HTTP API." |
|
) |
|
|
|
payload = { |
|
"sender": {"email": sender_email, "name": "Mail.tm Test"}, |
|
"to": [{"email": to_address}], |
|
"subject": subject, |
|
"textContent": body, |
|
} |
|
request = urllib.request.Request( |
|
BREVO_SEND_URL, |
|
data=json.dumps(payload).encode("utf-8"), |
|
headers={ |
|
"Accept": "application/json", |
|
"Content-Type": "application/json", |
|
"api-key": api_key, |
|
}, |
|
method="POST", |
|
) |
|
try: |
|
with urllib.request.urlopen(request) as response: |
|
print(f"Brevo accepted message (HTTP {response.status}).") |
|
except urllib.error.HTTPError as exc: |
|
detail = exc.read().decode("utf-8", errors="replace") |
|
raise MailTmError(f"Brevo send failed ({exc.code}): {detail}") from exc |
|
|
|
|
|
def send_via_smtp(to_address: str, subject: str, body: str) -> None: |
|
host = os.environ.get("SMTP_HOST") |
|
port = int(os.environ.get("SMTP_PORT", "587")) |
|
user = os.environ.get("SMTP_USER") |
|
password = os.environ.get("SMTP_PASS") |
|
sender = os.environ.get("SMTP_FROM", user) |
|
|
|
if not all([host, user, password, sender]): |
|
raise MailTmError( |
|
"Set SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASS, and optionally SMTP_FROM." |
|
) |
|
|
|
message = MIMEText(body) |
|
message["Subject"] = subject |
|
message["From"] = sender |
|
message["To"] = to_address |
|
|
|
with smtplib.SMTP(host, port, timeout=30) as smtp: |
|
smtp.starttls() |
|
smtp.login(user, password) |
|
smtp.send_message(message) |
|
print(f"SMTP message sent via {host}:{port}.") |
|
|
|
|
|
def send_test_email(to_address: str, subject: str, body: str, manual: bool) -> None: |
|
if manual: |
|
print("\nMail.tm cannot send email via API.") |
|
print(f"Send an email with body '{body}' to: {to_address}") |
|
input("Press Enter after you have sent it...") |
|
return |
|
|
|
if os.environ.get("BREVO_API_KEY"): |
|
send_via_brevo(to_address, subject, body) |
|
return |
|
|
|
if os.environ.get("SMTP_HOST"): |
|
send_via_smtp(to_address, subject, body) |
|
return |
|
|
|
raise MailTmError( |
|
"No outbound method configured. Set BREVO_API_KEY + BREVO_SENDER_EMAIL, " |
|
"or SMTP_* variables, or run with --manual." |
|
) |
|
|
|
|
|
def parse_args() -> argparse.Namespace: |
|
parser = argparse.ArgumentParser(description="Mail.tm API send/receive demo.") |
|
parser.add_argument( |
|
"--manual", |
|
action="store_true", |
|
help="Pause and wait for you to send the test email manually.", |
|
) |
|
parser.add_argument( |
|
"--subject", |
|
default=DEFAULT_SUBJECT, |
|
help=f"Email subject (default: {DEFAULT_SUBJECT!r}).", |
|
) |
|
parser.add_argument( |
|
"--body", |
|
default=DEFAULT_BODY, |
|
help=f"Email body text (default: {DEFAULT_BODY!r}).", |
|
) |
|
parser.add_argument( |
|
"--timeout", |
|
type=int, |
|
default=POLL_TIMEOUT_SECONDS, |
|
help=f"Seconds to wait for inbound mail (default: {POLL_TIMEOUT_SECONDS}).", |
|
) |
|
return parser.parse_args() |
|
|
|
|
|
def main() -> int: |
|
args = parse_args() |
|
client = MailTmClient() |
|
|
|
try: |
|
address = client.create_account() |
|
client.login() |
|
print(f"Created Mail.tm inbox: {address}") |
|
|
|
send_test_email(address, args.subject, args.body, manual=args.manual) |
|
|
|
print(f"Waiting up to {args.timeout}s for inbound message...") |
|
message = client.wait_for_message(timeout_seconds=args.timeout) |
|
|
|
text_content = message.get("text") or "" |
|
html_content = message.get("html") |
|
if isinstance(html_content, list): |
|
html_content = "\n".join(html_content) |
|
|
|
print("\n--- Received message ---") |
|
print(f"From: {message.get('from', {}).get('address')}") |
|
print(f"Subject: {message.get('subject')}") |
|
print(f"Text: {text_content or '(empty)'}") |
|
if html_content: |
|
print(f"HTML: {html_content}") |
|
|
|
if args.body not in (text_content or "") and args.body not in (html_content or ""): |
|
print( |
|
f"\nWarning: expected body {args.body!r} was not found in the message.", |
|
file=sys.stderr, |
|
) |
|
return 1 |
|
|
|
print("\nSuccess: Mail.tm API flow completed.") |
|
return 0 |
|
except (MailTmError, TimeoutError, urllib.error.URLError) as exc: |
|
print(f"Error: {exc}", file=sys.stderr) |
|
return 1 |
|
|
|
|
|
if __name__ == "__main__": |
|
raise SystemExit(main()) |