说实话第一次听到“MCP Server”这个词时我心里是有点不以为然的——这不就是给 AI 开个接口调工具吗后来真正自己动手搓了一个才发现这里面的门道比想象中多得多。项目标题说“让 AI 拥有手脚”这个比喻很准确大模型再聪明它也只能“说”不能“做”想让它真正帮你把服务器上的命令执行了、文件改了就需要一条让它“伸手”去操作的通道。而 SSH MCP Server就是给 AI 装上的那条“机械臂”。这篇文章我想把我从协议拆解、Python 原生实现到最终接通 AI 客户端的完整过程复盘一遍把那些文档里不写、但实际必踩的坑都摆出来给打算自己动手的读者一条能直接走通的路。1. 为什么偏要“手搓”一个 SSH MCP Server1.1 AI 缺的不是“脑子”是“手脚”接触过 AI Agent 的朋友应该都有体会模型本身的能力再强回答得再漂亮一旦涉及实际操作就歇菜。你问它“这个目录为什么满了”它能头头是道告诉你用du查、用df看但如果你不亲自复制粘贴命令到终端它什么都改变不了。MCPModel Context Protocol要解决的就是这个“最后一公里”。它把 AI 应用和外部工具、数据源之间的连接方式标准化了让 AI 能主动发起工具调用而不是只会输出文本。协议本身由 Anthropic 在 2024 年底开源后来迅速成了 AI 工具调用的事实标准OpenAI、Google 等也陆续兼容。打个比方如果说大模型是大脑插件市场是工具箱那 MCP 就是“神经系统”——大脑发出指令神经系统把指令翻译成具体的动作驱动手脚去执行。1.2 为什么业务场景偏偏选中 SSH市面上的 MCP Server 五花八门有读写文件的、有查数据库的、有操作浏览器的。但我个人最刚需的其实是 SSH——因为日常工作中大量场景都依赖远程服务器查日志、看负载、改配置、批量分发脚本、定时任务排障。这些东西全都跑在 Linux 服务器上而连接它们的标准通道就是 SSH。选 SSH 还有一个现实优势它是运维领域的事实协议生态成熟、可控性强、天然支持加密认证。给 AI 接上一个 SSH 工具相当于让 AI 具备了对远程主机的“操作权”这比单纯读本地文件有用得多。而且这个场景足够通用——不管你是运维、后端开发还是算法工程师只要服务器还存在于你的工作流里这个工具就有价值。1.3 原生实现的价值在哪为什么不直接拿现成的 SDK 跑官方确实提供了 Python SDK用起来很方便几行代码就能跑起一个带工具的服务。但问题在于SDK 把太多协议细节封装掉了一旦遇到协议兼容问题、客户端版本差异、传输层异常你会完全无从下手。我选择“手搓原生”目的就是把 MCP 的通信机制彻底搞明白消息怎么封包、请求怎么路由、工具怎么注册、错误怎么返回。这个过程就像学开车先拆一遍变速箱——费点时间但之后不管遇到什么兼容问题你都能从原理层去排查。而且实际上手写一个满足基本需求的 MCP Server 并没有想象中复杂核心代码量不到三百行这性价比非常高。2. 动手前的协议拆解与准备2.1 需要准备的环境和库开始写代码之前先把环境理清楚。我用的 Python 版本是 3.10推荐至少 3.9 以上因为后面参数校验和类型注解会用到新特性。依赖版本建议用途Python3.10运行环境paramiko2.12SSH 连接与命令执行pyyaml6.x可选的配置解析为什么选 paramiko 而不是直接调ssh命令行因为 paramiko 是 Python 生态里最成熟的 SSH 协议实现它直接封装了 SSH 的传输层和认证层你不需要管理子进程、解析终端输出代码可控性高很多。另外在 Windows 上直接调ssh.exe会遇到换行符和回显的问题用 paramiko 可以绕开这些麻烦。2.2 MCP 协议核心机制回顾MCP 基于 JSON-RPC 2.0传输层常用两种方式stdio标准输入输出和 Streamable HTTP。手写原生实现时我选的是 stdio——因为大多数 AI 客户端比如 Claude Desktop都是通过启动一个子进程来与 MCP Server 通信的stdio 方式免去了开端口、配防火墙这些额外工作。整个通信流程是这样的客户端发送initialize请求包含协议版本和客户端能力描述服务端返回协议版本、服务端能力和服务信息客户端发送notifications/initialized通知表示初始化完成客户端调用tools/list获取可用工具列表客户端调用tools/call触发具体工具执行服务端返回结构化结果在 stdio 模式下消息的封包格式和 LSPLanguage Server Protocol一脉相承先发送一段 Header格式是Content-Length: 长度\r\n\r\n紧接着是 JSON 序列化的消息体。这个细节非常关键官网文档很少强调但实际调试中你遇到的 80% 问题都出在消息封包上。2.3 为什么核心功能定为“命令 文件”在设计工具集的时候我没有一上来就追求功能大而全而是严格遵循“够用、可控、可审计”三个原则。最终只做了三个工具exec_command在远程服务器上执行 shell 命令返回标准输出和标准错误read_file读取远程文件内容支持指定路径和行数范围write_file写入或追加内容到远程文件支持备份和权限设置三件套乍看简单但覆盖了日常运维 80% 的高频操作查状态用第一个看配置用第二个改配置用第三个。更重要的是工具数量越少AI 的“误操作面”就越小这在后面聊安全的时候会细说。同时每个工具的参数设计我都刻意做成显式声明而不是让 AI 自由拼接 shell 命令——这样能在入口处先挡掉一部分乱来。3. 手写 MCP Server 的核心实现3.1 项目结构与消息编码层我习惯把一个完整的小项目拆成清晰的模块既方便自己维护也方便读者看懂。整个项目的文件结构如下ssh_mcp_server/ ├── server.py # MCP 协议主循环与消息处理 ├── ssh_client.py # paramiko 连接池与远程执行 ├── tools.py # 工具定义与参数校验 ├── config.yaml # SSH 目标主机与安全配置 └── requirements.txt # 依赖清单先写最底层、最容易出错的消息封包层。MCP 在 stdio 模式下每一帧消息都必须严格遵循“Header Body”的格式# server.py 片段 —— 消息读取与写入 import sys import json import re def read_message(): 从标准输入读取一帧 MCP 消息 content_length None # 循环读取 Header 部分 while True: line sys.stdin.buffer.readline() if not line: return None if line in (b\r\n, b\n): break match re.match(rbContent-Length: (\d), line) if match: content_length int(match.group(1)) if content_length is None: raise ValueError(Missing Content-Length header) payload sys.stdin.buffer.read(content_length) return json.loads(payload) def write_message(message: dict): 向标准输出写入一帧 MCP 消息 body json.dumps(message, ensure_asciiFalse).encode(utf-8) header fContent-Length: {len(body)}\r\n\r\n.encode(utf-8) sys.stdout.buffer.write(header body) sys.stdout.buffer.flush()这段代码是整个服务的基石。我最初犯过一个低级错误直接在 handler 函数里用print()输出调试信息结果客户端一直报解析失败调试了半天才发现 stdout 是协议通道任何多余的输出都会把消息包撑破。后来我所有日志全部改走 stderr 或文件问题才消失。3.2 SSH 连接封装复用与加固接下来写 paramiko 的封装层。这里的核心设计决策是“连接复用”——不能让每次工具调用都新建一次 SSH 连接因为握手和认证的开销非常大实测一次完整的 SSH 连接建立需要 1-2 秒根本扛不住 AI 连续调用几个命令的节奏。# ssh_client.py —— 连接管理与命令执行 import paramiko class SSHConnection: def __init__(self, host, port, username, key_path, timeout10): self.host host self.port port self.username username self.key_path key_path self.timeout timeout self.client None def connect(self): if self.client and self.client.get_transport() and self.client.get_transport().is_active(): return self.client self.client paramiko.SSHClient() self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.client.connect( hostnameself.host, portself.port, usernameself.username, key_filenameself.key_path, timeoutself.timeout, banner_timeoutself.timeout ) return self.client def exec(self, command, inner_timeout30): client self.connect() # 用 timeout 命令包裹防止某些命令长期卡死 wrapped ftimeout {inner_timeout} bash -lc {command!r} stdin, stdout, stderr client.exec_command(wrapped) exit_code stdout.channel.recv_exit_status() out stdout.read().decode(utf-8, errorsreplace) err stderr.read().decode(utf-8, errorsreplace) return exit_code, out, err def close(self): if self.client: self.client.close()有两个容易被忽略但实际影响很大的细节exec_command本身没有全局超时概念它默认会一直等下去。我选择用 Linux 的timeout命令包裹这样即使内部命令挂起外层也能强行终止避免 client 端一直等待。我用了bash -lc而不是直接执行命令。这是因为 paramiko 默认是非交互式的 shellPATH 环境变量不完整很多装在/usr/local/bin或用户目录下的命令根本找不着。用bash -lc可以加载用户的登录环境大幅减少“command not found”的尴尬。3.3 协议路由initialize 与 notifications完成了传输层接下来要处理协议层也就是对initialize、notifications/initialized、tools/list、tools/call四大核心方法的响应。# server.py 片段 —— 核心路由 def handle_initialize(req_id, params): return { jsonrpc: 2.0, id: req_id, result: { protocolVersion: 2025-06-18, capabilities: {tools: {listChanged: False}}, serverInfo: {name: ssh-mcp-server, version: 1.0.0} } } def handle_tools_list(req_id): from tools import TOOL_SCHEMAS return { jsonrpc: 2.0, id: req_id, result: {tools: TOOL_SCHEMAS} } def handle_tools_call(req_id, params): name params.get(name, ) arguments params.get(arguments, {}) from tools import dispatch_tool result dispatch_tool(name, arguments) if error in result: return { jsonrpc: 2.0, id: req_id, result: {content: [{type: text, text: result[error]}], isError: True} } return { jsonrpc: 2.0, id: req_id, result: { content: [{type: text, text: result.get(output, )}], isError: False } }initialize里有个容易踩的坑protocolVersion字段必须返回一个客户端认识的值。如果返回一个过旧或过新的版本号有些严格校验的客户端会直接拒绝继续连接。我一开始用了无参数的2024-11-05Claude Desktop 一直连接不上后来换成2025-06-18才通。建议手写的时候先在官方 SDK 源码里查一下当前协议版本号。另外initialize返回的capabilities要如实声明。如果你没有实现resources/list和prompts/list就不要在 capabilities 里声称支持否则客户端会把你的能力探测一遍然后发现一堆不支持的方法行为就变得不可预测了。3.4 主循环把整个服务跑起来有了前面三块主循环其实就是个简单的“读消息—分发给 handler—写响应”的无限循环。# server.py 主体 def main(): connection None # 延迟初始化 SSH 连接 while True: message read_message() if message is None: break method message.get(method) req_id message.get(id) params message.get(params, {}) if method initialize: write_message(handle_initialize(req_id, params)) elif method notifications/initialized: # 通知类消息没有 id不需要返回响应 continue elif method tools/list: write_message(handle_tools_list(req_id)) elif method tools/call: if connection is None: from ssh_client import SSHConnection connection SSHConnection( hostCONFIG[host], portCONFIG[port], usernameCONFIG[username], key_pathCONFIG[key_path] ) # 把 connection 注入 dispatch from tools import set_connection set_connection(connection) write_message(handle_tools_call(req_id, params)) elif method ping: write_message({jsonrpc: 2.0, id: req_id, result: {}}) else: if req_id is not None: write_message({ jsonrpc: 2.0, id: req_id, error: {code: -32601, message: fMethod not found: {method}} }) if __name__ __main__: main()启动方式很简单python server.py不过单独跑起来不会有任何输出因为它在等 stdin 的消息。真正的“启动”动作发生在你把它注册到 AI 客户端的时候客户端会以子进程的方式拉起这个脚本然后开始协议握手。这也是我第一次接触 stdio 模式时最不适应的一点——它的运行模式更像一个被外部驱动的“服务函数”而不是一个独立开端口等待连接的常驻进程。3.5 工具的注册与参数约束工具注册是协议层和业务层的交界点。每个工具需要提供name、description和inputSchema。inputSchema是 JSON Schema 格式AI 会根据这个 Schema 来生成调用参数所以工具描述写得越清晰AI 的调用准确率就越高。# tools.py 片段 —— 工具定义 TOOL_SCHEMAS [ { name: exec_command, description: 在远程服务器上执行指定的 shell 命令返回标准输出、标准错误与退出码。适合查看日志、检查状态、运行诊断命令。, inputSchema: { type: object, properties: { command: { type: string, description: 要执行的完整 shell 命令例如 ls -lh /tmp }, workdir: { type: string, description: 执行命令的工作目录默认使用用户主目录 } }, required: [command] } } ] def dispatch_tool(name, arguments): if name exec_command: command arguments.get(command, ) # 安全校验 blocked [rm -rf /, mkfs, dd if, /dev/sda] for pattern in blocked: if pattern in command: return {error: fCommand blocked by security policy: {pattern}} code, out, err connection.exec(command) return {output: fExit code: {code}\nSTDOUT:\n{out}\nSTDERR:\n{err}}这里我故意加了一个最粗糙的安全过滤把高危命令直接挡在门外。当然这种字符串匹配方式远远不够但作为第一道防线至少能拦住 AI 因为“理解错意图”而误执行的破坏性操作。后面还会提到更完善的方案。4. 与 AI 客户端联调让大模型真正驱动 SSH4.1 客户端配置我日常主力客户端是 Claude Desktop也支持 Claude Code 和兼容 MCP 的第三方客户端它的配置方式是修改客户端自己的配置文件把 MCP Server 作为“子进程应用”注册进去。以 macOS 为例Claude Desktop 的配置文件路径是~/Library/Application Support/Claude/claude_desktop_config.jsonWindows 对应路径为%APPDATA%\Claude\claude_desktop_config.json配置内容如下{ mcpServers: { ssh-mcp: { command: /usr/bin/python3, args: [ /Users/me/ssh_mcp_server/server.py ], env: { SSH_CONFIG_FILE: /Users/me/ssh_mcp_server/config.yaml } } } }注意几个细节。第一command尽量写 Python 解释器的绝对路径避免客户端子进程 PATH 不完整导致找不到解释器。第二如果把 SSH 密钥路径写进代码里后续换机器很麻烦更合理的做法是放到config.yaml里通过环境变量指定配置文件位置。第三配置文件是 JSON 格式不要手滑多打逗号。4.2 真实调用场景从对话到执行配置完成后重启客户端就能在工具列表里看到ssh-mcp的注册信息了。我实际测试了几个典型场景场景一查看远程服务器的磁盘占用用户帮我看看那台测试机上 /tmp 目录被哪些文件占满了。大模型会调用exec_command(commanddu -sh /tmp/* | sort -rh | head -20)工具返回结果后大模型再把输出翻译成自然语言“占用最大的是/tmp/core.dump达到 3.2GB建议确认后可删除。”场景二排查一个进程的启动失败问题用户帮我查看 nginx 服务为什么启动失败。大模型会依次调用exec_command(commandsystemctl status nginx --no-pager) exec_command(commandjournalctl -u nginx --no-pager -n 50) exec_command(commandnginx -t)三个命令串联下来基本能把问题定位清楚。这就是“手脚”的价值——它不只是一个命令执行器而是能跟据实际情况自主判断下一步做什么形成一个小型的运维 Agent 闭环。4.3 联调中常见的协议层报错联调过程不会一帆风顺。我遇到过最典型的报错是客户端提示Error: not initialized。原因在于客户端启动后MCP Server 进程里要能正确响应initialize请求。如果服务端代码里在读取消息时抛了异常比如 JSON 解析失败初始化请求就会因为缺少响应而超时客户端就会放弃连接。我的排查手记先手动向 server 进程发一个initialize请求看它能不能正确返回。如果返回不正常检查 server 进程的 stderr 日志——我特意把内部错误全部traceback输出到 stderr 文件方便定位。日志正常但客户端还是失败多半是protocolVersion不匹配换一个客户端认可的版本号。最稳妥的调试方法写一个小的模拟客户端直接用 Python 调用这个 server这样协议问题一目了然。下面这段脚本是我常用的“冒烟测试”python - EOF import subprocess, json, os proc subprocess.Popen( [python, server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE ) def send(obj): payload json.dumps(obj).encode() proc.stdin.write(fContent-Length: {len(payload)}\r\n\r\n.encode() payload) proc.stdin.flush() send({jsonrpc:2.0,id:1,method:initialize,params:{}}) # 读取一行响应 import re length 0 while True: line proc.stdout.readline() if line in (b\r\n, b\n): break m re.match(rbContent-Length: (\d), line) if m: length int(m.group(1)) resp json.loads(proc.stdout.read(length)) print(json.dumps(resp, indent2, ensure_asciiFalse)) proc.terminate() EOF如果能正常打印出protocolVersion和serverInfo说明消息层没问题可以放心去连客户端。5. 安全边界、异常处理与性能优化5.1 权限控制不能只靠“提示词”给 AI 接上 SSH 之后安全问题就不再是理论上的了——你等于给一个“可能理解错意图”的系统发了一把远程服务器的钥匙。我的应对思路是三层防线连接层不要用 root 账号直接连。我专门创建了一个受限的运维账号只有必要目录的读写权限不能 sudo。这样即使 AI 发了危险的命令它能造成的破坏也是有限的。工具层在高危命令黑名单的基础上增加“白名单模式”——通过配置文件指定只允许执行的命令前缀AI 想执行任何不在此列的命令都会收到错误提示。这比黑名单安全得多因为黑名单永远不可能列全。审计层所有工具调用都记录详细日志包括用户对话的内容、执行的命令、返回的结果、耗时全部写到独立的日志文件里。一旦出事可以追溯到具体是哪一轮对话触发了什么操作。# config.yaml 中的安全配置片段 security: # 白名单模式只允许以这些命令开头的执行 allowed_commands: - ls - df - du - ps - top - systemctl status - journalctl - cat - tail - head - grep blocked_commands: - rm -rf / - mkfs - dd if allow_write_file: - /tmp/** - /home/ops/**仔细看这个配置读操作放得很开写操作只限定在/tmp和用户主目录。这样 AI 可以自由地查状态、看日志但要真的改系统配置文件会被工具层直接拦截。要调整生产环境的配置我会手动执行不会把这个权限轻易交给 AI。5.2 超时、编码与输出大小大模型的“手”要稳除了安全工程上的稳定性也直接决定体验。我总结了三个必须处理的点。第一是超时。AI 提问后通常有等待时间阈值比如 60 秒如果工具调用耗时太长客户端就会判定超时并终止会话。我在前面用timeout命令包裹外部命令就是这个原因。但致命的命令是ping、traceroute这类会自循环很长一段时间的所以我把工具层的超时上限也限制了——超过 90 秒直接返回错误信息不让 AI 一直傻等。第二是编码。很多服务器上的默认 locale 是C或POSIX如果命令输出包含中文文件名或 UTF-8 字符paramiko 在 decode 的时候可能会报错。我在exec方法里特意用了errorsreplace虽然会替换个别无法解析的字符为?但能保证进程不会因编码异常崩溃。更好的方案是在命令开头加export LANGen_US.UTF-8但并非所有系统都有对应 locale这个方法更通用。第三是输出长度。某些命令比如cat /var/log/xxx.log可能会返回几十 MB 内容这会直接撑爆 MCP 消息的响应体有些客户端会拒绝接收超大消息。我的策略是对输出做截断超过 32KB 直接截掉并附加一句话说明“输出过长仅显示前 32KB”。从实际使用来看绝大多数排障场景根本不需要完整日志截断后的开头部分反而更聚焦问题。5.3 连接管理与并发控制paramiko 的连接本身不是线程安全的如果多个请求同时复用同一个 SSH 连接可能出现数据错乱。虽然 MCP 客户端在 stdio 模式下默认是串行请求但严谨起见我加了全局锁。# server.py 中增加锁 import threading SSH_LOCK threading.Lock() def handle_tools_call(req_id, params): with SSH_LOCK: # 串行化对 SSE 连接的访问 result dispatch_tool(name, arguments)另外我实现了“空闲断连”机制如果 AI 会话结束SSH 连接会继续保持一段时间默认 5 分钟。下次用户再问问题时可以快速复用不用重新握手。实现很简单就是每次调用时更新last_used_time后台线程定期检查是否超时并关闭连接。这里有个性能对比数据值得分享首次连接需要约 1.2 秒而连接复用后的每次命令执行耗时基本稳定在 80~150ms取决于远程命令本身。这个差距在日常使用中非常明显——AI 连续查 10 条命令时复用连接几乎是“秒出”不复用的话光握手就要等十几秒。6. 常见问题与排查速查表手写方案虽然灵活但踩坑是免不了的。我把这几周联调中遇到的问题整理成了一张速查表方便大家直接照方抓药。现象可能原因排查方向 / 解决方案客户端报 “not initialized”initialize 未正确处理用冒烟测试脚本直接发 initialize 看返回检查协议版本号命令执行报 command not foundnon-interactive shell PATH 不完整改用bash -lc包裹命令加载用户登录环境客户端长时间无响应远程命令卡死检查是否用了 timeout 包裹看服务端 stderr 日志中文输出乱码服务器 locale 不支持 UTF-8decode 时用errorsreplace命令前置export LANGC.UTF-8超大输出导致客户端连接断开响应体超过客户端的消息大小限制对输出做 32KB 截断并附加提示SSH 连接频繁断开连接空闲被服务端踢掉增加 ClientAliveInterval 参数实现空闲重连机制工具调用时能执行多重命令AI 把多个命令拼接在中白名单校验需要落在“首条命令”上而非整个字符串Windows 上 paramiko 找不到密钥路径分隔符或权限问题用~展开后打印绝对路径给私钥文件加合适权限这里特别想展开说一下“白名单校验落在首条命令”这个问题。很多初级实现只简单判断command字符串是否以ls开头但 AI 可能拼接出ls /tmp rm -rf /。所以我在校验地时候会用shlex.split先拆出第一条完整命令匹配白名单通过后才放行。虽然不能解决所有注入问题但至少能拦住大多数“拼接式”的危险操作。另一个许多人都踩过的坑是 Windows 下运行这个 Server。paramiko 在 Windows 上读取私钥时要求文件权限不能太开放。Git Bash 生成的密钥文件默认权限是 600问题不大但如果用的是通过某些工具解压出来的密钥权限可能是 644paramiko 会直接报错。解决办法是确保私钥文件只对当前用户可读。小结我的几点体会写这个项目的过程中一个让我反复琢磨的细节是AI 工具调用的“度”到底在哪。我一开始把工具设计得大开大合什么命令都敢放结果测试时 AI 差点把一个临时目录整个删掉——虽然毫无恶意但模型对命令后果的理解确实没有人类那么有“常识”。后来把白名单、目录限制、只读优先这些约束加上整个系统才真正“能用”。我个人体会给 AI 装手脚最难的不是让协议跑通而是让这套工具在“足够自由”和“足够安全”之间找到平衡点。一个能干活但不会乱干活的 MCP Server比一个功能全但需要时刻盯着的 MCP Server 有价值得多。如果你只是想在本地测试一下这个思路可以直接用配置文件里的精简版工具集跑起来试试。但假如要让 AI 主动操作生产环境里的服务器我会非常不建议一步到位先把权限收敛到只有只读查询跑一两周确认没有异常再逐步放开写操作和命令白名单。最后再分享一个小技巧MCP 协议的tools/list返回里每个工具的description字段其实是可以“调教”模型的——你把它写得更具体、更强调参数边界AI 的调用成功率会明显提升。这算是我在反复调试里发现的一个低成本优化手段值得一试。