Python smtplib in Production: Timeouts, Retry, Idempotency

Python's smtplib is a wonderfully simple module. You connect, you call send_message, you get a response. That simplicity hides a stack of operational landmines: no default timeout, no retry, no idempotency, no connection reuse, no graceful TLS upgrade. Every one of these will surface eventually in production.

This article shows you the pattern we use for production Python email senders: explicit timeouts, classified retries with backoff, idempotency keys, connection pooling, structured logging. Copy-paste ready.

The Baseline Anti-Pattern

import smtplib
from email.message import EmailMessage

msg = EmailMessage()
msg["From"] = "noreply@example.com"
msg["To"] = "user@example.com"
msg["Subject"] = "Test"
msg.set_content("Hello")

with smtplib.SMTP("smtp.targetsmtp.it", 587) as s:
    s.starttls()
    s.login(USER, PASS)
    s.send_message(msg)

This works in a notebook. In production:

• No timeout → hung process when the SMTP server is slow.
• No retry → one transient blip and the email is lost.
• No idempotency → re-running drops a duplicate.
• One connection per send → handshake cost on every message.
• No structured log → debugging at 3 AM is painful.

Step 1: Timeouts

Always pass timeout to the constructor. Without it, smtplib uses the OS default, which is typically infinite.

s = smtplib.SMTP("smtp.targetsmtp.it", 587, timeout=10)

The timeout applies to all socket operations. 10 seconds is generous; 5 is fine for healthy infrastructure.

Step 2: TLS Done Right

Use SMTP_SSL for implicit TLS (port 465), or SMTP + starttls() for STARTTLS (port 587). Verify the cert.

import ssl

ctx = ssl.create_default_context()
ctx.minimum_version = ssl.TLSVersion.TLSv1_2

# STARTTLS
s = smtplib.SMTP("smtp.targetsmtp.it", 587, timeout=10)
s.starttls(context=ctx)
s.login(USER, PASS)

⚠️ Warning: Never disable cert verification. ssl._create_unverified_context() is a security incident waiting to happen.

Step 3: Retry With Classification

SMTP errors come back as SMTPResponseException subclasses or as plain SMTPException. We want to retry transient failures (4xx) and connection problems, but not permanent failures (5xx).

import smtplib, socket, time, random, logging
from email.message import EmailMessage

log = logging.getLogger("mailer")

TRANSIENT_ERRORS = (
    smtplib.SMTPServerDisconnected,
    smtplib.SMTPConnectError,
    smtplib.SMTPHeloError,
    socket.timeout,
    ConnectionError,
)

def classify(exc):
    if isinstance(exc, TRANSIENT_ERRORS):
        return "transient"
    if isinstance(exc, smtplib.SMTPResponseException):
        code = exc.smtp_code
        if 400 <= code < 500:
            return "transient"
        if 500 <= code < 600:
            return "permanent"
    return "unknown"

def send_with_retry(msg, max_attempts=5):
    for attempt in range(1, max_attempts + 1):
        try:
            return _send_once(msg)
        except Exception as e:
            kind = classify(e)
            log.warning("send failed", extra={
                "attempt": attempt, "kind": kind, "error": str(e)
            })
            if kind == "permanent" or attempt == max_attempts:
                raise
            delay = min(60, 2 ** attempt) * random.uniform(0.8, 1.2)
            time.sleep(delay)

Step 4: Idempotency

Raw SMTP has no idempotency primitive. You must add one yourself or use the provider's REST API. The simplest pattern: keep a small cache of "already sent" message identifiers.

def send_idempotent(msg, idem_key, cache):
    if cache.exists(idem_key):
        log.info("duplicate suppressed", extra={"key": idem_key})
        return cache.get(idem_key)
    result = send_with_retry(msg)
    cache.set(idem_key, result, ttl=86400)
    return result

Cache can be Redis, Memcached, or a small SQLite table. TTL of 24 hours is enough for retries from worker restarts; longer is harmless.

Step 5: Connection Pooling

For volume above ~10 messages/second per process you want to reuse the SMTP connection. The connection is not thread-safe, so the simplest pattern is a queue of connections.

import queue, threading

class SMTPPool:
    def __init__(self, size=5, **smtp_args):
        self.q = queue.Queue(maxsize=size)
        self.args = smtp_args
        for _ in range(size):
            self.q.put(self._open())

    def _open(self):
        s = smtplib.SMTP(timeout=10, **self.args)
        s.starttls(context=ssl.create_default_context())
        s.login(USER, PASS)
        return s

    def send(self, msg):
        conn = self.q.get()
        try:
            conn.send_message(msg)
        except (smtplib.SMTPServerDisconnected, ConnectionError):
            try: conn.quit()
            except Exception: pass
            conn = self._open()
            conn.send_message(msg)
        finally:
            self.q.put(conn)

Limit pool size: receivers throttle per IP. 5-10 connections is plenty for most.

Step 6: Structured Logging

Use a structured logger so 3 AM debugging is grep-friendly:

log.info("send ok", extra={
    "to": msg["To"],
    "subject": msg["Subject"],
    "message_id": msg["Message-ID"],
    "queue_id": queue_id,
    "duration_ms": int((time.time() - start) * 1000),
})

Always log Message-ID. It is the only stable identifier across your logs and the receiver's.

Step 7: Headers That Matter

import uuid

msg["Message-ID"] = f"<{uuid.uuid4()}@example.com>"
msg["Feedback-ID"] = f"transactional:invoice:acme:invoice-{invoice.id}"
msg["X-Entity-Ref-ID"] = str(invoice.id)
msg["List-Unsubscribe"] = ", "
msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"

The List-Unsubscribe-Post + List-Unsubscribe pair is required by Gmail and Yahoo for bulk senders since February 2024. It applies to anything you ship at scale.

Step 8: Async

For high-throughput async services use aiosmtplib with the same patterns. The retry/idempotency logic is identical, just awaited.

import aiosmtplib

async def send_async(msg):
    async with aiosmtplib.SMTP(hostname="smtp.targetsmtp.it",
                               port=587,
                               start_tls=True,
                               timeout=10) as s:
        await s.login(USER, PASS)
        await s.send_message(msg)

Step 9: Test Harness

Test your retry logic with a fake SMTP server. aiosmtpd is excellent for this. Spin one up that returns 421 (transient) on the first two attempts then 250 OK, and assert your retry logic recovers.

The Production Wrapper

Putting it together, the production-ready send function:

def send_production(to, subject, body, idem_key=None, pool=None):
    msg = EmailMessage()
    msg["From"] = FROM_ADDR
    msg["To"] = to
    msg["Subject"] = subject
    msg["Message-ID"] = f"<{uuid.uuid4()}@example.com>"
    msg.set_content(body)

    if idem_key and cache.exists(idem_key):
        return {"status": "duplicate"}

    try:
        pool.send(msg) if pool else send_with_retry(msg)
    except smtplib.SMTPResponseException as e:
        if 500 <= e.smtp_code < 600:
            mark_hard_bounce(to)
        raise
    if idem_key:
        cache.set(idem_key, "sent", ttl=86400)
    return {"status": "sent", "message_id": msg["Message-ID"]}

Closing

Python smtplib is a low-level tool. Production code wraps it in classification, retry, idempotency and pooling. Once you have that wrapper, the underlying provider can switch without touching anything else. Target SMTP exposes both raw SMTP and a REST API; the REST API gives you native idempotency keys and lifts the retry burden off your code entirely. The Send-Time Firewall applies the same policies to both transports, so a missing idempotency key on the SMTP side does not produce duplicate sends to the same recipient.