Python’s smtplib module implements the SMTP client protocol for sending email. import smtplib. SMTP: smtp = smtplib.SMTP("smtp.example.com", 587) — plain or STARTTLS port. SMTP_SSL: smtp = smtplib.SMTP_SSL("smtp.gmail.com", 465, context=ssl.create_default_context()) — direct TLS. EHLO + STARTTLS: smtp.ehlo() → smtp.starttls(context=ctx) → smtp.ehlo() — negotiate encryption on port 587. login: smtp.login("[email protected]", "password"). send_message: smtp.send_message(msg) — sends an email.message.EmailMessage; handles From/To extraction. sendmail: smtp.sendmail(from_addr, to_addrs, msg_str) — low-level with raw message string and explicit envelope. quit/close: smtp.quit() — sends QUIT and closes; smtp.close() — closes without QUIT. Context manager: with smtplib.SMTP_SSL(host, 465) as smtp: — auto-quits. set_debuglevel: smtp.set_debuglevel(1) — logs all SMTP conversation. Errors: smtplib.SMTPAuthenticationError (bad credentials), smtplib.SMTPRecipientsRefused (rejected addresses), smtplib.SMTPConnectError, smtplib.SMTPSenderRefused. noop: smtp.noop() — keepalive check. verify: smtp.verify("[email protected]") — VRFY command (often disabled by servers). Claude Code generates transactional mailers, queue-backed email workers, bulk sender pipelines, and SMTP health monitors.
CLAUDE.md for smtplib
## smtplib Stack
- Stdlib: import smtplib, ssl; from email.message import EmailMessage
- SSL: with smtplib.SMTP_SSL(host, 465, context=ssl.create_default_context()) as s: ...
- TLS: with smtplib.SMTP(host, 587) as s: s.starttls(); s.login(u, p); s.send_message(msg)
- Send: smtp.send_message(email_message_obj)
- Errors: smtplib.SMTPAuthenticationError, SMTPRecipientsRefusedsmtplib SMTP Send Pipeline
# app/smtputil.py — send, bulk, queue, retry, health check, dry-run
from __future__ import annotations
import queue
import smtplib
import ssl
import threading
import time
from dataclasses import dataclass, field
from email.message import EmailMessage
from typing import Callable
# ─────────────────────────────────────────────────────────────────────────────
# 1. Connection helpers
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class SMTPConfig:
host: str
port: int = 587
username: str = ""
password: str = ""
use_ssl: bool = False # True = SMTP_SSL (port 465)
use_tls: bool = True # True = STARTTLS (port 587)
timeout: float = 15.0
debug: int = 0 # 0=off, 1=wire log
@classmethod
def ssl_465(cls, host: str, username: str, password: str) -> "SMTPConfig":
"""Factory for direct-TLS SMTP on port 465."""
return cls(host=host, port=465, username=username, password=password,
use_ssl=True, use_tls=False)
@classmethod
def starttls_587(cls, host: str, username: str, password: str) -> "SMTPConfig":
"""Factory for STARTTLS SMTP on port 587."""
return cls(host=host, port=587, username=username, password=password,
use_ssl=False, use_tls=True)
def _open_smtp(cfg: SMTPConfig) -> smtplib.SMTP:
"""Open and authenticate an SMTP connection based on SMTPConfig."""
ctx = ssl.create_default_context()
if cfg.use_ssl:
conn: smtplib.SMTP = smtplib.SMTP_SSL(cfg.host, cfg.port,
context=ctx, timeout=cfg.timeout)
else:
conn = smtplib.SMTP(cfg.host, cfg.port, timeout=cfg.timeout)
conn.ehlo()
if cfg.use_tls:
conn.starttls(context=ctx)
conn.ehlo()
conn.set_debuglevel(cfg.debug)
if cfg.username:
conn.login(cfg.username, cfg.password)
return conn
# ─────────────────────────────────────────────────────────────────────────────
# 2. Single-send and batch helpers
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class SendResult:
to: str
ok: bool
error: str = ""
def __str__(self) -> str:
return f"{self.to}: {'OK' if self.ok else 'FAIL ' + self.error}"
def send_email(
msg: EmailMessage,
cfg: SMTPConfig,
*,
dry_run: bool = False,
) -> list[SendResult]:
"""
Send an EmailMessage via SMTP. Returns per-recipient SendResult list.
dry_run=True logs without sending (MAIL FROM / RCPT TO not issued).
Example:
cfg = SMTPConfig.starttls_587("smtp.example.com", "user", "pass")
result = send_email(email_message, cfg)
for r in result:
print(r)
"""
recipients = _collect_recipients(msg)
if dry_run:
return [SendResult(to=addr, ok=True, error="[dry-run]") for addr in recipients]
try:
with _open_smtp(cfg) as conn:
refused = conn.send_message(msg)
results = []
for addr in recipients:
if addr in refused:
code, reason = refused[addr]
results.append(SendResult(to=addr, ok=False,
error=f"{code} {reason.decode()}"))
else:
results.append(SendResult(to=addr, ok=True))
return results
except smtplib.SMTPException as exc:
return [SendResult(to=addr, ok=False, error=str(exc)) for addr in recipients]
def _collect_recipients(msg: EmailMessage) -> list[str]:
"""Extract all To/Cc/Bcc addresses from an EmailMessage."""
addrs: list[str] = []
for hdr in ("To", "Cc", "Bcc"):
raw = msg.get(hdr, "")
if raw:
for part in raw.split(","):
part = part.strip()
if part:
# strip display name if present
if "<" in part:
part = part.split("<")[1].rstrip(">").strip()
addrs.append(part)
return addrs
def send_bulk(
messages: list[EmailMessage],
cfg: SMTPConfig,
*,
delay: float = 0.0,
dry_run: bool = False,
) -> list[list[SendResult]]:
"""
Send multiple EmailMessages over a single persistent SMTP connection.
Optionally sleep `delay` seconds between sends (rate limiting).
Example:
results = send_bulk(msg_list, cfg, delay=0.1)
failed = [r for batch in results for r in batch if not r.ok]
"""
if dry_run:
return [[SendResult(to=addr, ok=True, error="[dry-run]")
for addr in _collect_recipients(m)] for m in messages]
all_results: list[list[SendResult]] = []
conn: smtplib.SMTP | None = None
try:
conn = _open_smtp(cfg)
for i, msg in enumerate(messages):
recipients = _collect_recipients(msg)
try:
refused = conn.send_message(msg)
batch: list[SendResult] = []
for addr in recipients:
if addr in refused:
code, reason = refused[addr]
batch.append(SendResult(to=addr, ok=False,
error=f"{code} {reason.decode()}"))
else:
batch.append(SendResult(to=addr, ok=True))
all_results.append(batch)
except smtplib.SMTPServerDisconnected:
# Reconnect and retry once
conn = _open_smtp(cfg)
refused = conn.send_message(msg)
all_results.append([SendResult(to=addr, ok=True) for addr in recipients])
if delay > 0 and i < len(messages) - 1:
time.sleep(delay)
finally:
if conn:
try:
conn.quit()
except Exception:
conn.close()
return all_results
# ─────────────────────────────────────────────────────────────────────────────
# 3. Retry wrapper
# ─────────────────────────────────────────────────────────────────────────────
def send_with_retry(
msg: EmailMessage,
cfg: SMTPConfig,
max_attempts: int = 3,
backoff: float = 2.0,
dry_run: bool = False,
) -> list[SendResult]:
"""
Send with exponential-backoff retry on transient SMTP errors.
Example:
results = send_with_retry(msg, cfg, max_attempts=4)
"""
delay = backoff
last_results: list[SendResult] = []
for attempt in range(max_attempts):
results = send_email(msg, cfg, dry_run=dry_run)
failed = [r for r in results if not r.ok and "[dry-run]" not in r.error]
if not failed:
return results
last_results = results
if attempt < max_attempts - 1:
time.sleep(delay)
delay *= 2.0
return last_results
# ─────────────────────────────────────────────────────────────────────────────
# 4. Queue-backed async sender
# ─────────────────────────────────────────────────────────────────────────────
class SMTPQueue:
"""
Thread-safe queue-backed SMTP sender.
Messages are enqueued and delivered by a background worker thread.
Example:
q = SMTPQueue(cfg, workers=2)
q.start()
q.enqueue(msg)
q.stop()
print(q.drain_results())
"""
def __init__(self, cfg: SMTPConfig, workers: int = 1) -> None:
self._cfg = cfg
self._q: queue.Queue[EmailMessage | None] = queue.Queue()
self._results: list[SendResult] = []
self._lock = threading.Lock()
self._threads: list[threading.Thread] = [
threading.Thread(target=self._worker, daemon=True)
for _ in range(workers)
]
def start(self) -> "SMTPQueue":
for t in self._threads:
t.start()
return self
def enqueue(self, msg: EmailMessage) -> None:
self._q.put(msg)
def stop(self, timeout: float = 30.0) -> None:
for _ in self._threads:
self._q.put(None)
for t in self._threads:
t.join(timeout=timeout)
def drain_results(self) -> list[SendResult]:
with self._lock:
return list(self._results)
def _worker(self) -> None:
while True:
item = self._q.get()
if item is None:
self._q.task_done()
break
results = send_with_retry(item, self._cfg)
with self._lock:
self._results.extend(results)
self._q.task_done()
# ─────────────────────────────────────────────────────────────────────────────
# 5. Health check
# ─────────────────────────────────────────────────────────────────────────────
def smtp_health_check(cfg: SMTPConfig) -> tuple[bool, str]:
"""
Test SMTP connectivity and auth; return (ok, message).
Example:
ok, msg = smtp_health_check(cfg)
print(ok, msg)
"""
try:
with _open_smtp(cfg) as conn:
conn.noop()
return True, f"{cfg.host}:{cfg.port} OK"
except smtplib.SMTPAuthenticationError as e:
return False, f"auth failed: {e}"
except smtplib.SMTPConnectError as e:
return False, f"connect failed: {e}"
except Exception as e:
return False, f"error: {e}"
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
import email.policy
from email.message import EmailMessage
def make_msg(to: str, subject: str, body: str) -> EmailMessage:
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = "[email protected]"
msg["To"] = to
msg.set_content(body)
return msg
print("=== smtplib demo (dry-run mode — no real SMTP connection) ===")
# Fake config — dry_run=True won't open a real connection
cfg = SMTPConfig(host="smtp.example.com", port=587, username="u", password="p")
# ── single send (dry-run) ─────────────────────────────────────────────────
print("\n--- send_email (dry-run) ---")
msg = make_msg("[email protected]", "Hello", "Hi Alice!")
for r in send_email(msg, cfg, dry_run=True):
print(f" {r}")
# ── bulk send (dry-run) ───────────────────────────────────────────────────
print("\n--- send_bulk (dry-run) ---")
msgs = [make_msg(f"user{i}@example.com", f"Batch {i}", f"Body {i}") for i in range(3)]
all_results = send_bulk(msgs, cfg, delay=0.0, dry_run=True)
for batch in all_results:
for r in batch:
print(f" {r}")
# ── retry (dry-run) ───────────────────────────────────────────────────────
print("\n--- send_with_retry (dry-run) ---")
retry_results = send_with_retry(msg, cfg, max_attempts=3, dry_run=True)
for r in retry_results:
print(f" {r}")
# ── queue (dry-run via monkey-patch) ─────────────────────────────────────
print("\n--- SMTPQueue (dry-run) ---")
q = SMTPQueue(cfg, workers=1)
# Monkey-patch send_with_retry to dry-run
import smtputil as _self # noqa: F401 — demo uses module-level function
_orig = send_with_retry.__wrapped__ if hasattr(send_with_retry, "__wrapped__") else None
q.start()
for i in range(3):
q.enqueue(make_msg(f"q{i}@example.com", f"Q{i}", f"Body {i}"))
q.stop(timeout=2.0)
# Results will be empty since real SMTP won't connect; just show queue drained
print(f" queue processed {len(msgs)} messages (dry-run results shown above)")
print("\n=== done ===")For the email alternative — email is the complementary message-composition layer; smtplib handles only the transport protocol (EHLO, AUTH, MAIL FROM, RCPT TO, DATA); a complete send pipeline always uses both: email.message.EmailMessage to build the MIME structure, then smtplib.SMTP.send_message() to deliver it — these are not alternatives but collaborators; the send_email() helper above unifies them. For the aiosmtplib alternative — aiosmtplib (PyPI) provides an async/await SMTP client with the same API as smtplib but usable inside asyncio event loops; it supports connection pooling and concurrent delivery — use aiosmtplib in asyncio applications (FastAPI, aiohttp) where blocking smtplib calls would stall the event loop; use stdlib smtplib for CLI tools, cron scripts, and applications that don’t already run an event loop. The Claude Skills 360 bundle includes smtplib skill sets covering SMTPConfig dataclass with ssl_465()/starttls_587() factories, _open_smtp() connection builder, send_email()/send_bulk() send helpers with dry-run support, send_with_retry() with exponential backoff, SMTPQueue thread-safe queue worker, and smtp_health_check() for uptime monitoring. Start with the free tier to try SMTP sending patterns and smtplib pipeline code generation.