Paramiko implements SSH2 protocol in Python. pip install paramiko. Connect: import paramiko; client = paramiko.SSHClient(). client.set_missing_host_key_policy(paramiko.AutoAddPolicy()). client.connect("host", username="user", key_filename="/home/user/.ssh/id_rsa"). Password: client.connect("host", username="user", password="secret"). Command: stdin, stdout, stderr = client.exec_command("ls -la"). output = stdout.read().decode(). exit_code = stdout.channel.recv_exit_status(). Timeout: client.exec_command("cmd", timeout=30). SFTP: sftp = client.open_sftp(). sftp.get("/remote/file.txt", "/local/file.txt"). sftp.put("/local/file.txt", "/remote/file.txt"). sftp.listdir("/remote/dir"). sftp.listdir_attr("/remote/dir") → SFTPAttributes. sftp.mkdir("/remote/newdir"). sftp.stat("/remote/file"). sftp.remove("/remote/file"). sftp.rename("/remote/old", "/remote/new"). Close: sftp.close(); client.close(). Context: with paramiko.SSHClient() as client: .... Transport: transport = paramiko.Transport(("host", 22)); transport.connect(username="u", pkey=key). sftp = paramiko.SFTPClient.from_transport(transport). Key: key = paramiko.RSAKey.from_private_key_file("~/.ssh/id_rsa"). key = paramiko.Ed25519Key.from_private_key_file("~/.ssh/id_ed25519"). Jump host: set sock to a channel opened through the first host. SSHConfig: reads ~/.ssh/config. chan = client.get_transport().open_channel("direct-tcpip", dest, src). Claude Code generates Paramiko SSH connections, SFTP transfers, and bastion host tunneling.
CLAUDE.md for Paramiko
## Paramiko Stack
- Version: paramiko >= 3.4 | pip install paramiko
- Connect: client.connect(host, username=u, key_filename=path) | password=p
- Command: stdin, stdout, stderr = client.exec_command(cmd); stdout.read()
- Exit code: stdout.channel.recv_exit_status() — wait for command to finish
- SFTP: sftp = client.open_sftp(); sftp.get(remote, local); sftp.put(local, remote)
- Key: RSAKey.from_private_key_file(path) | Ed25519Key.from_private_key_file(path)
- Jump: open_channel("direct-tcpip", dest, src) through intermediary SSH connectionParamiko SSH and SFTP Pipeline
# app/ssh_client.py — Paramiko SSH, SFTP, and bastion host patterns
from __future__ import annotations
import io
import logging
import os
import socket
import stat
import time
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Generator, Iterator
import paramiko
log = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# Connection config
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class SSHConfig:
host: str
port: int = 22
username: str = "ubuntu"
key_path: str | None = None
password: str | None = None
timeout: float = 30.0
banner_timeout: float = 60.0
known_hosts: str | None = None # None → AutoAddPolicy (dev); path → RejectPolicy (prod)
# ─────────────────────────────────────────────────────────────────────────────
# 1. SSH client factory
# ─────────────────────────────────────────────────────────────────────────────
def make_ssh_client(cfg: SSHConfig) -> paramiko.SSHClient:
"""
Build and connect an SSHClient.
AutoAddPolicy is convenient for dev — in production, load known_hosts.
"""
client = paramiko.SSHClient()
if cfg.known_hosts:
client.load_host_keys(cfg.known_hosts)
client.set_missing_host_key_policy(paramiko.RejectPolicy())
else:
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
connect_kwargs: dict = {
"hostname": cfg.host,
"port": cfg.port,
"username": cfg.username,
"timeout": cfg.timeout,
"banner_timeout": cfg.banner_timeout,
"look_for_keys": cfg.key_path is None and cfg.password is None,
"allow_agent": True,
}
if cfg.key_path:
connect_kwargs["key_filename"] = cfg.key_path
elif cfg.password:
connect_kwargs["password"] = cfg.password
client.connect(**connect_kwargs)
log.info("ssh_connected", extra={"host": cfg.host, "user": cfg.username})
return client
@contextmanager
def ssh_connection(cfg: SSHConfig) -> Generator[paramiko.SSHClient, None, None]:
"""Context manager: auto-closes the client on exit."""
client = make_ssh_client(cfg)
try:
yield client
finally:
client.close()
log.info("ssh_closed", extra={"host": cfg.host})
# ─────────────────────────────────────────────────────────────────────────────
# 2. Command execution
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class CommandResult:
command: str
stdout: str
stderr: str
exit_code: int
@property
def ok(self) -> bool:
return self.exit_code == 0
def run_command(
client: paramiko.SSHClient,
command: str,
timeout: float = 60.0,
env: dict[str, str] | None = None,
) -> CommandResult:
"""
exec_command runs a single command in a non-interactive shell.
recv_exit_status() blocks until the command finishes.
Always read stdout/stderr before calling recv_exit_status to avoid deadlock.
"""
log.debug("run_command", extra={"command": command})
stdin, stdout, stderr = client.exec_command(command, timeout=timeout, environment=env)
stdin.close() # we don't send stdin input
out = stdout.read().decode("utf-8", errors="replace")
err = stderr.read().decode("utf-8", errors="replace")
code = stdout.channel.recv_exit_status()
result = CommandResult(command=command, stdout=out, stderr=err, exit_code=code)
if not result.ok:
log.warning("command_failed", extra={"command": command, "exit_code": code, "stderr": err})
return result
def run_commands(
client: paramiko.SSHClient,
commands: list[str],
stop_on_error: bool = True,
) -> list[CommandResult]:
"""Run a sequence of commands, optionally stopping on first failure."""
results: list[CommandResult] = []
for cmd in commands:
result = run_command(client, cmd)
results.append(result)
if stop_on_error and not result.ok:
raise RuntimeError(
f"Command failed (exit {result.exit_code}): {cmd}\n{result.stderr}"
)
return results
# ─────────────────────────────────────────────────────────────────────────────
# 3. SFTP transfers
# ─────────────────────────────────────────────────────────────────────────────
@contextmanager
def sftp_connection(client: paramiko.SSHClient) -> Generator[paramiko.SFTPClient, None, None]:
"""Open an SFTP session from an existing SSH client."""
sftp = client.open_sftp()
try:
yield sftp
finally:
sftp.close()
def upload_file(
sftp: paramiko.SFTPClient,
local_path: Path,
remote_path: str,
callback: bool = True,
) -> None:
"""Upload a local file to a remote path."""
size = local_path.stat().st_size
def _progress(transferred: int, total: int) -> None:
pct = transferred / total * 100 if total else 0
log.debug("upload_progress", extra={"pct": f"{pct:.0f}%", "file": local_path.name})
sftp.put(str(local_path), remote_path, callback=_progress if callback else None)
log.info("file_uploaded", extra={"local": str(local_path), "remote": remote_path, "size": size})
def download_file(
sftp: paramiko.SFTPClient,
remote_path: str,
local_path: Path,
) -> None:
"""Download a remote file to a local path."""
local_path.parent.mkdir(parents=True, exist_ok=True)
sftp.get(remote_path, str(local_path))
log.info("file_downloaded", extra={"remote": remote_path, "local": str(local_path)})
def list_remote_dir(
sftp: paramiko.SFTPClient,
remote_path: str,
) -> list[dict]:
"""List a remote directory with metadata."""
entries = []
for attr in sftp.listdir_attr(remote_path):
entries.append({
"name": attr.filename,
"size": attr.st_size,
"modified": attr.st_mtime,
"is_dir": stat.S_ISDIR(attr.st_mode or 0),
})
return sorted(entries, key=lambda e: e["name"])
def sync_directory(
sftp: paramiko.SFTPClient,
local_dir: Path,
remote_dir: str,
pattern: str = "*",
) -> list[str]:
"""Upload all files matching pattern from local_dir to remote_dir."""
uploaded: list[str] = []
for local_file in local_dir.glob(pattern):
if local_file.is_file():
remote_path = f"{remote_dir}/{local_file.name}"
sftp.put(str(local_file), remote_path)
uploaded.append(remote_path)
return uploaded
# ─────────────────────────────────────────────────────────────────────────────
# 4. Jump / bastion host tunneling
# ─────────────────────────────────────────────────────────────────────────────
def connect_via_jump(
bastion: SSHConfig,
target: SSHConfig,
) -> paramiko.SSHClient:
"""
Open a TCP tunnel through the bastion host, then connect the target over it.
Equivalent to: ssh -J bastion target
"""
# Step 1: connect to bastion
bastion_client = make_ssh_client(bastion)
# Step 2: open a "direct-tcpip" channel through bastion to target
transport = bastion_client.get_transport()
channel = transport.open_channel(
"direct-tcpip",
dest_addr=(target.host, target.port),
src_addr=("localhost", 0),
)
# Step 3: connect target SSH over the bastion channel
target_transport = paramiko.Transport(channel)
target_transport.start_client()
if target.key_path:
key = _load_key(target.key_path)
target_transport.auth_publickey(target.username, key)
elif target.password:
target_transport.auth_password(target.username, target.password)
else:
agent = paramiko.Agent()
keys = agent.get_keys()
for k in keys:
try:
target_transport.auth_publickey(target.username, k)
break
except paramiko.AuthenticationException:
continue
target_client = paramiko.SSHClient()
target_client._transport = target_transport
log.info("jump_connected", extra={"bastion": bastion.host, "target": target.host})
return target_client, bastion_client # caller must close both
def _load_key(path: str) -> paramiko.PKey:
"""Load RSA or Ed25519 private key from path."""
expanded = os.path.expanduser(path)
for cls in (paramiko.Ed25519Key, paramiko.RSAKey, paramiko.ECDSAKey):
try:
return cls.from_private_key_file(expanded)
except (paramiko.SSHException, ValueError):
continue
raise ValueError(f"Could not load key from {path}")
# ─────────────────────────────────────────────────────────────────────────────
# 5. Deployment helper
# ─────────────────────────────────────────────────────────────────────────────
def deploy_application(
cfg: SSHConfig,
local_build: Path,
remote_dir: str,
service: str,
) -> dict:
"""
Full deployment: upload build artefact, extract, restart service.
"""
archive_name = local_build.name
remote_archive = f"/tmp/{archive_name}"
with ssh_connection(cfg) as client:
# Upload build artefact
with sftp_connection(client) as sftp:
upload_file(sftp, local_build, remote_archive)
# Extract and restart
cmds = [
f"tar -xzf {remote_archive} -C {remote_dir}",
f"rm {remote_archive}",
f"sudo systemctl restart {service}",
f"sudo systemctl is-active {service}",
]
results = run_commands(client, cmds)
status = results[-1].stdout.strip()
return {
"deployed": archive_name,
"remote": remote_dir,
"service": service,
"status": status,
}
# ─────────────────────────────────────────────────────────────────────────────
# Demo (dry-run — no real server required)
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("Paramiko patterns loaded.")
print(f" paramiko version: {paramiko.__version__}")
# Show key loading approach
default_key = Path.home() / ".ssh" / "id_ed25519"
if default_key.exists():
key = _load_key(str(default_key))
print(f" Loaded key: {key.get_name()}")
else:
print(" No SSH key found at ~/.ssh/id_ed25519 — skipping key demo")For the subprocess.run(["ssh", ...]) alternative — shelling out to the ssh binary requires the binary to be installed and on PATH, serializes output as bytes with no structured access to stdout/stderr separately, does not support programmatic key loading or agent forwarding in a cross-platform way, and provides no SFTP capability, while Paramiko implements SSH2 in pure Python so the same code runs on Windows, macOS, and Linux without the openssh-client package, exec_command() returns separate stdout/stderr channels with .recv_exit_status() for reliable exit code capture, and open_sftp() gives a full get/put/listdir API. For the fabric alternative — Fabric builds on top of Paramiko and adds higher-level abstractions like run(), put(), cd() context managers, and connection groups for multi-host operations, while raw Paramiko gives you full control over the SSH transport for cases like bastion host tunneling, interactive shell sessions, port forwarding, and custom key agent routing that Fabric does not expose. The Claude Skills 360 bundle includes Paramiko skill sets covering SSHClient connect with key and password auth, AutoAddPolicy vs known_hosts for host verification, exec_command with recv_exit_status deadlock avoidance, SFTPClient get/put with progress callback, listdir_attr for typed directory listings, sync_directory for batch uploads, bastion host tunneling via open_channel direct-tcpip, Ed25519Key and RSAKey loading, multi-command execution with stop_on_error, and deployment automation patterns. Start with the free tier to try SSH automation code generation.