Python’s poplib module implements the POP3 protocol for retrieving email from a mail server. import poplib. Connect: poplib.POP3(host, port=110, timeout=60) — cleartext; poplib.POP3_SSL(host, port=995, keyfile=None, certfile=None, context=None, timeout=60) — TLS. Auth: conn.user("alice") then conn.pass_("secret") — two-step; conn.apop("alice", "secret") — APOP challenge-response (server must support). Mailbox: conn.stat() → (msg_count, total_octets); conn.list(which=None) → (response, [b"1 3456", b"2 789", ...], octets) — list all or one message with sizes. Retrieve: conn.retr(which) → (response, [line_bytes, ...], octets) — full message; conn.top(which, maxlines) → same format, headers + N body lines only. UIDs: conn.uidl(which=None) → (response, [b"1 uid", ...], octets) — unique identifiers stable across sessions. Delete: conn.dele(which) — marks message; conn.rset() — cancels all pending deletes; conn.quit() — commits deletes and closes. Keepalive: conn.noop(). Use as context manager with with poplib.POP3_SSL(host) as conn:. Claude Code generates mail downloaders, duplicate detectors, inbox monitors, and message archive builders.
CLAUDE.md for poplib
## poplib Stack
- Stdlib: import poplib, email
- Connect: conn = poplib.POP3_SSL("mail.example.com")
- conn.user("alice"); conn.pass_("s3cr3t")
- Stat: count, size = conn.stat()
- List: resp, msgs, total = conn.list()
- Retr: resp, lines, octets = conn.retr(msg_num)
- msg = email.message_from_bytes(b"\n".join(lines))
- UID: resp, uids, _ = conn.uidl()
- Quit: conn.quit() # commits deletespoplib POP3 Email Retrieval Pipeline
# app/poplibutil.py — connect, download, parse, deduplicate, archive
from __future__ import annotations
import email
import email.message
import email.policy
import io
import poplib
import ssl
from dataclasses import dataclass, field
from pathlib import Path
# ─────────────────────────────────────────────────────────────────────────────
# 1. Connection helpers
# ─────────────────────────────────────────────────────────────────────────────
def connect(
host: str,
port: int = 995,
username: str = "",
password: str = "",
use_ssl: bool = True,
timeout: int = 30,
) -> poplib.POP3 | poplib.POP3_SSL:
"""
Open and authenticate a POP3 connection.
Returns the connected POP3 object.
Example:
conn = connect("mail.example.com", username="alice", password="s3cr3t")
count, _ = conn.stat()
print(f"{count} messages")
conn.quit()
"""
if use_ssl:
ctx = ssl.create_default_context()
conn = poplib.POP3_SSL(host, port=port, context=ctx, timeout=timeout)
else:
conn = poplib.POP3(host, port=port, timeout=timeout)
if username:
conn.user(username)
if password:
conn.pass_(password)
return conn
def mailbox_stat(conn: poplib.POP3) -> tuple[int, int]:
"""
Return (message_count, total_bytes) for the mailbox.
Example:
count, size = mailbox_stat(conn)
print(f"{count} messages, {size/1024:.1f} KB total")
"""
return conn.stat()
# ─────────────────────────────────────────────────────────────────────────────
# 2. Message listing and UID tracking
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class MessageEntry:
number: int # 1-based POP3 message number (session-local)
size: int # bytes
uid: str # UIDL unique identifier (persistent across sessions)
def __str__(self) -> str:
return f"#{self.number:4d} uid={self.uid:<30s} {self.size:>8d}B"
def list_messages(conn: poplib.POP3) -> list[MessageEntry]:
"""
Return a list of MessageEntry objects for all messages in the mailbox.
Merges LIST (sizes) and UIDL (uids) responses.
Example:
entries = list_messages(conn)
for e in entries:
print(e)
"""
_, list_lines, _ = conn.list()
_, uid_lines, _ = conn.uidl()
sizes: dict[int, int] = {}
for line in list_lines:
parts = line.decode().split()
if len(parts) >= 2:
sizes[int(parts[0])] = int(parts[1])
uids: dict[int, str] = {}
for line in uid_lines:
parts = line.decode().split()
if len(parts) >= 2:
uids[int(parts[0])] = parts[1]
entries = []
for num in sorted(sizes):
entries.append(MessageEntry(
number=num,
size=sizes[num],
uid=uids.get(num, ""),
))
return entries
# ─────────────────────────────────────────────────────────────────────────────
# 3. Message retrieval and parsing
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class ParsedMessage:
number: int
uid: str
raw: bytes
message: email.message.EmailMessage
@property
def subject(self) -> str:
return self.message.get("Subject", "(no subject)")
@property
def sender(self) -> str:
return self.message.get("From", "")
@property
def date(self) -> str:
return self.message.get("Date", "")
@property
def message_id(self) -> str:
return self.message.get("Message-ID", "")
def body_text(self) -> str:
"""Return the plain-text body, or empty string if none."""
if self.message.is_multipart():
for part in self.message.walk():
if part.get_content_type() == "text/plain":
payload = part.get_payload(decode=True)
charset = part.get_content_charset() or "utf-8"
return payload.decode(charset, errors="replace")
else:
if self.message.get_content_type() == "text/plain":
payload = self.message.get_payload(decode=True)
charset = self.message.get_content_charset() or "utf-8"
return payload.decode(charset, errors="replace")
return ""
def __str__(self) -> str:
return (f"#{self.number} From: {self.sender} "
f"Subject: {self.subject} Date: {self.date}")
def fetch_message(conn: poplib.POP3, number: int, uid: str = "") -> ParsedMessage:
"""
Retrieve and parse a POP3 message by its session number.
Example:
msg = fetch_message(conn, 1)
print(msg.subject, msg.sender)
print(msg.body_text()[:200])
"""
_, lines, _ = conn.retr(number)
raw = b"\r\n".join(lines)
message = email.message_from_bytes(raw, policy=email.policy.default)
return ParsedMessage(number=number, uid=uid, raw=raw, message=message) # type: ignore[arg-type]
def fetch_headers(conn: poplib.POP3, number: int, uid: str = "") -> ParsedMessage:
"""
Retrieve only headers + 0 body lines (fast peek without full download).
Example:
stub = fetch_headers(conn, 1)
print(stub.subject, stub.sender)
"""
_, lines, _ = conn.top(number, 0)
raw = b"\r\n".join(lines)
message = email.message_from_bytes(raw, policy=email.policy.default)
return ParsedMessage(number=number, uid=uid, raw=raw, message=message) # type: ignore[arg-type]
# ─────────────────────────────────────────────────────────────────────────────
# 4. Download session with deduplication
# ─────────────────────────────────────────────────────────────────────────────
def download_new(
conn: poplib.POP3,
seen_uids: set[str],
headers_only: bool = False,
max_messages: int | None = None,
) -> tuple[list[ParsedMessage], set[str]]:
"""
Download messages whose UIDs are not in seen_uids.
Returns (new_messages, updated_seen_uids).
Pass headers_only=True for fast subject/sender preview without body download.
Example:
seen: set[str] = set()
msgs, seen = download_new(conn, seen)
for m in msgs:
print(m)
"""
entries = list_messages(conn)
new_messages = []
new_seen = set(seen_uids)
for entry in entries:
if entry.uid in seen_uids:
continue
if max_messages is not None and len(new_messages) >= max_messages:
break
if headers_only:
msg = fetch_headers(conn, entry.number, entry.uid)
else:
msg = fetch_message(conn, entry.number, entry.uid)
new_messages.append(msg)
new_seen.add(entry.uid)
return new_messages, new_seen
def archive_messages(
conn: poplib.POP3,
archive_dir: str | Path,
seen_uids: set[str] | None = None,
delete_after_save: bool = False,
) -> tuple[list[Path], set[str]]:
"""
Save new messages as .eml files in archive_dir.
Returns (saved_paths, updated_seen_uids).
Example:
paths, seen = archive_messages(conn, "/var/mail/archive")
for p in paths:
print(f" saved {p.name}")
"""
dest = Path(archive_dir)
dest.mkdir(parents=True, exist_ok=True)
seen = set(seen_uids or [])
entries = list_messages(conn)
saved: list[Path] = []
to_delete: list[int] = []
for entry in entries:
if entry.uid in seen:
continue
msg = fetch_message(conn, entry.number, entry.uid)
safe_uid = "".join(c if c.isalnum() or c in "-_." else "_"
for c in entry.uid)
path = dest / f"{safe_uid}.eml"
path.write_bytes(msg.raw)
saved.append(path)
seen.add(entry.uid)
if delete_after_save:
to_delete.append(entry.number)
for num in to_delete:
conn.dele(num)
return saved, seen
# ─────────────────────────────────────────────────────────────────────────────
# Demo (requires a real POP3 server — shows offline simulation)
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== poplib demo (offline simulation) ===")
# Simulate a raw POP3 message
raw_eml = b"""\
From: [email protected]\r
To: [email protected]\r
Subject: Hello from poplib\r
Date: Mon, 11 Nov 2028 10:00:00 +0000\r
Message-ID: <[email protected]>\r
Content-Type: text/plain; charset=utf-8\r
\r
Hi Alice,\r
\r
This is a test message fetched via POP3.\r
"""
# Parse as if fetched from server
import email.policy
message = email.message_from_bytes(raw_eml, policy=email.policy.default)
pm = ParsedMessage(number=1, uid="uid-001", raw=raw_eml, message=message) # type: ignore[arg-type]
print(f"\n--- ParsedMessage ---")
print(f" {pm}")
print(f" body: {pm.body_text()[:60]!r}")
# ── Simulate mailbox stat ─────────────────────────────────────────────────
print("\n--- Simulated deduplication ---")
entries = [
MessageEntry(1, 2048, "uid-001"),
MessageEntry(2, 4096, "uid-002"),
MessageEntry(3, 1024, "uid-003"),
]
seen: set[str] = {"uid-001"} # already downloaded
new = [e for e in entries if e.uid not in seen]
print(f" {len(entries)} messages on server, {len(new)} new")
for e in new:
print(f" would download: {e}")
# ── Archive paths ─────────────────────────────────────────────────────────
print("\n--- Archive naming ---")
import tempfile
with tempfile.TemporaryDirectory() as tmp:
dest = Path(tmp) / "archive"
dest.mkdir()
for e in entries:
safe = "".join(c if c.isalnum() or c in "-_." else "_"
for c in e.uid)
path = dest / f"{safe}.eml"
path.write_bytes(raw_eml)
print(f" saved {path.name} ({e.size}B)")
print("\nTo connect to a real server:")
print(" conn = connect('mail.example.com', username='u', password='p')")
print(" msgs, seen = download_new(conn, seen_uids=set())")
print(" conn.quit()")
print("\n=== done ===")For the imaplib alternative — imaplib.IMAP4_SSL(host) plus conn.select("INBOX"), conn.search(None, "UNSEEN"), and conn.fetch(uid, "(RFC822)") provide full IMAP4 capabilities including server-side search, folder management, flag manipulation, partial message fetching, and persistent connections — use IMAP when you need to leave messages on the server, work with multiple folders, or sync mail across devices; use poplib only for simple download-and-delete workflows or legacy POP3-only servers (many modern email services have removed POP3 support in favor of IMAP). For the smtplib alternative — smtplib.SMTP_SSL(host, 465) with conn.sendmail(sender, recipients, message.as_bytes()) sends email — use smtplib for sending and poplib for receiving; the two modules are complementary and both support SSL/TLS with the standard ssl.create_default_context(). The Claude Skills 360 bundle includes poplib skill sets covering connect() with SSL and auth, MessageEntry with list_messages() mailbox lister, ParsedMessage with fetch_message()/fetch_headers() retrievers and body_text(), and download_new()/archive_messages() session managers with UID deduplication. Start with the free tier to try POP3 email patterns and poplib pipeline code generation.