简介面向网络运维人员与Python初学者的轻量自动化工具包聚焦如何通过SSH协议批量连接交换机并下发命令解决多设备重复配置效率低下的问题。脚本基于paramiko实现并读取hosts.txt与commands.txt分别存放交换机IP列表和待执行命令清单方便使用者按需维护无需改动代码即可扩展设备规模。资源共3个文件包含1个Python脚本和2个文本模板压缩包仅2KB结构精简适合快速上手。已有169人学习下载。除可直接运行的SSH连接与命令执行脚本外还体现了异常处理、结果输出等基础思路读者可据此改造为带日志记录或并发执行的运维工具也可结合netmiko等库进一步实现配置备份、批量巡检等场景。1. 网络运维里最耗时间的动作手动登录交换机敲命令如果你管过几台、几十台交换机大概率经历过这样的场景接到一个需求要把全网接入层设备的某个 VLAN 接口配置统一改掉或者收集所有设备上某个接口的流量统计。于是你打开终端工具一台一台登录输入用户名密码敲命令截图或者复制粘贴退出再登录下一台。十台以内还能接受几十台下来半天就没了中间还容易漏记某台设备的回显。Python 连接交换机自动执行命令就是把这套重复劳动交给脚本去做的方案用 SSH 协议连上设备自动完成登录、进入特权模式、逐条下发命令、抓取回显并保存。这套东西适合正在做网络运维但没写过自动化脚本的人也适合已经用着工具但想摆脱手工批量操作的从业者它解决的核心问题是把「登录—敲命令—存结果」这一连串动作变成一条可重复执行的脚本命令。2. 从 SSH 到脚本化连接交换机前的选型与协议基础2.1 SSH 协议与设备侧配置为什么脚本能接管命令行交换机的命令行管理方式主要有 Telnet 和 SSH 两种。Telnet 是明文传输用户名密码和命令内容在网络里裸奔生产环境基本不推荐SSH 是加密通道脚本连接交换机时默认走的就是 SSH 协议。你手动用终端工具登录本质上也是建立一个 SSH 会话然后在这个会话里发送命令、读取回显。脚本做的事情和人一样只是把「交互式敲键盘」变成了「程序化发送字符串」。需要确认的是设备侧的 SSH 服务开启状态。不同厂商的交换机开启方式略有差异但逻辑一致生成加密密钥、启用 SSH 服务、配置认证方式。以常见的企业级交换机为例设备上需要确保有 host key并且 vty 线路下配置了 transport input ssh 或 transport input all。如果设备还是老旧的、只开了 Telnet脚本就得改用 telnetlib 或模拟终端方式但那属于兼容方案遇到这种情况我会优先建议先把设备侧 SSH 打开别让安全问题留在那里。另外一个关键点是设备的分页输出。手动操作时输出一屏会停下来等你按空格这叫分页。脚本执行时如果碰到分页命令发出去了但结果只回了一屏就不动了脚本会一直卡在读回显的地方。处理方式一般是登录后先发一条关闭分页的命令比如 screen-length 0 temporary 或 terminal length 0具体命令看厂商。这一步非常重要我在后面避坑章节里会专门展开。2.2 选型对比paramiko 与 netmiko 的边界Python 生态里连接交换机最常用的两个库是 paramiko 和 netmiko。paramiko 是底层 SSH 实现它只负责「建立 SSH 连接、收发数据」不管设备是什么品牌也不帮你处理命令回显的边界。netmiko 是在 paramiko 之上封装的网络设备专用库它对不同厂商的设备做了适配屏蔽了登录提示符、分页关闭、特权模式切换等差异。我的建议是如果只是临时连几台设备用 paramiko 完全够用还能让你看清 SSH 连接的本质但如果你要写一个要长期维护、要覆盖多种品牌设备的脚本直接用 netmiko省下的事情远比引入一个依赖多。netmiko 内部处理了很多坑比如设备回显里的 prompt 匹配、enable 模式的切换、分页关闭命令自动下发这些如果用 paramiko 裸写全部要自己处理。从依赖面看netmiko 会把你带进 paramiko 生态所以没有「用了 netmiko 就不用 paramiko」的说法而是在 paramiko 之上多了一层设备适配。选型时还要考虑执行环境有没有网段限制如果脚本所在机器到交换机管理网段不通换什么库都没用。2.3 环境准备与连接参数一个能跑起来的最小骨架先准备 Python 环境建议 3.8 以上版本。虚拟环境里安装依赖没有特殊网络环境的情况下一条命令就能完成python -m venv netops_env source netops_env/bin/activate pip install netmiko paramiko创建虚拟环境是为了隔离依赖。netmiko 会依赖 paramiko所以装 netmiko 时 paramiko 会被一起装进来但单独写一两行字把它点出来是方便你在排查问题时意识到底层是 paramiko 在管连接。接着准备设备连接参数netmiko 的核心参数是 device_type、host、username、password其中最需要注意的是 device_type它决定了 netmiko 用哪一套交互逻辑去适配设备。from netmiko import ConnectHandler device { device_type: huawei, host: 192.0.2.10, username: netadmin, password: your_password, port: 22, timeout: 30, conn_timeout: 15, } with ConnectHandler(**device) as conn: output conn.send_command(display version) print(output)这段代码是连接交换机的最简骨架。device_type 填的是厂商和系统类型netmiko 支持的取值包括 cisco_ios、huawei、hp_comware、arista_eos 等timeout 是单条命令的执行超时conn_timeout 是建立 TCP 连接的超时。password 这块我先用占位符生产环境不建议明文写在脚本里后面会有专门章节讲怎么处理。运行脚本后如果打印出了设备版本信息说明连接链路、认证、命令发送、回显读取这一整条链路都已经通了。接下来的章节把这套骨架展开逐步加上健壮性处理。3. 核心脚本连接交换机并自动执行命令的完整流程3.1 连接建立与身份认证先解决连得上的问题连接建立这一步netmiko 的 ConnectHandler 会把 TCP 连接、SSH 握手、用户名密码认证、登录后进入用户视图这几个步骤全部封装掉。真正要关注的是认证失败和连接超时两个分支。认证失败时设备返回的提示是 Authentication failed这时候脚本要做的不是继续执行而是记录下来并跳到下一台设备。from netmiko import ConnectHandler from netmiko.ssh_exception import NetmikoAuthenticationException, NetmikoTimeoutException device { device_type: huawei, host: 192.0.2.11, username: netadmin, password: wrong_password, timeout: 20, } try: conn ConnectHandler(**device) output conn.send_command(display clock) print(output) conn.disconnect() except NetmikoAuthenticationException: print(f{device[host]} 认证失败请检查用户名或密码) except NetmikoTimeoutException: print(f{device[host]} 连接超时请检查管理地址是否可达、SSH服务是否开启)逻辑说明try 块里做连接和命令执行两个 except 分别捕获认证失败和超时。认证失败时不需要重试直接标记设备并跳过因为重试只会加重设备侧的登录失败锁定风险连接超时则需要区分原因可能是网络不通也可能是设备 SSH 服务没起。捕获异常后打印的提示信息里带上了主机地址方便事后定位是哪台设备出了问题。参数说明timeout 是命令执行等待时间conn_timeout 建议单独设置。某些设备在 SSH 握手阶段处理很慢如果 conn_timeout 设太短可能出现「设备其实活着但脚本报超时」的误判。我一般把 conn_timeout 设成 15 秒timeout 设成 30 秒如果设备特别慢再往大调。3.2 命令发送与回显捕获从 send_command 到异常处理连接建立后最常用的方法是 send_command。它会向设备发送一条命令然后把回显全部读出来。这里有一个隐藏逻辑netmiko 发送命令后会一直读取输出直到再次匹配到设备的提示符才算这条命令执行完成。所以 send_command 返回的字符串是完整回显不是半截。实际使用中命令回显的数据量差异很大。查询接口状态可能只有几行查询完整配置可能几百上千行。netmiko 内部根据设备型号自动处理了分页问题但如果是用 paramiko 裸连就得自己想办法。下面这段代码展示用 netmiko 执行多条命令并分别保存回显from netmiko import ConnectHandler import datetime device { device_type: huawei, host: 192.0.2.12, username: netadmin, password: your_password, } commands [display version, display interface brief, display vlan] conn ConnectHandler(**device) timestamp datetime.datetime.now().strftime(%Y%m%d_%H%M%S) for cmd in commands: output conn.send_command(cmd) filename f{device[host]}_{cmd.replace( , _)}_{timestamp}.txt with open(filename, w, encodingutf-8) as f: f.write(output) print(f命令 [{cmd}] 的回显已保存到 {filename}) conn.disconnect()逻辑说明commands 列表里放的是要在设备上执行的命令循环里逐条执行并写入独立文件。文件名里带上了设备 IP、命令关键字和时间戳这样事后找某台设备某条命令的结果时非常直接。执行完所有命令后再统一 disconnect避免频繁建立连接。参数说明如果某条命令执行时间特别长可以给 send_command 加 read_timeout 参数例如 conn.send_command(display cpu, read_timeout60)。不使用这个参数时netmiko 默认用的是连接时设置的 timeout 值。另有一个场景是命令需要进入交互确认比如清空某类配置时设备会问确认这时候 send_command 不够用要用 send_command_timing按设备提示逐次发送确认指令。3.3 执行模式切换与特权命令enable 与 config 模式的处理交换机的命令分视图。用户视图下只能做查询进入系统视图后才能改配置某些设备还有单独的特权模式用来执行 save、reboot 这类高危操作。netmiko 用 send_command_timing 或者直接调用 enable() 方法来处理模式切换。from netmiko import ConnectHandler device { device_type: huawei, host: 192.0.2.13, username: netadmin, password: your_password, secret: enable_password, } conn ConnectHandler(**device) conn.enable() output conn.send_command(display current-configuration) print(output) conn.disconnect()逻辑说明enable() 方法完成从用户视图到特权模式的切换设备提示输入 enable 密码时netmiko 自动用 secret 参数里配置的密码去应答。有的设备开启 local privilege levelenable 密码和登录密码是同一个这种情况下 secret 可以和 password 填一样。进入特权模式后可以执行 save、reboot 之类的特权命令但这类命令有风险我在后面会建议加确认机制。参数说明secret 参数不是必填但如果设备配置了 enable 密码不传这个参数enable() 执行时会因为无法应答密码而失败。某些设备上进入系统视图也用 enable()某些则必须先 send_command(system-view) 再执行配置命令最好先在测试环境里确认设备行为再铺开。如果要在配置模式下批量下发配置需要逐条发送命令并确认回显。netmiko 的 send_config_set 可以接受一个命令列表并进入配置模式逐条下发但它依赖设备提示符匹配遇到不同厂商差异时需要反复验证。我一般把配置下发类操作单独封装一个函数先做全量测试再上生产。4. 批量执行与安全加固让脚本真正落地到生产环境4.1 批量交换机执行串行与线程并行的取舍单台设备的脚本跑通后自然会想到对一批设备执行同样命令。最简单的方式是循环一台一台串行处理。from netmiko import ConnectHandler from netmiko.ssh_exception import NetmikoAuthenticationException, NetmikoTimeoutException devices [ {device_type: huawei, host: 192.0.2.21, username: netadmin, password: your_password}, {device_type: huawei, host: 192.0.2.22, username: netadmin, password: your_password}, {device_type: huawei, host: 192.0.2.23, username: netadmin, password: your_password}, ] for device in devices: try: conn ConnectHandler(**device) output conn.send_command(display cpu) print(f {device[host]} ) print(output) conn.disconnect() except (NetmikoAuthenticationException, NetmikoTimeoutException) as e: print(f{device[host]} 执行失败: {str(e)})逻辑说明devices 列表里维护一批设备连接参数for 循环逐台连接、执行、退出。这样写的好处是逻辑清晰失败时能明确知道是哪台设备出的问题。几十台设备以内串行执行完全够用一台设备耗时几秒二十台也就是一两分钟的事。当设备数量达到上百台时串行等待时间会比较明显这时考虑用线程池并发执行from concurrent.futures import ThreadPoolExecutor, as_completed from netmiko import ConnectHandler def run_on_device(device): host device[host] try: conn ConnectHandler(**device) output conn.send_command(display memory) conn.disconnect() return host, output except Exception as e: return host, fERROR: {str(e)} devices [...] # 同前 with ThreadPoolExecutor(max_workers10) as executor: futures {executor.submit(run_on_device, dev): dev[host] for dev in devices} for future in as_completed(futures): host, result future.result() print(f {host} ) print(result)逻辑说明ThreadPoolExecutor 创建线程池把每台设备的连接和执行任务丢进去并行跑。max_workers 是并发数也就是同时建立多少个 SSH 连接。返回结果用 as_completed 收集谁先执行完谁先处理不需要等待所有任务结束。参数说明max_workers 的取值要克制。并发数太高设备侧 SSH 连接数会冲破限制常见的报错是 No more connections 或者连接被重置。我一般根据设备型号把并发压在 5 到 15 之间老设备并发更小。线程并发的收益在设备数量多时才明显别一上来就上并发很多生产事故就是并发过大导致管理网段带宽被顶满。4.2 密码与密钥管理不让明文口令留在脚本里脚本里写死密码是最省事、也最危险的做法。只要脚本文件被传阅、被提交到代码仓库口令就等于泄露了。常见做法是用环境变量或者本地密钥库来存放口令。环境变量的方式改动最小import os from netmiko import ConnectHandler device { device_type: huawei, host: 192.0.2.31, username: os.environ.get(NET_DEVICE_USER), password: os.environ.get(NET_DEVICE_PASS), } conn ConnectHandler(**device) output conn.send_command(display clock) print(output) conn.disconnect()逻辑说明运行时从环境变量里读取用户名和密码。脚本本身不包含明文口令即便文件被拷走拿不到运行环境里的环境变量就没法连设备。执行前在 shell 里设置环境变量或者在 CI 平台里配置好对应的 secret 项。我在生产环境里一般两种方式结合CRT 工具连接时用 SSH key但脚本执行的时候仍然用口令因为很多老设备并不支持密钥认证。这种情况下口令的存放建议用专门的密钥管理工具避免散落在各台机器上。密码变更后只需要更新密钥库脚本本身不用动。4.3 结果落盘与日志执行完以后怎么回溯批量执行完结果如果只打在屏幕上事后回溯非常痛苦。我会把每次执行的任务信息、设备 IP、命令、回显全文、执行状态写成一个带时间戳的目录目录下每台设备一个文件。import os import json import datetime from netmiko import ConnectHandler devices [ {device_type: huawei, host: 192.0.2.41, username: netadmin, password: your_password}, {device_type: huawei, host: 192.0.2.42, username: netadmin, password: your_password}, ] task_time datetime.datetime.now().strftime(%Y%m%d_%H%M%S) base_dir ftask_{task_time} os.makedirs(base_dir, exist_okTrue) summary [] for device in devices: host device[host] result {host: host, status: OK, error: None, output: None} try: conn ConnectHandler(**device) output conn.send_command(display version) conn.disconnect() result[output] output with open(os.path.join(base_dir, f{host}.txt), w, encodingutf-8) as f: f.write(output) except Exception as e: result[status] FAILED result[error] str(e) summary.append(result) with open(os.path.join(base_dir, summary.json), w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2) for item in summary: print(f{item[host]} - {item[status]})逻辑说明每个任务运行前生成一个 task_ 前缀的时间戳目录避免多次运行结果互相覆盖。设备回显写入以 IP 命名的文本文件执行状态、错误信息汇总到 summary.json 里。脚本跑完扫一眼 summary 就能知道哪些设备成功、哪些失败不需要逐个打开文件。参数说明json.dump 里设置 ensure_asciiFalse是为了让中文错误信息在 JSON 文件里保持可读。indent2 让文件结构清晰方便直接用文本编辑器查看。如果后续要做数据统计也可以把 summary 存成 CSV用 pandas 或 Excel 打开处理。5. 避坑连接交换机时常见的五个翻车现场5.1 未知主机密钥导致连接中断现象脚本第一次连接某台设备时报错提示 Host key not found 或 Server host key 不匹配程序直接退出。原因SSH 客户端首次连接时会检查设备的主机密钥不存在或者与 known_hosts 中记录的不一致时客户端会拒绝继续。netmiko 内部用 paramiko 实现 SSH默认策略会校验 host key新设备自然而然触发这个问题。解决在设备连接参数里指定 SSH 策略或者在环境层面统一管理 known_hosts 文件。最常见的做法是允许未知主机密钥加入但生产环境要注意中间人风险更稳妥的是先手动连接一次把设备密钥加入 known_hostsfrom netmiko import ConnectHandler device { device_type: huawei, host: 192.0.2.51, username: netadmin, password: your_password, conn_timeout: 15, } conn ConnectHandler(**device) output conn.send_command(display clock) print(output) conn.disconnect()如果仍然报 host key 相关错误可以改用 paramiko 的 AutoAddPolicy但这属于「临时绕过」的范畴长期维护时不建议全局使用。更好的方案是让脚本维护一个独立的 known_hosts 文件把设备密钥变更纳入变更审批流程而不是静静地自动接受。5.2 分页导致输出被截断现象发送 show running-config 这类长输出命令回显只返回了第一屏后面的内容一直没有输出脚本长时间不返回或返回残缺内容。原因设备默认开启了分页显示输出满屏后等待用户交互脚本不发送空格或回车设备就一直等待下去。解决登录后先关闭分页。netmiko 在连接时根据设备类型做了适配多数主流设备会自动执行关闭分页命令例如思科设备发 terminal length 0华为设备发 screen-length 0 temporary。但如果你用 paramiko 裸写必须在登录后手动发送关闭分页命令否则输出必断。关闭分页的命令要确认临时生效还是永久生效还是用临时参数比较保守避免改了设备配置。5.3 特权模式切换失败现象调用 enable() 后收到错误提示或者后续命令返回权限不足某些 display 命令能执行但 save 命令被拒绝。原因设备配置了 enable 密码但脚本的 secret 参数没有设置或者设备使用本地用户提权enable 密码和登录密码不同。部分设备还需要再 send_command(enable) 之后额外等待一个交互提示netmiko 的 enable() 没等到预期提示符就会超时。解决确认设备上 enable 密码是否存在、是否与登录密码一致然后把正确的密码填到 secret 参数里。仍不稳定时用 send_command_timing 手动完成 enable 过程先发 enable读到 Password: 提示后再发密码这样交互过程完全可控也方便打日志排查。5.4 缓冲读取时的空结果现象send_command 返回了空字符串或者只返回命令本身但没有回显内容偶发情况下同一段代码这次成功、下次却是空的。原因设备响应慢命令发出后回显还没到netmiko 的读循环就已经超时返回了或者命令本身无输出但被误判为正常。缓冲区的读取时机和设备的回显速度之间存在竞态。解决给 send_command 显式设置 read_timeout并适当调大。如果命令确实可能有空回显改用 send_command_timing并在其后主动读取设备提示符之前的全部数据。另外排查执行环境是否存在代理或跳板机代理的超时同样会导致回显读取失败。5.5 并发连接过多导致设备拒绝现象批量脚本并发数调到 20、30 后部分设备报 SSH 连接被拒绝或者连接建立但认证立刻失败设备 CPU 在任务执行期间飙高。原因多数交换机的 SSH 服务有最大连接数限制通常个位数到十几个超过后新连接直接拒绝。线程并发方式下所有线程几乎同时发起连接很容易打破这个上限。解决把 max_workers 调整到设备能接受的范围之内比如 5 左右单台设备执行完再释放连接。确认设备侧 ssh server session 限制配置必要时调整。并发场景下还容易出现设备侧登录锁定账号密码错误次数多了设备会把来源 IP 列入黑名单这种情况连手动登录都会受影响只能等待锁定超时或手工解锁。6. 进阶把脚本固化成定时巡检任务6.1 生成巡检报告并输出 CSV单次执行已经能解决很多问题但如果想让它发挥持续价值下一步是把脚本变成定时巡检工具。巡检的核心是「按固定周期抓关键指标、对比历史数据、发现异常」。我把设备信息输出整理成 CSV方便后续用电子表格处理。import csv from netmiko import ConnectHandler devices [ {device_type: huawei, host: 192.0.2.61, username: netadmin, password: your_password}, {device_type: huawei, host: 192.0.2.62, username: netadmin, password: your_password}, ] rows [] for device in devices: try: conn ConnectHandler(**device) version conn.send_command(display version) cpu conn.send_command(display cpu) mem conn.send_command(display memory) conn.disconnect() rows.append({ host: device[host], version: version.splitlines()[0] if version else , cpu: cpu.splitlines()[1] if len(cpu.splitlines()) 1 else , memory: mem.splitlines()[1] if len(mem.splitlines()) 1 else , }) except Exception as e: rows.append({host: device[host], version: fERROR: {e}, cpu: , memory: }) with open(inspection_report.csv, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnames[host, version, cpu, memory]) writer.writeheader() writer.writerows(rows) print(巡检报告已生成: inspection_report.csv)逻辑说明每台设备抓取版本、CPU、内存三类信息经过简单的文本截取后存入字典列表最后统一写入 CSV。encoding 用 utf-8-sig 是让 Excel 直接打开 CSV 时中文不乱码这是很多人容易漏掉的细节。text split 处理只适合结构稳定的回显如果设备回显格式不稳定建议把原始回显存文件CSV 里只放状态码。参数说明csv.DictWriter 的 fieldnames 顺序决定了 CSV 列的顺序。newline 是防止行与行之间出现空行这个坑在 Windows 环境下最明显。6.2 定时执行与失败重试定时执行最常见的方式是系统自带的定时任务Linux 里用 cronWindows 里用任务计划程序。脚本本身保持可单次执行的状态定时器只是到点把它拉起来。失败重试的逻辑放在脚本内部import time from netmiko import ConnectHandler def exec_with_retry(device, command, retries3, interval10): for attempt in range(1, retries 1): try: conn ConnectHandler(**device) output conn.send_command(command) conn.disconnect() return output except Exception as e: print(f第 {attempt} 次尝试失败: {e}) if attempt retries: time.sleep(interval) raise RuntimeError(f设备 {device[host]} 重试 {retries} 次后仍失败) device { device_type: huawei, host: 192.0.2.63, username: netadmin, password: your_password, } output exec_with_retry(device, display clock) print(output)逻辑说明exec_with_retry 包装了连接和执行动作失败后按设定的间隔重试。重试次数和间隔要根据场景调巡检任务间隔一般比较长重试间隔设 10 秒到 30 秒都没有问题但如果是用户手动触发的脚本重试间隔太长反而让人等得烦躁。参数说明interval 参数控制重试等待秒数避免连续快速重试把设备连接数打满。这里有个细节重试时要捕获的是连接层的异常如果连接成功但命令执行报错可能不需要重试因为重试大概率得到同样的错误。6.3 验证脚本的三种方式脚本写完不是直接丢到生产环境跑需要先验证。我的习惯是先用一台测试设备跑单条查询命令确认连接参数、设备型号适配都没有问题。然后在模拟环境或者备用设备上执行批量任务确认输出文件名、内容格式符合预期。最后在少量生产设备上小范围执行观察设备 CPU 和管理通道的负载情况。验证时要特意制造几个异常场景密码错误、设备不可达、命令拼写错误确认脚本能够捕获异常并给出清晰的错误信息。一个容易被忽视的检查点是脚本的退出码定时任务依赖退出码判定任务成功还是失败如果脚本内部处理了所有异常但没有把失败状态传递出去定时任务会误以为一切正常。从那以后我每次把脚本交给定时任务之前都会强制走一遍「正常执行、异常模拟、超时模拟」三组测试确认退出码符合预期再放上去。希望帮到你。本文还有配套的精品资源点击获取