Python’s ftplib module implements the FTP client protocol for uploading and downloading files. import ftplib. FTP: ftp = ftplib.FTP("ftp.example.com") → ftp.login("user", "pass"). FTP_TLS: ftp = ftplib.FTP_TLS("ftp.example.com"); ftp.login("user", "pass"); ftp.prot_p() — enables data channel encryption. Passive mode (default): ftp.set_pasv(True). List: ftp.nlst() → filename list; ftp.mlsd() → iterator of (name, facts) with size, type, modify; ftp.dir() → prints directory listing. Download binary: ftp.retrbinary("RETR file.bin", callback) — callback receives bytes chunks. Download text: ftp.retrlines("RETR file.txt", callback) — callback receives decoded lines. Upload binary: ftp.storbinary("STOR dest.bin", fp) — reads from file-like object. Upload text: ftp.storlines("STOR dest.txt", fp). Size: ftp.size("file.bin") — file size in bytes (server must support SIZE). Directory ops: ftp.cwd("subdir"), ftp.pwd(), ftp.mkd("newdir"), ftp.rmd("emptydir"). File ops: ftp.delete("old.txt"), ftp.rename("old", "new"). Context manager: with ftplib.FTP("host") as ftp: ftp.login(...). quit/close: ftp.quit() — sends QUIT; ftp.close() — silent close. Timeout: ftplib.FTP(timeout=30). Claude Code generates FTP directory mirrors, incremental sync tools, automated report uploaders, and FTP health monitors.
CLAUDE.md for ftplib
## ftplib Stack
- Stdlib: import ftplib, ssl, io
- Plain: with ftplib.FTP("host", timeout=30) as ftp: ftp.login(user, pass)
- TLS: ftp = ftplib.FTP_TLS("host"); ftp.login(u, p); ftp.prot_p()
- List: [(name, facts) for name, facts in ftp.mlsd()]
- Get: buf = io.BytesIO(); ftp.retrbinary("RETR file", buf.write); buf.seek(0)
- Put: ftp.storbinary("STOR dest", open("local", "rb"))ftplib FTP Pipeline
# app/ftputil.py — download, upload, sync, directory ops, health check
from __future__ import annotations
import ftplib
import hashlib
import io
import ssl
import time
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Generator, Iterator
# ─────────────────────────────────────────────────────────────────────────────
# 1. Connection helpers
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class FTPConfig:
host: str
port: int = 21
username: str = "anonymous"
password: str = ""
tls: bool = False
passive: bool = True
timeout: float = 30.0
encoding: str = "utf-8"
def _open_ftp(cfg: FTPConfig) -> ftplib.FTP:
"""Open and authenticate an FTP(S) connection."""
if cfg.tls:
ctx = ssl.create_default_context()
ftp: ftplib.FTP = ftplib.FTP_TLS(
host=cfg.host, user=cfg.username, passwd=cfg.password,
context=ctx, timeout=cfg.timeout, encoding=cfg.encoding,
)
ftp.prot_p() # encrypt data channel
else:
ftp = ftplib.FTP(
host=cfg.host, user=cfg.username, passwd=cfg.password,
timeout=cfg.timeout, encoding=cfg.encoding,
)
ftp.set_pasv(cfg.passive)
return ftp
@contextmanager
def ftp_session(cfg: FTPConfig) -> Generator[ftplib.FTP, None, None]:
"""
Context manager that opens, yields, and cleanly quits an FTP connection.
Example:
with ftp_session(cfg) as ftp:
files = list_files(ftp)
"""
ftp = _open_ftp(cfg)
try:
yield ftp
finally:
try:
ftp.quit()
except Exception:
ftp.close()
# ─────────────────────────────────────────────────────────────────────────────
# 2. Directory listing
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class FTPEntry:
name: str
is_dir: bool
size: int # 0 if directory or unknown
modified: str # modify timestamp string (YYYYMMDDHHMMSS)
def __str__(self) -> str:
kind = "DIR" if self.is_dir else "FILE"
return f"{kind:4s} {self.name:40s} {self.size:12,d} {self.modified}"
def list_files(ftp: ftplib.FTP, path: str = ".") -> list[FTPEntry]:
"""
List a remote directory using MLSD; returns FTPEntry list.
Falls back to NLST if server doesn't support MLSD.
Example:
entries = list_files(ftp)
for e in entries:
print(e)
"""
try:
entries = []
for name, facts in ftp.mlsd(path):
if name in (".", ".."):
continue
is_dir = facts.get("type", "").lower() in ("dir", "cdir", "pdir")
size = int(facts.get("size", 0))
modified = facts.get("modify", "")
entries.append(FTPEntry(name=name, is_dir=is_dir, size=size, modified=modified))
return entries
except ftplib.error_perm:
# Fallback to NLST
names = ftp.nlst(path)
return [FTPEntry(name=n, is_dir=False, size=0, modified="") for n in names]
def find_files(
ftp: ftplib.FTP,
directory: str = ".",
pattern: str = "*",
recursive: bool = False,
) -> list[str]:
"""
Return remote file paths matching a simple fnmatch pattern.
Example:
pdfs = find_files(ftp, "/reports", "*.pdf")
"""
import fnmatch
results: list[str] = []
_find_recursive(ftp, directory, pattern, recursive, results)
return results
def _find_recursive(ftp: ftplib.FTP, directory: str, pattern: str,
recursive: bool, results: list[str]) -> None:
import fnmatch
for entry in list_files(ftp, directory):
path = f"{directory.rstrip('/')}/{entry.name}"
if entry.is_dir:
if recursive:
_find_recursive(ftp, path, pattern, recursive, results)
else:
if fnmatch.fnmatch(entry.name, pattern):
results.append(path)
# ─────────────────────────────────────────────────────────────────────────────
# 3. Download helpers
# ─────────────────────────────────────────────────────────────────────────────
def download_bytes(ftp: ftplib.FTP, remote_path: str) -> bytes:
"""
Download a file and return its content as bytes.
Example:
data = download_bytes(ftp, "/data/report.csv")
"""
buf = io.BytesIO()
ftp.retrbinary(f"RETR {remote_path}", buf.write)
return buf.getvalue()
def download_file(
ftp: ftplib.FTP,
remote_path: str,
local_path: str | Path,
*,
on_progress: Callable[[int, int], None] | None = None,
) -> Path:
"""
Download remote_path to a local file.
on_progress called with (bytes_done, total_size).
Example:
download_file(ftp, "/backups/db.sql.gz", "/tmp/db.sql.gz")
"""
dest = Path(local_path)
dest.parent.mkdir(parents=True, exist_ok=True)
try:
total = ftp.size(remote_path) or 0
except Exception:
total = 0
done = 0
with dest.open("wb") as fout:
def _write(chunk: bytes) -> None:
nonlocal done
fout.write(chunk)
done += len(chunk)
if on_progress:
on_progress(done, total)
ftp.retrbinary(f"RETR {remote_path}", _write)
return dest
# ─────────────────────────────────────────────────────────────────────────────
# 4. Upload helpers
# ─────────────────────────────────────────────────────────────────────────────
def upload_bytes(ftp: ftplib.FTP, data: bytes, remote_path: str) -> None:
"""
Upload bytes to remote_path.
Example:
upload_bytes(ftp, csv_bytes, "/exports/data.csv")
"""
ftp.storbinary(f"STOR {remote_path}", io.BytesIO(data))
def upload_file(ftp: ftplib.FTP, local_path: str | Path, remote_path: str) -> None:
"""
Upload a local file to remote_path.
Example:
upload_file(ftp, "/tmp/report.pdf", "/incoming/report.pdf")
"""
with Path(local_path).open("rb") as fin:
ftp.storbinary(f"STOR {remote_path}", fin)
def ensure_remote_dir(ftp: ftplib.FTP, remote_dir: str) -> None:
"""
Create remote_dir (and all parents) if it doesn't exist.
Example:
ensure_remote_dir(ftp, "/archive/2025/q2")
"""
parts = [p for p in remote_dir.split("/") if p]
path = ""
for part in parts:
path = f"{path}/{part}"
try:
ftp.cwd(path)
except ftplib.error_perm:
ftp.mkd(path)
ftp.cwd("/")
# ─────────────────────────────────────────────────────────────────────────────
# 5. Sync and utility helpers
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class SyncResult:
uploaded: list[str] = field(default_factory=list)
skipped: list[str] = field(default_factory=list)
failed: list[tuple[str, str]] = field(default_factory=list) # (path, error)
def __str__(self) -> str:
return (f"uploaded={len(self.uploaded)} "
f"skipped={len(self.skipped)} "
f"failed={len(self.failed)}")
def sync_directory_to_ftp(
ftp: ftplib.FTP,
local_dir: str | Path,
remote_dir: str,
*,
pattern: str = "*",
overwrite: bool = False,
) -> SyncResult:
"""
Upload files from local_dir to remote_dir; skip if remote already exists
and overwrite=False.
Example:
result = sync_directory_to_ftp(ftp, "/tmp/reports", "/incoming/reports")
print(result)
"""
import fnmatch
local = Path(local_dir)
result = SyncResult()
ensure_remote_dir(ftp, remote_dir)
remote_names = {e.name for e in list_files(ftp, remote_dir) if not e.is_dir}
for fpath in local.iterdir():
if not fpath.is_file():
continue
if not fnmatch.fnmatch(fpath.name, pattern):
continue
remote_path = f"{remote_dir.rstrip('/')}/{fpath.name}"
if fpath.name in remote_names and not overwrite:
result.skipped.append(remote_path)
continue
try:
upload_file(ftp, fpath, remote_path)
result.uploaded.append(remote_path)
except Exception as exc:
result.failed.append((remote_path, str(exc)))
return result
def ftp_health_check(cfg: FTPConfig) -> tuple[bool, str]:
"""
Test FTP connectivity and auth; return (ok, message).
Example:
ok, msg = ftp_health_check(cfg)
"""
try:
with ftp_session(cfg) as ftp:
pwd = ftp.pwd()
return True, f"{cfg.host}:{cfg.port} OK (cwd={pwd})"
except ftplib.error_perm as e:
return False, f"auth error: {e}"
except Exception as e:
return False, f"connection error: {e}"
# ─────────────────────────────────────────────────────────────────────────────
# Demo (uses anonymous FTP — skip gracefully if offline)
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
import tempfile
print("=== ftplib demo ===")
# ── Show FTPEntry repr ────────────────────────────────────────────────────
print("\n--- FTPEntry ---")
entries = [
FTPEntry("readme.txt", False, 4096, "20250101120000"),
FTPEntry("archive", True, 0, "20250201000000"),
FTPEntry("report.csv", False, 2097152, "20250301093000"),
]
for e in entries:
print(f" {e}")
# ── Anonymous FTP to public server ────────────────────────────────────────
print("\n--- anonymous FTP (ftp.gnu.org) ---")
gnu_cfg = FTPConfig(host="ftp.gnu.org", username="anonymous", password="guest@",
timeout=10.0)
try:
with ftp_session(gnu_cfg) as ftp:
top = list_files(ftp, "/gnu")[:5]
print(f" /gnu listing (first 5):")
for e in top:
print(f" {e}")
except Exception as e:
print(f" network error: {e} (offline?)")
# ── Download to bytes ─────────────────────────────────────────────────────
print("\n--- download_bytes ---")
try:
with ftp_session(gnu_cfg) as ftp:
data = download_bytes(ftp, "/gnu/README")
print(f" /gnu/README: {len(data)} bytes")
print(f" first line: {data.splitlines()[0].decode(errors='replace')!r}")
except Exception as e:
print(f" network error: {e} (offline?)")
# ── SyncResult shape ─────────────────────────────────────────────────────
print("\n--- SyncResult ---")
result = SyncResult(
uploaded=["/remote/a.csv", "/remote/b.csv"],
skipped=["/remote/c.csv"],
failed=[("/remote/d.csv", "Permission denied")],
)
print(f" {result}")
print("\n=== done ===")For the paramiko / pysftp alternative — paramiko (PyPI) implements SSH2 and SFTP, which transfers files over an encrypted SSH channel; SFTP is the modern successor to FTP for secure transfers on port 22 — use paramiko or pysftp for any new deployment where the server supports SSH; use ftplib only when connecting to legacy servers that offer only plain FTP or when FTPS (FTP over TLS) is the explicit server requirement. For the ftputil alternative — ftputil (PyPI) wraps ftplib with an os.path-like API where you can call ftputil.FTPHost.path.exists(), ftputil.FTPHost.walk(), ftputil.FTPHost.open() in the same style as local filesystem code; it caches directory listings and handles recursive directory operations cleanly — use ftputil when you need to write FTP-access code that structurally mirrors os/pathlib patterns; use ftplib directly for simple upload/download scripts or when you want a thin layer with no PyPI dependencies. The Claude Skills 360 bundle includes ftplib skill sets covering FTPConfig dataclass, ftp_session() context manager, FTPEntry dataclass with list_files()/find_files() directory helpers, download_bytes()/download_file() with progress callback, upload_bytes()/upload_file()/ensure_remote_dir() upload helpers, SyncResult with sync_directory_to_ftp(), and ftp_health_check(). Start with the free tier to try FTP transfer patterns and ftplib pipeline code generation.