Python 用 paramiko 操作远程服务器:SSH 登录、传文件、批量执行,看这一篇就够了

写给谁看:运维兄弟、刚学 Python 想做自动化的同学、被重复手工操作折磨的打工人
需要什么基础:会装 Python、会 pip 安装包即可
目标:看完就能上手,把日常手动操作变成脚本一键搞定


一、先认识一下 paramiko 是什么

paramiko 是一个 Python 库,专门用来做 SSH 相关的操作。

平时我们用 XShell、PuTTY、FinalShell 连服务器,底层走的就是 SSH 协议。paramiko 相当于把这些操作搬到了 Python 代码里——你不用打开那些软件,写几行脚本就能完成登录、敲命令、拿结果、传文件这些事。

安装方法

pip install paramiko

如果网络慢,加个国内源:

pip install paramiko -i https://pypi.tuna.tsinghua.edu.cn/simple

装完就可以用了。


二、最基础的:SSH 登录 + 执行一条命令

先写一个能跑起来的 demo,连上服务器、执行一条命令、看输出:

import paramiko

# 创建 SSH 客户端
client = paramiko.SSHClient()

# 自动接受陌生服务器的密钥(不弹确认框)
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 连上服务器
client.connect(
    hostname='192.168.1.100',   # 改成你的服务器 IP
    port=22,                     # SSH 端口,默认 22
    username='root',             # 登录用户名
    password='你的密码'           # 登录密码
)

# 执行一条命令
stdin, stdout, stderr = client.exec_command('free -h')

# 拿到命令输出
output = stdout.read().decode()
print('=== 命令执行结果 ===')
print(output)

# 看看有没有报错
err_msg = stderr.read().decode()
if err_msg:
    print('=== 错误信息 ===')
    print(err_msg)

# 关掉连接
client.close()

把上面的 IP、用户名、密码换成你自己的,保存为 ssh_demo.py,然后运行:

python ssh_demo.py

正常的话,你会看到远程服务器的内存使用情况。

💡 stdoutstderr 是啥?

  • stdout.read().decode() → 命令正常执行后返回的结果
  • stderr.read().decode() → 命令报错时的信息
  • 为什么要 .decode()?因为网络传回来的是字节(bytes),得转成我们能看懂的字符串

三、进阶:处理需要交互的场景(比如连路由器/交换机)

上面的方式,适合执行一条命令就拿结果走人。但有些设备不走寻常路——比如思科、华为的路由器/交换机,或者某些需要 sudo 提权后输密码的场景。这时候 exec_command 就不大够用了。

得换成 invoke_shell(),模拟一个真实的终端对话:

import paramiko
import time

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

client.connect(
    hostname='192.168.1.1',   # 你的路由器 IP
    port=22,
    username='admin',
    password='admin123'
)

# 打开一个交互式 Shell 通道
shell = client.invoke_shell()

# 等设备初始化完毕
time.sleep(1)

# 敲命令(每条命令后面要加 \n,模拟回车)
shell.send('display version\n')
time.sleep(2)  # 等设备返回结果

shell.send('display ip interface brief\n')
time.sleep(2)

# 把屏幕上的所有内容读出来
result = shell.recv(65535).decode()
print(result)

client.close()

那什么时候用哪个?这里给个判断方法:

场景 用什么 原因
Linux 服务器执行一条命令 exec_command 简单直接,一次一结果
连续执行多条命令 exec_command 也行,但每条单独调 每次都是独立连接
路由器/交换机登录 invoke_shell 需要模拟终端交互
需要 sudo 提权后输密码 invoke_shell 要等人机交互
登录后需要先进入某个模式 invoke_shell 类似 putty 手动操作

一句话总结:exec_command 是发一条命令拿一次结果,invoke_shell 是模拟你在终端里从头操作到尾。


四、实用的:通过 SFTP 上传/下载文件

光敲命令肯定不够,传文件也是高频需求——备份配置、上传更新包、下载日志。

paramiko 自带了 SFTP 功能,用法也挺简单:

上传文件到服务器

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
    hostname='192.168.1.100',
    port=22,
    username='root',
    password='你的密码'
)

# 开启 SFTP 通道
sftp = client.open_sftp()

# 上传本地文件到服务器
sftp.put(
    localpath='C:/backup/config.tar.gz',      # 本地文件路径
    remotepath='/home/backup/config.tar.gz'    # 服务器上的目标路径
)
print('上传完成 ✅')

sftp.close()
client.close()

从服务器下载文件

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
    hostname='192.168.1.100',
    port=22,
    username='root',
    password='你的密码'
)

sftp = client.open_sftp()

# 从服务器下载文件到本地
sftp.get(
    remotepath='/var/log/nginx/access.log',    # 服务器上的文件
    localpath='D:/logs/access.log'             # 保存到本地的路径
)
print('下载完成 ✅')

sftp.close()
client.close()

⚠️ 注意两点

  1. 本地路径的目录要提前建好,sftp 不会自动帮你创建文件夹
  2. 如果目标文件已存在,默认会直接覆盖

再加个小功能:列出远程目录的文件

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
    hostname='192.168.1.100',
    port=22,
    username='root',
    password='你的密码'
)

sftp = client.open_sftp()

# 列出远程目录
files = sftp.listdir('/home')
for f in files:
    print(f)

sftp.close()
client.close()

五、批量管理几十台服务器

上班最烦什么?几十台服务器,每台都要上去敲同样的命令。比如检查磁盘空间、更新配置文件、重启服务。

手动一台台连效率太低。用多线程做批量执行,几秒钟就跑完所有机器。

先搞清楚思路

不要一台台顺序执行(那样太慢),而是同时启动多个线程,每个线程负责连一台机器,并发执行。相当于同时开好几个终端窗口。

批量执行脚本

import paramiko
import threading
import sys


def run_command(ip, username, password, command, port=22):
    """在单台服务器上执行命令并打印结果"""
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(
            hostname=ip,
            port=port,
            username=username,
            password=password,
            timeout=10  # 超时控制,防止卡死
        )

        stdin, stdout, stderr = ssh.exec_command(command)
        result = stdout.read().decode()
        error = stderr.read().decode()

        print(f'\n{"="*50}')
        print(f'▶ 服务器: {ip}')
        print(f'▶ 执行: {command}')
        if result:
            print(f'▶ 结果:\n{result}')
        if error:
            print(f'▶ 错误:\n{error}')
        print(f'{"="*50}')

        ssh.close()
    except Exception as e:
        print(f'\n❌ 服务器 {ip} 连接失败: {e}')


def batch_run(hosts, command):
    """批量在多台服务器上执行命令(多线程并发)"""
    threads = []

    for ip, username, password in hosts:
        t = threading.Thread(
            target=run_command,
            args=(ip, username, password, command)
        )
        threads.append(t)
        t.start()  # 启动线程

    # 等待所有线程执行完毕
    for t in threads:
        t.join()

    print('\n全部执行完毕 ✅')


if __name__ == '__main__':
    # 服务器列表:每行是 (IP, 用户名, 密码)
    server_list = [
        ('192.168.1.10', 'root', 'pass123'),
        ('192.168.1.11', 'root', 'pass456'),
        ('192.168.1.12', 'root', 'pass789'),
    ]

    # 从命令行参数获取要执行的命令
    # 用法:python batch_ssh.py "df -h"
    cmd = sys.argv[1] if len(sys.argv) > 1 else 'uptime'

    batch_run(server_list, cmd)

怎么用?

# 检查所有服务器的磁盘空间
python batch_ssh.py "df -h"

# 查看所有服务器运行了多久
python batch_ssh.py "uptime"

# 重启某个服务
python batch_ssh.py "systemctl restart nginx"

把服务器列表单独放一个文件(更规范)

服务器信息写死在代码里,既不灵活也不安全。建议单独写个配置文件:

servers.txt

192.168.1.10 root pass123
192.168.1.11 root pass456
192.168.1.12 root pass789

然后在代码里读取这个文件:

def load_servers(file_path):
    """从配置文件加载服务器列表"""
    hosts = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith('#'):  # 跳过空行和注释
                parts = line.split()
                if len(parts) >= 3:
                    hosts.append((parts[0], parts[1], parts[2]))
    return hosts


if __name__ == '__main__':
    server_list = load_servers('servers.txt')
    cmd = sys.argv[1] if len(sys.argv) > 1 else 'uptime'
    batch_run(server_list, cmd)

以后加机器直接改 servers.txt 就行,不用动代码。


六、一个完整的实用脚本:每天定时备份服务器配置

把上面所有功能串起来,写一个实际能用的小工具——每天自动登录服务器,打包备份关键配置文件,下载到本地:

import paramiko
import os
from datetime import datetime


def backup_server(ip, username, password, backup_dir='./backups'):
    """备份单台服务器的关键配置"""
    today = datetime.now().strftime('%Y%m%d')
    
    # 在服务器上创建备份目录
    remote_backup = f'/tmp/backup_{today}'
    
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname=ip, username=username, password=password)

    # 执行备份命令
    commands = [
        f'mkdir -p {remote_backup}',
        f'cp /etc/nginx/nginx.conf {remote_backup}/ 2>/dev/null; echo "done"',
        f'cp /etc/ssh/sshd_config {remote_backup}/ 2>/dev/null; echo "done"',
        f'crontab -l > {remote_backup}/crontab.txt 2>/dev/null; echo "done"',
        f'tar -czf /tmp/backup_{today}.tar.gz -C /tmp backup_{today} 2>/dev/null'
    ]
    
    for cmd in commands:
        ssh.exec_command(cmd)

    # 用 SFTP 把打包好的文件下载到本地
    sftp = ssh.open_sftp()
    remote_file = f'/tmp/backup_{today}.tar.gz'
    local_file = f'{backup_dir}/{ip}_backup_{today}.tar.gz'
    
    os.makedirs(backup_dir, exist_ok=True)
    
    try:
        sftp.get(remote_file, local_file)
        print(f'✅ {ip} 备份完成 → {local_file}')
    except FileNotFoundError:
        print(f'⚠️ {ip} 备份文件未生成,可能是没有对应配置')
    
    # 清理远程临时文件
    ssh.exec_command(f'rm -rf /tmp/backup_{today}*')
    
    sftp.close()
    ssh.close()


def batch_backup(servers):
    """批量备份所有服务器"""
    threads = []
    for ip, user, pwd in servers:
        t = threading.Thread(target=backup_server, args=(ip, user, pwd))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    print('\n所有服务器备份完成 ✅')


if __name__ == '__main__':
    # 这里改成你自己的服务器列表
    my_servers = [
        ('192.168.1.10', 'root', 'pass123'),
        ('192.168.1.11', 'root', 'pass456'),
    ]
    
    batch_backup(my_servers)

把这个脚本加到 crontab(Linux)或者任务计划程序(Windows)里,每天自动跑一次就行。这样万一哪天手抖改坏了配置,还能找回之前的版本。


七、避坑指南(新手最容易踩的雷)

玩 paramiko 的过程中,这几个坑几乎每个人都会遇到,提前知道能省不少时间:

❌ 坑1:连接报错 Authentication failed

  • 原因:密码不对,或者服务器只支持密钥登录
  • 解决:先手动 SSH 一遍确认密码是对的;如果用密钥登录,参考下面的写法:
# 用密钥登录的方式
key = paramiko.RSAKey.from_private_key_file('C:/Users/me/.ssh/id_rsa')
client.connect(
    hostname='192.168.1.100',
    port=22,
    username='root',
    pkey=key
)

❌ 坑2:执行命令没输出

  • 原因:命令还没执行完就去读结果了
  • 解决:加个 time.sleep(1) 等一会儿,或者用 stdout.channel.recv_exit_status() 等命令真正结束再读

❌ 坑3:中文乱码

  • 原因:服务器返回的编码不是 UTF-8
  • 解决:试一下 decode('gbk')decode('utf-8', errors='ignore')
output = stdout.read().decode('gbk', errors='ignore')

❌ 坑4:连接多了报 socket.error

  • 原因:服务器限制了并发连接数,或者代码里没及时关闭连接
  • 解决:每次用完记得调用 ssh.close(),并发数别太大(建议不超过 20)

八、总结一下

功能 关键代码 适用场景
SSH 登录执行命令 exec_command('命令') Linux 服务器日常运维
交互式终端 invoke_shell() + send()/recv() 路由器/交换机、交互式程序
上传文件 sftp.put(本地, 远程) 部署代码、上传安装包
下载文件 sftp.get(远程, 本地) 拉取日志、备份配置文件
批量管理 threading.Thread + paramiko 几十台服务器同时操作

核心流程:

SSH 操作 = 创建客户端 → 设置策略 → 连接 → 执行操作 → 关闭

掌握了这些,日常大部分运维操作基本都能写成脚本自动化。剩下的无非是加一些错误处理、日志记录之类的完善工作。

如果你工作中有什么手动操作觉得很烦,不妨想想能不能用 paramiko 写成脚本。写完之后你会发现:以前那些重复劳动,其实都可以交给代码去做。


📝 说明:文中所有代码在 Python 3.6+ 环境下实测可用。
🐛 遇到问题先自查:① paramiko 装了吗 ② IP 和密码对吗 ③ 防火墙 22 端口开了吗

更多推荐