尧图精选

手写一个SSH MCP Server:让AI直接操作远程Linux服务器

🕒 发布时间:2026/10/1 19:26:17 📁 来源:尧图网络
1. 写在前面为什么我会手搓一个 SSH MCP Server先聊个现象。我身边很多朋友现在都习惯让 AI 帮忙写脚本、查资料但真要让 AI 直接去操作一台远程服务器大多数人还是犹豫的——不是不想而是没找到顺手的路子。我也一样。直到 MCPModel Context Protocol这个概念被炒热我意识到一个关键点如果只把 AI 当聊天窗口它就永远只是个“脑子”可一旦把它接上 SSH它就等于拥有了手脚能登录服务器、看日志、改配置、跑命令甚至帮我把整个排查流程完整走一遍。这个项目做的就是这件事用 Python 写一个原生 SSH MCP Server让支持 MCP 的客户端比如 Trae IDE、Claude Desktop 或者其他兼容工具能够通过自然语言直接操纵远程 Linux 服务器。你不需要在对话里贴命令让 AI 猜而是由 AI 自己调用定义好的工具在远端真正执行 ls、cat、scp 这些操作然后把结果拿回来继续推理。这篇文章面向的读者有两类一是已经接触过 MCP 但还没动手写服务端的人想看看一个可落地的 Server 到底怎么写二是手里管着几台 Linux 服务器、想把运维操作交一部分给 AI 的开发者。文章里会给出完整代码、配置思路、安全策略和我实测踩过的一堆坑。读完之后你应该能在 20 分钟内搭出一个自己的 SSH MCP Server。2. 为什么坚持“原生 SSH”而不是直接用 paramiko2.1 复用 ssh_config省掉一堆“连接配置”写 Python 连 SSH很多人的第一反应是上 paramiko。paramiko 没问题但它在实际落地时会带来两个烦恼第一它不认你~/.ssh/config里的别名、跳板机、端口映射和密钥规则你得在代码里重新维护一套连接配置第二它把 SSH 的密钥链和 agent 机制隔在了一层皮后面本地习惯好的同事用ssh ops-server一条命令能搞定的事在 paramiko 里还要单独处理。原生 SSH 就简单了。我直接在代码里调系统自带的ssh、scp命令所有连接细节都交给 OpenSSH 这个已经被全球运维用了二十多年的成熟组件。你只需要在配置文件里写好别名Host ops-server HostName 192.168.1.101 User deploy Port 2222 IdentityFile ~/.ssh/ops_ed25519之后 MCP Server 里的工具只需要传一个alias比如ops-server剩下的事情全部由 SSH 客户端处理。好处是显而易见的团队里任何一个人都能把自己的密钥和跳板配置直接复用不需要额外维护业务侧的连接信息。2.2 减少依赖、进程隔离出了问题不牵连主进程安全性和稳定性也是我选原生命令的重要原因。paramiko 在做长连接复用、断线重连时代码复杂度会直线上升而原生 SSH 每次调用都是一个独立子进程连接、认证、超时、退出都由 OpenSSH 自身管理。子进程崩了不会影响 MCP Server 本体排查问题的时候只要看子进程的状态就行心智负担小很多。再一个现实因素是依赖体积。paramiko 本身会携带 cryptography 这类重量级依赖在部分内网环境或者精简版 Python 镜像里安装起来很痛苦。而原生 SSH 方案Python 侧只依赖mcp这一个包真正的核心能力全部由 OpenSSH 提供环境随处可跑。3. 核心实现MCP Server 的整体架构与代码骨架3.1 架构总览整个项目的结构分成三层。最外层是 MCP 客户端也就是你日常使用的 IDE 或者 MCP 调试工具中间是我们要写的 MCP Server 进程它通过 stdio 与客户端通信最底层是 SSH 工具链MCP Server 通过 subprocess 调用系统ssh和scp。这三层关系很简单客户端下达工具调用指令Server 解析参数并组装 ssh 命令执行后把 stdout、stderr 拼起来返回给 AI。为什么要用 stdio 而不是 HTTP 或 SSE因为 stdio 模式最简单客户端启动 Server 子进程后用标准输入输出直接通信不需要考虑端口占用、鉴权、跨域问题。MCP 客户端在本地启动项目时这种模式是首选。如果你想部署到远程机器再考虑 streamable HTTP但那是后话先把本地跑通最重要。3.2 代码目录与依赖安装我习惯先建一个干净的项目目录然后再动手mkdir ssh-mcp-server cd ssh-mcp-server python3 -m venv venv source venv/bin/activate pip install mcp[cli]Python 版本要求 3.10 以上这没什么好商量的MCP 官方 SDK 用了不少新语法老版本跑不起来。装完依赖之后目录里只需要三个核心文件config.py放配置ssh_tools.py放底层 SSH 执行逻辑server.py放 MCP Server 入口。3.3 Server 入口FastMCP 让工具定义变得异常简单MCP 官方 Python SDK 提供了 FastMCP 这个封装层写起来类似 FastAPI装饰器一挂就是一个工具。直接看代码from mcp.server.fastmcp import FastMCP from ssh_tools import execute_remote_command, list_remote_dir mcp FastMCP(ssh-mcp-server) mcp.tool() def execute_command(alias: str, command: str, timeout: int 30) - str: 在指定服务器上执行一条 shell 命令返回命令输出。 Args: alias: ssh_config 中配置的主机别名 command: 要执行的命令 timeout: 超时时间单位秒 return execute_remote_command(alias, command, timeout) mcp.tool() def list_dir(alias: str, remote_path: str) - str: 列出远程目录内容。 Args: alias: ssh_config 中配置的主机别名 remote_path: 远程目录路径 return list_remote_dir(alias, remote_path) if __name__ __main__: mcp.run()FastMCP 的mcp.tool()装饰器会自动把函数的签名、文档字符串、参数类型转换成 MCP 协议里的工具定义。AI 客户端拿到这些定义之后就会在合适场景下自动生成参数并调用。这就是为什么你的函数注释必须写清楚参数含义——AI 就是靠这些注释来理解怎么调用的。有一点需要提醒工具函数的 docstring 一定要写清楚参数格式尤其是command这种自由文本参数AI 会根据描述自动决定传什么值。如果描述写得含糊AI 就可能传一个带引号包裹整条命令的字符串后面执行阶段就会出问题。3.4 SSH 执行器正确姿势是 subprocess 列表传参底层执行器的核心代码如下这一步是全文最容易出错的地方import subprocess import re import logging logger logging.getLogger(ssh_mcp) def _sanitize_alias(alias: str) - str: if not re.fullmatch(r[A-Za-z0-9_-], alias): raise ValueError(f非法的主机别名: {alias}) return alias def execute_remote_command(alias: str, command: str, timeout: int 30) - str: alias _sanitize_alias(alias) ssh_cmd [ ssh, alias, -o, ConnectTimeout10, -o, BatchModeyes, ftimeout {int(timeout)} {command} ] try: result subprocess.run( ssh_cmd, capture_outputTrue, textTrue, timeoutint(timeout) 15, ) except subprocess.TimeoutExpired: return [执行超时命令已被终止] output result.stdout result.stderr logger.info(cmd on %s: %s, exit%s, alias, command, result.returncode) return output[:8000]两个关键点。第一ssh_cmd必须是一个字符串列表不能拼成一个长字符串丢给shellTrue。如果用了shellTrue本地这层子进程就会把 alias 当作 shell 参数解析相当于给本机开了远程代码执行的口子。用列表传参、shellFalse参数就不会被本地方 shell 篡改。第二远端命令本身会被 ssh 发到对方的 shell 去执行所以我加了一个timeout命令包裹层。这个 timeout 是 GNU coreutils 自带的绝大多数 Linux 发行版都有它能在指定秒数后杀掉远端命令防止 AI 调出个tail -f之类永不结束的命令把 MCP Server 进程拖死。4. 文件传输让 AI 能上传下载但每一步都在可控范围4.1 上传与下载工具的封装能执行命令已经算是“有手脚”了但真正让 AI 处理实际工作场景的是文件传输。比如 AI 拿到了服务器上某个日志文件的内容想让你本地留存或者 AI 帮你写好了一段 Nginx 配置想直接放回服务器。这时候就需要封装 scp 工具。scp 和 ssh 有一个非常阴间的区别ssh 的端口参数是小写-pscp 的端口参数是大写-P。这个坑我至少踩过三次每次都是报错之后盯着命令看了半天才想起来。封装时我直接忽略端口配置让系统去读 ssh_config连接细节不用代码操心def upload_file(alias: str, local_path: str, remote_path: str) - str: alias _sanitize_alias(alias) scp_cmd [scp, local_path, f{alias}:{remote_path}] result subprocess.run(scp_cmd, capture_outputTrue, textTrue, timeout60) return result.stdout result.stderr def download_file(alias: str, remote_path: str, local_path: str) - str: alias _sanitize_alias(alias) scp_cmd [scp, f{alias}:{remote_path}, local_path] result subprocess.run(scp_cmd, capture_outputTrue, textTrue, timeout60) return result.stdout result.stderr调用 scp 时同样要使用列表传参。这里的风险点主要在路径上如果 local_path 以-开头scp 会把它当成参数选项解析所以最好在传给 scp 之前做一次校验拒绝所有以-开头或者含明显路径穿越的输入。4.2 文件操作的目录白名单文件上传下载比命令执行更危险。一次误操作rm -rf可能只影响一个目录但scp把本地文件覆盖到生产目录或者把远程/etc/shadow拉回本地都是足以让人睡不着觉的场景。我的方案是引入目录白名单ALLOWED_REMOTE_PREFIXES (/home/deploy/www, /home/deploy/logs, /tmp) def _check_remote_path(remote_path: str) - str: normalized remote_path if not normalized.startswith(ALLOWED_REMOTE_PREFIXES): raise ValueError(f远程路径不在允许范围内: {remote_path}) if .. in normalized: raise ValueError(不允许包含 .. 的路径) return normalized注意str.startswith存在一个隐患/home/deploy/www-backup也会通过/home/deploy/www的校验。所以路径检查时最好把前缀按目录层级比较比如先os.path.normpath规范化再确认规范化后的路径等于某个允许目录或者是该目录的子路径。这种细节看着琐碎但安全防护就是由一个一个细节堆出来的。5. 安全设计给 AI 的手脚套上缰绳5.1 为什么要做命令白名单假设你辛辛苦苦把 AI 接入服务器第一句话就是“帮我清一下磁盘空间”AI 大概率会执行rm -rf类命令去删除文件。如果你没有约束它完全可能把系统目录删出问题。所以我的项目里默认开启一个安全策略只读模式自动开启白名单外的写操作直接拒绝。实现上不复杂在执行命令之前检查命令的第一个 tokenREADONLY_COMMANDS { ls, pwd, cat, tail, head, grep, find, df, free, ps, uname, uptime, whoami, date, du, stat, systemctl, journalctl } def _check_command_allowed(command: str) - None: first_token command.strip().split()[0] if first_token not in READONLY_COMMANDS: raise ValueError(f命令 {first_token} 不在白名单中请在配置中显式启用该命令)很多朋友看到这里会问只让 AI 跑白名单命令那它还能干多少活我的理解是第一阶段AI 的主要价值是快速理解系统状态、定位问题而不是替你做破坏性变更。等你在生产环境验证一段时间、对它足够信任了再把白名单扩展到 git、systemctl restart 这类可控操作。上线 AI 运维助手安全边界永远要比能力边界更保守。5.2 危险命令模式的黑名单白名单能拦住“AI 主动调用危险命令”但拦不住“AI 被已有命令诱导执行危险操作”。比如你让它cat /var/log/nginx/error.log日志里恰好有一行提示mkfs.ext4 /dev/sdb这样可疑内容AI 可能觉得这是个好建议并继续执行。所以黑名单同样必要。我在代码里维护了一组危险模式的正则DANGEROUS_PATTERNS [ r\brm\s-rf\s/, r\bmkfs\.\w, r:\(\)\{:\|:;\};:, r\bdd\sif.*of/dev/, r\bshutdown\s, r\breboot\s, ]) def _check_dangerous_pattern(command: str) - None: for pattern in DANGEROUS_PATTERNS: if re.search(pattern, command): raise ValueError(f命令包含危险模式, 已被拦截: {command})这套黑名单当然存在绕过空间所以它只能作为最后一道兜底防线不能当成唯一的依赖。在安全策略上真正值得投入的是审计日志和审批机制也就是下文要讲的。5.3 审计日志每一步操作都要留痕AI 操作服务器的每一件事都应该能追溯。我单独建了一个 audit 日志每个工具被调用时都会写入一条结构化记录import logging from logging.handlers import RotatingFileHandler audit_logger logging.getLogger(ssh_mcp.audit) audit_logger.setLevel(logging.INFO) handler RotatingFileHandler(logs/audit.log, maxBytes5 * 1024 * 1024, backupCount10) handler.setFormatter(logging.Formatter(%(asctime)s|%(levelname)s|%(message)s)) audit_logger.addHandler(handler) def audit(tool_name: str, **kwargs): audit_logger.info(%s|%s, tool_name, json.dumps(kwargs, ensure_asciiFalse))然后在每个工具函数里调用一次audit(execute_command, aliasalias, commandcommand)。这样一旦出问题你能清楚看到 AI 在哪个时间点调用了什么命令完整还原现场。日志文件用 RotatingFileHandler 滚动避免长时间运行撑爆磁盘。这个习惯所有做 AI Agent 相关工作的朋友都该从第一天就养成。6. 客户端接入与本地联调测试6.1 在 Trae IDE 中配置自己的 MCP Server我平时主要用 Trae IDE 来测 MCP配置非常直接。打开 IDE 的 MCP 配置入口添加一个新服务器{ mcpServers: { ssh-mcp: { command: python, args: [/absolute/path/to/server.py], env: { PATH: /usr/local/bin:/usr/bin:/bin } } } }这里的command和args对应的是 MCP Server 的启动命令。IDE 会把它作为子进程拉起然后通过 stdio 通信。注意command必须填python的绝对路径或者在 PATH 中能找到的路径如果你用了虚拟环境最好写venv/bin/python的绝对路径避免 IDE 的环境变量和终端不一致导致找不到依赖。这个细节我在不同机器上踩过好几次坑终端里明明能跑IDE 里一启动就报 ModuleNotFoundError八成就是 Python 路径不对。配置保存后如果工具列表自动刷新出来了execute_command、list_dir、upload_file这些工具说明 Server 已经接上了。6.2 不依赖 IDE 的本地联调写一个 MCP 测试客户端开发阶段每次改完代码都去 IDE 里点半天效率很低。我更推荐写一个几十行的本地测试客户端直接验证 MCP Server 逻辑是否正确。用的是 MCP SDK 自带的 stdio 客户端链import asyncio, json from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters( commandvenv/bin/python, args[server.py], env{PATH: /usr/local/bin:/usr/bin:/bin}, ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: tools await session.list_tools() print(可用工具:, [t.name for t in tools]) result await session.call_tool( execute_command, {alias: ops-server, command: uname -a, timeout: 20}, ) for item in result.content: print(输出:, item.text) asyncio.run(main())这个脚本的价值在于跑得极快每次改完 Server 代码几十秒就能完成全流程验证。联调通过后再去 IDE 里面做最终体验效率会高很多。7. 日志与实战调试经验7.1 自定义日志管理的正确打开方式在 MCP Server 里写日志有个细节特别容易让人困惑。FastMCP 底层用的是 Python 标准库 logging如果你直接在模块里logging.getLogger(ssh_mcp)然后info()默认情况下日志是看不到的——因为标准库默认日志级别是 WARNING且没有配置 handler。我在项目里单独做了一个统一的日志初始化模块def setup_logging(): logs_dir logs os.makedirs(logs_dir, exist_okTrue) fmt %(asctime)s [%(levelname)s] %(name)s: %(message)s runtime_handler RotatingFileHandler( os.path.join(logs_dir, runtime.log), maxBytes5 * 1024 * 1024, backupCount5, encodingutf-8 ) runtime_handler.setFormatter(logging.Formatter(fmt)) logging.getLogger().addHandler(runtime_handler) logging.getLogger().setLevel(logging.INFO) return audit_logger然后在server.py启动时调用setup_logging()。这样你能同时看到 MCP 通信过程日志和你的业务日志。遇到“客户端能连接但工具调用失败”这种问题别在 IDE 里干瞪眼直接看runtime.log里最后发生了什么通常一两分钟就能定位。7.2 实战踩坑记录第一个坑就是 SSH host key 校验。AI 使用BatchModeyes时如果目标主机的 host key 不在 known_hosts 里SSH 会直接失败返回Host key verification failed。这个错误特别隐蔽因为你自己在终端里第一次连接时肯定会手动输入 yes 确认但 MCP Server 是无人值守的子进程它没法输入 yes。解决方案是在准备阶段自己先手动ssh-keyscan 主机 ~/.ssh/known_hosts或者在 ssh_config 里对指定主机配置StrictHostKeyChecking accept-new。第二个坑是命令输出太多导致 MCP 通信卡顿。有一次让 AI 查日志AI 直接调用了cat /var/log/nginx/access.log几万行日志瞬间铺满返回内容MCP 客户端直接卡死。我在执行器里加了一个输出截断限制单次返回最多 8000 字符并在末尾补充一句提示“输出过长已被截断建议使用 tail 或 grep”。这是又一个适合做成工具内建习惯的细节。第三个坑是 AI 生成的命令里带了引号。无论用户怎么描述AI 有时还是会传ls -la /home/deploy/www这样的命令。因为 SSH 远端会重新用 shell 解析这条命令字符串带引号的路径在远端语义是正确的本地列表传参不会破坏它所以最终执行没问题。但如果我在参数拼接时用了shlex.join或者手动加引号反而会引入二次解析错误。这里的经验是远端命令以字符串形式传给 ssh等于让远端 shell 来做语义解析本地不要擅自做任何加引号处理。8. 常见问题速查表问题现象可能原因解决方案工具调用报Permission denied (publickey)密钥未加载到 ssh-agent 或密钥路径不对先手动ssh alias确认能连检查IdentityFile配置报Host key verification failedknown_hosts 里没有目标主机指纹ssh-keyscan 主机 ~/.ssh/known_hosts或配置StrictHostKeyChecking accept-new输出被截断单次命令输出量过大增加截断阈值或让 AI 改用 tail / grep 缩小范围scp: -P: option requires an argument混淆了 ssh 和 scp 的参数确认用的是大写-P且不要传给ssh命令命令执行后没有返回任何输出命令本身无 stdout或 stdout 和 stderr 都被吞掉检查是否该命令输出在 stderr确保合并了result.stderr服务启动成功但工具列表为空Python 解释器环境不对或装饰器未生效确认 venv 中的命令路径检查mcp.run()是否在if __name__ __main__内中文文件名或内容乱码SSH 客户端和远端 locale 不一致命令前加export LANGen_US.UTF-8;或配置SendEnv LANGstdout 无内容但 exit code 非 0远端执行了超时或被 timeout 命令杀掉检查日志确认是否有timeout包裹层生效9. 写在最后这个项目从头写到实际跑通我最大的体会是AI 拥有手脚之后的边界感远远比 AI 的能力本身更重要。技术难度几乎没有真正的难点在于你想让它在哪些范围里自由、在哪些范围里受限以及出问题之后如何追溯。白名单机制、路径校验、审计日志这些在很多“极客炫技”类的教程里不会有人认真讲但恰恰是它们决定了这个工具能不能从玩具变成生产力。如果你自己部署了这套方案建议接下来朝两个方向扩展一是把sudo密码的交互处理做成带审批的流程让 AI 在执行高权限命令前先向用户确认二是把审计日志接入团队的告警系统任何异常操作都能实时通知。等这两步做完你的 AI 运维助手才算真正具备了上一线干活的资格。
上一篇/下一篇内容由系统自动关联 返回资讯列表 →