SSH with paramiko Part 1: Connect & Run

By the end of this lesson you can:

• Connect to a server over SSH with paramiko, using key-based auth.
• Run a remote command and read its stdout, stderr, and exit code.
• Handle host keys safely (and understand why blindly trusting them is risky).
• Use a context manager so connections always close.

You've probably typed ssh user@server and then run commands. paramiko does exactly that programmatically: open a secure connection, send a command, get the output back as strings — so you can manage many servers in a loop, or fold remote steps into a pipeline.

pip install paramiko

Connecting with a key

import paramiko, os

client = paramiko.SSHClient()
client.load_system_host_keys()          # trust hosts you've connected to before
client.set_missing_host_key_policy(paramiko.RejectPolicy())  # refuse unknowns

client.connect(
    hostname=os.environ["SSH_HOST"],
    username=os.environ["SSH_USER"],
    key_filename=os.environ["SSH_KEY"],  # path to your private key
    timeout=15,
)
# …use it…
client.close()

• Key-based auth (key_filename) is the norm — far safer than passwords and the only thing you should automate with.
• Host/user/key paths come from the environment (Lesson 8), never hard-coded.
• Always close() — or use a context manager (below).

Running a command

stdin, stdout, stderr = client.exec_command("df -h /")
exit_code = stdout.channel.recv_exit_status()    # waits for completion
out = stdout.read().decode()
err = stderr.read().decode()

print("exit:", exit_code)
print(out)
if err:
    print("stderr:", err)

exec_command returns three streams. Read the exit status via stdout.channel.recv_exit_status() — it also blocks until the command finishes. Then decode the bytes to strings. It's the remote twin of Lesson 10's capture_output.

Host-key safety

A safe context-manager wrapper

from contextlib import contextmanager
import paramiko, os

@contextmanager
def ssh_connection():
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    try:
        client.connect(hostname=os.environ["SSH_HOST"],
                       username=os.environ["SSH_USER"],
                       key_filename=os.environ["SSH_KEY"], timeout=15)
        yield client
    finally:
        client.close()           # always closes, even on error

def run_remote(client, command: str) -> tuple[int, str, str]:
    _, out, err = client.exec_command(command)
    code = out.channel.recv_exit_status()
    return code, out.read().decode(), err.read().decode()

with ssh_connection() as ssh:
    code, out, err = run_remote(ssh, "uptime")
    print(out.strip())

The context manager guarantees the connection closes; run_remote gives you a clean (code, out, err) tuple per command. This pair is the foundation for everything remote.

Goal: SSH into a server, run several diagnostic commands, parse their output, and return a tidy health dict — the start of a server monitor.

import os, logging
from contextlib import contextmanager
import paramiko

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("ssh")

@contextmanager
def ssh_connection(host: str):
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    try:
        client.connect(hostname=host, username=os.environ["SSH_USER"],
                       key_filename=os.environ["SSH_KEY"], timeout=15)
        yield client
    finally:
        client.close()

def run(client, cmd: str) -> tuple[int, str, str]:
    _, out, err = client.exec_command(cmd, timeout=30)
    return out.channel.recv_exit_status(), out.read().decode(), err.read().decode()

def health(host: str) -> dict:
    info = {"host": host}
    try:
        with ssh_connection(host) as ssh:
            _, uptime, _ = run(ssh, "uptime -p")
            info["uptime"] = uptime.strip()

            code, disk, _ = run(ssh, "df -h / | tail -1 | awk '{print $5}'")
            info["disk_used"] = disk.strip()

            _, mem, _ = run(ssh, "free -m | awk '/Mem:/ {printf \"%d%%\", $3/$2*100}'")
            info["mem_used"] = mem.strip()

            _, who, _ = run(ssh, "who | wc -l")
            info["users"] = who.strip()
        info["reachable"] = True
    except Exception as e:
        log.error("cannot reach %s: %s", host, e)
        info["reachable"] = False
    return info

for h in os.environ.get("SSH_HOSTS", "").split(","):
    if h:
        print(health(h))

{'host': 'web-01', 'uptime': 'up 12 days, 4 hours',
 'disk_used': '63%', 'mem_used': '41%', 'users': '2', 'reachable': True}
ERROR cannot reach db-02: timed out
{'host': 'db-02', 'reachable': False}

Read the code

The context manager ensures each connection closes even if a command fails, and the whole per-host block is wrapped so an unreachable server is recorded as reachable: False rather than crashing the loop — so one dead box doesn't stop you checking the rest. We run standard Unix one-liners remotely and parse their stdout into a dict, exactly like capturing local subprocess output. Loop over a list of hosts and you have a fleet health snapshot — which Lesson 43 turns into a full monitor.

No server? Spin up a free tier VM, use a Raspberry Pi, or run an SSH server locally (you can ssh localhost on most systems after enabling it). Generate a key with ssh-keygen and add the public key to the server's authorized_keys.

for cmd in ["ls /", "ls /nope"]:
    code, out, err = run_remote(ssh, cmd)
    print(cmd, "→ exit", code, "|", (err or out).strip()[:40])

def run_on_all(hosts, command):
    results = {}
    for host in hosts:
        try:
            with ssh_connection(host) as ssh:
                results[host] = run_remote(ssh, command)[:2]
        except Exception as e:
            results[host] = ("unreachable", str(e))
    return results

paramiko SSHes into a server from Python: connect with key-based auth (paths from the environment), run a command with exec_command, and read its stdout/stderr plus the exit status via recv_exit_status() — the remote equivalent of Lesson 10's subprocess capture. Wrap connections in a context manager so they always close, and handle host keys responsibly: pre-load known hosts and reject unknowns rather than blindly AutoAddPolicy. Loop over hosts to manage a fleet. Next: moving files over the same connection with SFTP.