You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何编写Python脚本用paramiko SSH批量连接Linux设备并执行命令

原理解释(大白话版)

你可以把整个流程类比成给小区每家送快递:

  • 你手里有两张清单:一张是所有要送快递的住户地址+开门密码(对应主机列表txt),另一张是要给每家递的东西清单(对应命令列表txt)
  • paramiko就是你雇的快递员,他会走SSH这个专用通道上门
  • 流程就是:先把两张清单的内容都读熟,然后让快递员挨家跑:
    1. 到一户人家,核对地址密码开门(SSH连接设备)
    2. 按照递东西的清单,一样一样递给住户,住户给你签收条(执行命令,拿到返回结果)
    3. 这家送完关门,去下一家,直到所有住户都送完

那些复杂脚本加的多余逻辑大多是处理特殊场景的,你的基础需求完全不需要用到。


前置准备
  1. 先安装依赖库:pip install paramiko
  2. 准备两个纯文本文件,和脚本放在同一个文件夹里:
  • hosts.txt:每行写一台设备的信息,格式为IP地址,用户名,密码,示例内容:

192.168.1.10,root,test123
192.168.1.11,admin,abc456
192.168.1.12,root,root789

  • cmds.txt:每行写一条要执行的命令,示例内容:

uname -r
pwd
df -h
free -h


批量执行脚本代码
import paramiko

# 读取主机列表文件
def read_hosts(file_path):
    hosts = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            # 跳过空行
            if not line:
                continue
            ip, user, pwd = line.split(',')
            hosts.append((ip.strip(), user.strip(), pwd.strip()))
    return hosts

# 读取命令列表文件
def read_cmds(file_path):
    cmds = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            # 跳过空行
            if line:
                cmds.append(line)
    return cmds

if __name__ == '__main__':
    # 加载两个配置文件的内容
    host_list = read_hosts('hosts.txt')
    cmd_list = read_cmds('cmds.txt')

    # 遍历每台设备执行操作
    for ip, user, pwd in host_list:
        print(f"\n===== 正在处理设备:{ip} =====")
        try:
            # 以下是你原有单设备逻辑的复用
            ssh_client = paramiko.SSHClient()
            ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            # 加10秒超时避免连接失败卡很久
            ssh_client.connect(hostname=ip, username=user, password=pwd, timeout=10)

            # 遍历执行所有命令
            for cmd in cmd_list:
                print(f"\n执行命令:{cmd}")
                stdin, stdout, stderr = ssh_client.exec_command(cmd)
                # 分别读取正常输出和错误输出
                out = stdout.read().decode('utf-8')
                err = stderr.read().decode('utf-8')
                if out:
                    print(f"输出:\n{out}")
                if err:
                    print(f"错误:\n{err}")
            
            # 关闭当前设备的SSH连接
            ssh_client.close()
            print(f"===== 设备 {ip} 处理完成 =====")
        except Exception as e:
            print(f"处理设备 {ip} 失败,错误信息:{str(e)}")
            # 单台设备出错不影响后续设备处理
            continue

脚本逻辑拆解
  • 先写了两个工具函数,分别负责读取主机文件和命令文件,自动跳过空行,把内容整理成方便调用的格式
  • 主程序先加载两个文件的内容,拿到所有要处理的设备和要执行的命令
  • 外层循环遍历每一台设备:
    • 用你原来的逻辑创建SSH客户端,设置自动接受主机密钥,尝试连接,加10秒超时避免长时间卡顿
    • 连接成功就进入内层循环,挨个执行所有命令,把正常输出和错误输出都打印出来
    • 执行完所有命令就关闭当前设备的SSH连接
    • 如果连接或者执行过程中出任何错,就打印错误信息,跳过这台设备继续处理下一台
  • 所有设备处理完脚本自动结束

内容的提问来源于stack exchange,提问作者rick_sorkin

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.06 04:39:03