作为一个常年折腾自动化工具的人，我一直在思考一个问题：AI 再聪明，如果只能停留在对话框里输出文字，那它的价值就折损了大半。真正让 AI 从“顾问”变成“执行者”的，是让它能直接操作真实环境——比如连上你的服务器，执行命令、读取日志、修改配置。这篇文章要聊的，就是如何用纯 Python 手写一个原生 SSH MCP Server，让 AI 通过标准化的 MCP 协议安全地接入 SSH，拥有真正意义上的“手脚”。
内容会覆盖三块：MCP 协议的核心通信机制、原生 SSH 客户端的封装思路、以及如何把两者组装成一个可以被 Claude Desktop、Cline、Trae IDE 等任意 MCP 客户端直接调用的服务器。适合已经跑通过 MCP Hello World、但想深入源码级理解，或者对依赖各种重型框架感到厌倦的开发者。
1. 为什么选择原生实现而非 FastMCP 等现成框架
1.1 我对“原生”的理解和对框架的“偏见”
先坦白一个立场：FastMCP、MCP SDK 这些框架确实好用，几行代码就能注册一个工具。但如果你只用框架，你学到的是“怎么调 API”，而不是“协议是怎么设计的”。我个人的习惯是，第一版工具尽量用标准库手写，跑通之后再引入框架优化。好处是：出了问题你知道往哪一层去排查——是 JSON-RPC 层的错误，还是 SSH 执行层的问题，还是 stdio 传输层的解析错误。
原生实现还有一个实际收益：依赖极少。在这个项目里，除了paramiko这一个 SSH 库之外，其余全部用 Python 标准库完成。这意味着你可以在任何装有 Python 3.10+ 的机器上直接运行，不挑虚拟环境，不挑操作系统，也不容易被依赖冲突折磨。
1.2 MCP 协议的核心拆分：传输层与协议层
MCP（Model Context Protocol）本质上是一个基于 JSON-RPC 2.0 的请求-响应协议。整个通信可以拆成两层理解：
传输层（Transport Layer）：负责把 JSON 消息从一端搬到另一端。官方支持 stdio（标准输入输出管道）和 SSE（Server-Sent Events）两种模式，本项目用 stdio。原因是它最简单、最安全——客户端启动服务器子进程，然后通过管道读写消息，不需要开端口，不需要处理 CORS。
协议层（Protocol Layer）：定义消息的格式。每条消息是一个 JSON 对象，包含jsonrpc、id、method、params等字段。你不需要实现全部 MCP 方法，只需实现客户端会调用的核心几个：initialize（握手）、tools/list（上报工具清单）、tools/call（执行工具调用）。
用生活类比来解释：stdio 是水管，JSON-RPC 是水，MCP 协议决定了水往哪流。我这个项目做的就是自己铺设水管，并且实现水流的阀门逻辑。
2. SSH 工具调用的核心设计：从命令拼接到执行回传
2.1 安全地构造远程命令：不要“字符串拼接”
SSH MCP Server 最危险的操作就是执行远程命令。如果你允许 AI 传入任意 shell 命令并直接拼接到ssh user@host "command"里，那基本等于给了 AI 一把万能钥匙——它会执行rm -rf还是读取/etc/passwd，完全取决于它的幻觉有多严重。
我的做法是：预先定义白名单指令集。server 启动时从配置文件加载允许执行的命令模板，例如list_files、read_file、run_command三类。其中run_command只允许执行用户明确指定的少数命令（如df -h、free -m、uptime），并且通过shlex模块进行参数解析和转义，杜绝 shell 注入。
2.2 关键点：SSH 连接复用与会话保持
每次调用都新建 SSH 连接是不现实的——握手要花时间，而且会导致大量的 TIME_WAIT 端口堆积。我在 server 内部做了一个连接池：首次调用时建立 SSHClient 连接，之后复用同一个 Transport。为了处理连接断开的情况，每次执行命令前先执行一个exec_command('echo ok')探活，失败则重连。
还有一个细节是Channel 的超时控制。AI 可能会发出一个永远执行不完的命令（比如ping google.com），这会卡住整个工具调用。我的解决策略是给channel.settimeout设置默认 10 秒，超出时间会抛出socket.timeout，server 捕获后向客户端返回结构化错误，而不是让请求一直挂起。
3. 手写 JSON-RPC 调度核心：initialize、tools/list、tools/call 的实现
3.1 消息循环：从 stdin 读取，按 id 分发
原生实现的骨架是一个无限循环：从sys.stdin.buffer.readline()读取一行 JSON，解析后根据method分发到对应处理函数，处理结果通过sys.stdout.write()写回。注意必须用write + flush，不能依赖 print 的默认缓冲，否则客户端会一直等不到响应。
这里直接给出核心代码结构：
import sys import json import shlex class McpSshServer: def __init__(self, ssh_config: dict): self.ssh_config = ssh_config self.client = None self._ensure_conn() def _ensure_conn(self): if self.client is not None: try: self.client.exec_command("echo ok", timeout=5) return except Exception: self.client.close() self.client = _create_ssh_client(self.ssh_config) def handle_message(self, line: str) -> dict | None: msg = json.loads(line) method = msg.get("method") msg_id = msg.get("id") if method == "initialize": return self._handle_initialize(msg_id, msg.get("params", {})) elif method == "tools/list": return self._handle_list_tools(msg_id) elif method == "tools/call": return self._handle_call_tool(msg_id, msg.get("params", {})) elif method == "notifications/initialized": return None # 通知类消息不需要响应 else: return { "jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Method not found: {method}"} }你可能会问：为什么tools/call的返回要这样包一层content数组？因为 MCP 协议规定工具调用结果使用content数组 +is_error标志位的方式表达。即使你的工具本质上是文本输出，也必须遵守协议规定的“信封”格式。
3.2 工具注册表：让 AI 知道它能“碰”什么
tools/list返回的是一个 JSON 数组，每个元素描述一个工具的名称、描述和输入参数 schema。这块值得花心思打磨：
- 描述要写清楚边界：比如
run_command的 description 里要写明“仅支持系统状态类命令，禁止写操作”，这能有效降低 AI 误用工具的概率。 - 参数 schema 要严格：
required字段要明确，枚举值要列出。AI 在调用工具前会读取这个 schema 来规划参数，你写得越清晰，它越不会乱来。
举个例子，文件读取工具的参数定义：
{ "name": "read_file", "description": "读取远程服务器上的文本文件（禁止读取敏感文件）", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "文件的绝对路径"} }, "required": ["path"] } }4. SSH 连接层的工程化封装：paramiko 的高级用法
4.1 为什么要选 paramiko 而不是直接 subprocess 调系统 ssh 命令
第一版我确实是用subprocess去调用本机的ssh和scp命令，但很快就放弃了。原因有三个：
- 密钥管理混乱：ssh 命令依赖
~/.ssh下的密钥文件和 known_hosts，多主机切换时很容易串。 - 输出解析困难：命令混合了 stderr、stdout、交互式提示符，不好区分。
- 超时控制脆弱：ssh 进程可能挂死，需要额外写 watchdog。
paramiko 是一个纯 Python 实现的 SSHv2 协议库，不需要系统安装 OpenSSH 客户端就能工作。它提供了exec_command、open_sftp等高层 API，让我能把连接管理、命令执行、错误处理全部统一在一个进程内。
连接参数配置示例：
import paramiko def _create_ssh_client(cfg: dict) -> paramiko.SSHClient: client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostname=cfg["host"], port=cfg.get("port", 22), username=cfg["user"], password=cfg.get("password"), key_filename=cfg.get("key_filename"), timeout=cfg.get("timeout", 10), allow_agent=False, look_for_keys=False, ) return client关键细节在最后两个参数：allow_agent=False和look_for_keys=False。这是为了确保不会意外加载本机的 SSH agent 或默认密钥，避免在服务器上把当前用户的凭据暴露给 AI。
4.2 执行命令并“结构化”返回输出
paramiko 的exec_command返回三个文件对象：stdin、stdout、stderr。注意一个坑：如果你先读 stdout 再读 stderr，可能会死锁。因为 channel buffer 是共享的，正确做法是用recv_exit_status配合并发读取，或者简单粗暴地先读满两个流再收状态。
以下是完善的执行函数：
def run_remote_command(self, command: str, timeout: int = 10) -> dict: self._ensure_conn() stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout) out = stdout.read().decode("utf-8", errors="replace") err = stderr.read().decode("utf-8", errors="replace") status = stdout.channel.recv_exit_status() return { "exit_code": status, "stdout": out, "stderr": err, }errors="replace"很重要。远程服务器可能输出非 UTF-8 内容（比如 locale 是 GBK），直接.decode()会抛异常，用 errors 参数能保证不会因为一个乱码就中断整个调用。
5. 组装与手工测试：在没有客户端的情况下验证 server 逻辑
5.1 用echo模拟 JSON-RPC 请求，测试握手全流程
写完代码，第一件事不是接入 AI 客户端，而是先用文本 JSON 模拟请求，验证 server 的响应结构是否符合 MCP 规范。测试方式如下：
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"manual-test","version":"1.0"}}}在 bash 里执行echo '{...}' | python ssh_mcp_server.py，观察返回结果。我第一次跑的时候就在这一步发现了一个 bug：initialize 响应里漏了protocolVersion字段，导致部分客户端直接拒绝握手。这种问题如果不手测，等到接客户端时排查会非常痛苦。
5.2 在真实客户端里的完整效果
以 Claude Desktop 或 Cline 为例，配置一个 MCP server 只需在 JSON 配置里指定启动命令：
{ "mcpServers": { "ssh-server": { "command": "python", "args": ["/path/to/ssh_mcp_server.py"], "env": { "SSH_HOST": "your-server-ip", "SSH_USER": "root", "SSH_PASSWORD": "your-password" } } } }启动后，AI 在对话中就能调用list_files("/etc")、read_file("/etc/hostname")这类操作。实测下来，Claude 能准确根据我的自然语言指令选择正确的工具，并且会主动询问权限边界。
我建议你在配置里允许 AI 执行的第一批命令是：df -h（查看磁盘），free -m（查看内存），uptime（查看负载）。这些命令没有破坏性，又能让 AI 真正“有用”，体验感极佳。
6. 避开这三个坑，你才算是真正跑通了 SSH MCP Server
6.1 坑一：stdout 被非 JSON 输出污染
如果服务器在运行过程中不小心 print 了一行调试信息到 stdout，那整条消息流就断了。因为客户端在按行解析 JSON 时，遇到非 JSON 字符串会直接报错。
规避方案：所有调试日志一律走sys.stderr或logging模块，绝不能往 stdout 写。我在代码里封了一个_log()方法，统一把日期、级别、消息写到 stderr，这才避免了“线上静默崩溃、找不到原因”的尴尬。
6.2 坑二：AI 幻觉出参引发 SSH 异常连锁
AI 可能生成一个不存在的路径参数，也可能生成一个跨平台不一致的路径（比如 Windows 风格的C:），这在 Linux 服务器上会直接报错。我的处理方式是在工具层面把异常包起来，所有 SSH 执行错误统一返回is_error: true的 MCP 响应，而不是让异常抛出到主循环导致 server 崩溃。
6.3 坑三：长期保持的回话连接被防火墙切断
SSH 连接如果长时间闲置，会被 NAT 网关或服务端的ClientAliveInterval掐断。要解决这个问题，paramiko 的Transport可以设置 keepalive：
self.client.get_transport().set_keepalive(30)这样每 30 秒会发送一个 keepalive 包，保证连接长期活跃。我在实测中让 server 挂机一夜，第二天仍然能正常执行命令，靠的就是这个参数。
7. 把工具变成真正的“助手”：使用场景演进与进阶排查思路
7.1 从“查询”到“操作”的边界控制
当你确认基础版稳定之后，可以做几个进阶扩展：
- SFTP 文件传输:paramiko 的
open_sftp可以读取、上传、下载文件，扩展之后 AI 就能做日志备份。 - 命令回放审计:在
run_command里把每次执行的命令和结果追加到本地日志文件，方便回溯 AI 的行为。 - 多服务器配置:配置文件支持多个 host 的数组，用环境变量切换当前目标，实现“一个服务、多台机器”。
7.2 如果接上不，怎么快速定位是哪一层的问题
遇到“AI 无法连接 server”这类问题，按这个顺序排查：
- 启动阶段:直接在终端运行
python ssh_mcp_server.py，看有没有异常退出。注意 stdout 有没有非 JSON 输出。 - 握手阶段:用前面提到的手工 JSON 模拟 initialize，看响应包结构。
- 工具发现阶段:模拟
tools/list，确认返回的工具 schema 格式正确。 - 执行阶段:模拟
tools/call，传入固定参数，观察 paramiko 是否连接成功、命令是否执行、响应是否封装正确。
这四层如果都通了，那剩下只可能是客户端配置格式的问题。这个排查思路适用于任何 MCP server 开发，不只是 SSH 类工具。
7.3 最后的进阶：从纯执行到“有理解力”
等 core 功能稳定后，我给 server 加了一层轻量级“意图路由”：AI 传入自然语言指令，server 本地用正则和关键词匹配，映射到不同的安全工具上。
比如用户（也就是 AI 客户端）传入：“看看服务器磁盘是不是满了”，server 会优先匹配到df -h工具，而不是把这句话直接当作 shell 命令执行。这种思路能有效减少“幻觉命令”带来的不可控风险，还能让没有接受过工具调用训练的模型也能稳妥使用。
我实际使用下来最深的体会是：让 AI 拥有“手脚”是一回事，让它安全地拥有“手脚”又是另一回事。这套原生实现的 SSH MCP Server 让我能在完全不依赖重量级框架的情况下，看清 MCP 的每一字节流转，也让我在接入不同客户端（Claude Desktop、Cline、Trae IDE）时，能自信地告诉别人：“如果它不能跑，那一定是配置的问题，而不是协议的问题。” 最后再分享一个技巧：每次改动代码后，先用那套手工 JSON 模拟流程跑一遍全部工具，五分钟以内就能确认有没有回归——这份时间绝对花得值。
