System administrators and DevOps engineers frequently need to run ad-hoc commands across multiple servers—such as checking disk usage, verifying service st
System administrators and DevOps engineers frequently need to run ad-hoc commands across multiple servers—such as checking disk usage, verifying service status, or restarting a daemon across an entire cluster. Running these commands manually via separate SSH sessions is tedious, slow, and prone to human error.
While orchestration platforms like Ansible or SaltStack exist, setting them up for quick, one-off operational tasks or embedding them into lightweight CI/CD pipelines can be unnecessary overhead. This script provides a lightweight, self-contained Python solution using standard SSH protocols. It allows you to run shell commands concurrently across multiple hosts, handle host-specific connection details, capture exit status and output, and gracefully process connection failures.
The script reads target inventory details from a specified text file containing hostnames or IP addresses (with optional user and port syntax like user@hostname:port). It parses command-line arguments using Python's argparse module, falling back to configurable environment variables for defaults such as key file paths, usernames, timeouts, and concurrency limits.
Execution relies on the paramiko library for SSH transport. When the --parallel flag is specified, the runner dispatches execution tasks to a concurrent.futures.ThreadPoolExecutor, enabling simultaneous SSH connections and command execution across the server list. Each thread opens an SSH channel, runs the requested shell command, reads stdout and stderr streams, and captures the remote process exit code. Results are collected, formatted, and summarized in real time or upon completion.
#!/usr/bin/env python3
"""
Multi-Server SSH Command Runner
Executes arbitrary commands concurrently or sequentially across multiple servers using Paramiko.
"""
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
import getpass
import os
from pathlib import Path
import sys
from typing import Optional
import paramiko
# Default configurations read from environment variables
DEFAULT_USER = os.environ.get("SSH_RUNNER_USER", getpass.getuser())
DEFAULT_KEY_PATH = os.environ.get(
"SSH_RUNNER_KEY_PATH", str(Path.home() / ".ssh" / "id_rsa")
)
DEFAULT_TIMEOUT = int(os.environ.get("SSH_RUNNER_TIMEOUT", "10"))
DEFAULT_MAX_WORKERS = int(os.environ.get("SSH_RUNNER_MAX_WORKERS", "10"))
@dataclass
class HostTarget:
"""Represents an SSH target host configuration."""
hostname: str
username: str
port: int = 22
@dataclass
class CommandResult:
"""Encapsulates the result of a remote command execution."""
target: HostTarget
exit_code: int
stdout: str
stderr: str
error: Optional[str] = None
def parse_host_line(line: str, default_user: str, default_port: int = 22) -> HostTarget | None:
"""Parse a host string into a HostTarget instance.
Supported formats:
- hostname
- user@hostname
- user@hostname:port
- hostname:port
"""
cleaned = line.strip()
if not cleaned or cleaned.startswith("#"):
return None
username = default_user
port = default_port
if "@" in cleaned:
username, cleaned = cleaned.split("@", 1)
if ":" in cleaned:
hostname, port_str = cleaned.split(":", 1)
try:
port = int(port_str)
except ValueError:
port = default_port
else:
hostname = cleaned
return HostTarget(hostname=hostname.strip(), username=username.strip(), port=port)
def load_hosts(hosts_file: Path, default_user: str) -> list[HostTarget]:
"""Load host definitions from a text file."""
if not hosts_file.is_file():
print(f"Error: Hosts file '{hosts_file}' not found.", file=sys.stderr)
sys.exit(1)
targets: list[HostTarget] = []
with open(hosts_file, "r", encoding="utf-8") as file:
for line_num, line in enumerate(file, 1):
target = parse_host_line(line, default_user)
if target:
targets.append(target)
if not targets:
print(f"Error: No valid targets found in '{hosts_file}'.", file=sys.stderr)
sys.exit(1)
return targets
def execute_remote_command(
target: HostTarget,
command: str,
key_filename: str | None,
timeout: int,
) -> CommandResult:
"""Connect to a single remote host and execute a command using Paramiko."""
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
expanded_key_path = os.path.expanduser(key_filename) if key_filename else None
try:
# Establish connection
client.connect(
hostname=target.hostname,
port=target.port,
username=target.username,
key_filename=expanded_key_path,
timeout=timeout,
banner_timeout=timeout,
auth_timeout=timeout,
)
# Execute remote command
_, stdout_stream, stderr_stream = client.exec_command(command, timeout=timeout)
# Retrieve outputs and exit status
stdout = stdout_stream.read().decode("utf-8", errors="replace").strip()
stderr = stderr_stream.read().decode("utf-8", errors="replace").strip()
exit_code = stdout_stream.channel.recv_exit_status()
return CommandResult(
target=target,
exit_code=exit_code,
stdout=stdout,
stderr=stderr,
)
except paramiko.AuthenticationException:
return CommandResult(
target=target,
exit_code=-1,
stdout="",
stderr="",
error="Authentication failed.",
)
except paramiko.SSHException as e:
return CommandResult(
target=target,
exit_code=-1,
stdout="",
stderr="",
error=f"SSH protocol error: {e}",
)
except Exception as e:
return CommandResult(
target=target,
exit_code=-1,
stdout="",
stderr="",
error=f"Connection error: {e}",
)
finally:
client.close()
def print_result(result: CommandResult) -> None:
"""Format and print execution results to stdout."""
target_str = f"{result.target.username}@{result.target.hostname}:{result.target.port}"
print("=" * 60)
if result.error:
print(f"❌ [FAILED] {target_str}")
print(f" Error: {result.error}")
elif result.exit_code != 0:
print(f"⚠️ [NON-ZERO EXIT: {result.exit_code}] {target_str}")
if result.stdout:
print(f" STDOUT:\n{result.stdout}")
if result.stderr:
print(f" STDERR:\n{result.stderr}")
else:
print(f"✅ [SUCCESS] {target_str}")
if result.stdout:
print(f" STDOUT:\n{result.stdout}")
if result.stderr:
print(f" STDERR:\n{result.stderr}")
def run_all(
targets: list[HostTarget],
command: str,
key_filename: str | None,
timeout: int,
parallel: bool,
max_workers: int,
) -> list[CommandResult]:
"""Coordinate sequential or parallel command execution across targets."""
results: list[CommandResult] = []
if parallel:
workers = min(len(targets), max_workers)
with ThreadPoolExecutor(max_workers=workers) as executor:
future_to_target = {
executor.submit(
execute_remote_command, target, command, key_filename, timeout
): target
for target in targets
}
for future in as_completed(future_to_target):
res = future.result()
results.append(res)
print_result(res)
else:
for target in targets:
res = execute_remote_command(target, command, key_filename, timeout)
results.append(res)
print_result(res)
return results
def main() -> None:
parser = argparse.ArgumentParser(
description="Run SSH commands across multiple servers concurrently or sequentially."
)
parser.add_argument(
"--hosts",
"-f",
required=True,
type=Path,
help="Path to hosts file containing IP/hostnames",
)
parser.add_argument(
"--cmd",
"-c",
required=True,
type=str,
help="Command string to execute remotely",
)
parser.add_argument(
"--parallel",
"-p",
action="store_true",
help="Run commands concurrently using threads",
)
parser.add_argument(
"--user",
"-u",
default=DEFAULT_USER,
help=f"Default SSH user (default: {DEFAULT_USER})",
)
parser.add_argument(
"--key-filename",
"-i",
default=DEFAULT_KEY_PATH,
help=f"Path to private SSH key (default: {DEFAULT_KEY_PATH})",
)
parser.add_argument(
"--timeout",
"-t",
type=int,
default=DEFAULT_TIMEOUT,
help=f"SSH connection timeout in seconds (default: {DEFAULT_TIMEOUT})",
)
parser.add_argument(
"--max-workers",
"-w",
type=int,
default=DEFAULT_MAX_WORKERS,
help=f"Maximum thread pool workers for parallel execution (default: {DEFAULT_MAX_WORKERS})",
)
args = parser.parse_args()
targets = load_hosts(args.hosts, args.user)
print(f"Loaded {len(targets)} host(s). Starting execution...")
results = run_all(
targets=targets,
command=args.cmd,
key_filename=args.key_filename,
timeout=args.timeout,
parallel=args.parallel,
max_workers=args.max_workers,
)
print("=" * 60)
failed_count = sum(1 for r in results if r.error or r.exit_code != 0)
success_count = len(results) - failed_count
print(f"Execution complete: {success_count} succeeded, {failed_count} failed.")
if failed_count > 0:
sys.exit(1)
if __name__ == "__main__":
main()
The script accepts CLI arguments, but default behaviors can be customized via standard environment variables:
| Environment Variable | CLI Argument | Default Value | Description |
|---|---|---|---|
SSH_RUNNER_USER |
--user, -u |
Current system user | Fallback SSH username when not specified in hosts.txt. |
SSH_RUNNER_KEY_PATH |
--key-filename, -i |
~/.ssh/id_rsa |
Default path to the SSH private key. |
SSH_RUNNER_TIMEOUT |
--timeout, -t |
10 |
Timeout in seconds for connection and execution. |
SSH_RUNNER_MAX_WORKERS |
--max-workers, -w |
10 |
Maximum parallel execution threads. |
hosts.txt)Create a host inventory file. Lines starting with # and empty lines are skipped.
# General server inventory
10.0.0.10
ubuntu@10.0.0.11
admin@10.0.0.12:2222
webserver.example.com
Runs a df -h / command host-by-host using the default key file and user.
python3 ssh_runner.py --hosts hosts.txt --cmd "df -h /"
Output:
Loaded 2 host(s). Starting execution...
============================================================
✅ [SUCCESS] admin@10.0.0.10:22
STDOUT:
Filesystem Size Used Avail Use% Mounted on
/dev/sda1 50G 12G 38G 25% /
============================================================
✅ [SUCCESS] ubuntu@10.0.0.11:22
STDOUT:
Filesystem Size Used Avail Use% Mounted on
/dev/sda1 50G 20G 30G 40% /
============================================================
Execution complete: 2 succeeded, 0 failed.
Restarting Nginx across nodes concurrently using a specific Ed25519 identity key.
python3 ssh_runner.py \
--hosts hosts.txt \
--cmd "sudo systemctl restart nginx" \
--parallel \
--key-filename ~/.ssh/id_ed25519
Output:
Loaded 3 host(s). Starting execution...
============================================================
✅ [SUCCESS] ubuntu@10.0.0.11:22
STDOUT:
============================================================
❌ [FAILED] admin@10.0.0.12:2222
Error: Authentication failed.
============================================================
⚠️ [NON-ZERO EXIT: 1] admin@10.0.0.10:22
STDERR:
Job for nginx.service failed because the control process exited with error code.
============================================================
Execution complete: 1 succeeded, 2 failed.
This script handles common SSH operational failures explicitly:
paramiko.AuthenticationException): Caught per host without crashing the batch execution. The output flags the failure clearly (Authentication failed.).paramiko.SSHException, socket.error): Handled in the general exception block. Outages on single targets do not block command runs on other hosts.⚠️ [NON-ZERO EXIT: X]), preserving stdout and stderr output.--json CLI option to format execution results into structural JSON data suitable for consumption by monitoring pipelines or dashboard backends.execute_remote_command to support SSH password authentication or handle sudo password prompts using Paramiko's interactively handled