01MCP 是什么
一句话:MCP 是「AI 应用连接外部世界的 USB-C 接口」。 它定义了一套统一的协议:模型(Host)怎么发现工具、怎么调用工具、 工具怎么返回结果。任何实现了这套协议的 server,都可以被任何支持 MCP 的客户端 直接使用,不用为每个客户端单独写集成。
核心角色
| 角色 | 是谁 | 干什么 |
|---|---|---|
| Host(宿主) | Claude Desktop、Cursor、VS Code、Hermes…… | 持有模型,向用户展示工具,替模型调用工具 |
| Client(客户端) | 宿主内部的协议组件 | 与 MCP server 建立连接、握手、发现工具、发调用请求 |
| Server(服务器) | 你写的工具进程,如文件系统、数据库、服务器运维 | 声明「我能干什么」,执行调用,返回结构化结果 |
一次调用的完整旅程
用户提问「帮我看下服务器磁盘」
│
▼
Host(Claude Desktop / Hermes / Cursor …)
│ initialize 握手 + list_tools 发现工具
▼
MCP Client ──stdio 或 HTTP──▶ MCP Server(你的工具进程)
│ 执行真实动作
▼
文件系统 / SSH / 数据库 / API…
│ 结构化结果(文本 + 资源 + 图片)
▼
模型组织回答 → 用户看到结果
02为什么需要它:没有 MCP 的世界
想象你要给 AI 助手接上「查数据库」的能力。没有 MCP 时,每一步都是手工活:
| 没有 MCP | 有了 MCP |
|---|---|
| 为每个客户端写一套集成:Claude Desktop 一套、Cursor 一套、自己的应用再一套 | 写一个 MCP server,所有客户端通用 |
| 工具越多,prompt 越长,模型越容易「忘了」有哪些工具 | 工具按 server 分组发现,按需加载 |
| 每个工具自己定义 JSON 格式,错误处理各写各的 | 协议统一:工具声明、调用、结果、错误都有标准形态 |
| 给工具加鉴权、限流、审计要重新发明轮子 | 协议层自带结构化能力,可以统一做安全层 |
03一个极简 MCP 示例(可运行)
依赖只有一个:pip install mcp(本教程验证版本 mcp==2.0.0)。
下面这个文件就是一个完整的 MCP server:
"""极简 MCP server:两个工具 + stdio 传输。"""
from mcp.server.mcpserver import MCPServer
server = MCPServer("hello-mcp")
@server.tool()
def add(a: int, b: int) -> int:
"""两个整数相加。"""
return a + b
@server.tool()
def greet(name: str) -> str:
"""打个招呼。"""
return f"你好,{name}!我是通过 MCP 被调用的。"
if __name__ == "__main__":
server.run(transport="stdio")
逐行拆解
| 代码 | 作用 |
|---|---|
MCPServer("hello-mcp") | 创建 server 实例,名字会出现在客户端的工具列表里 |
@server.tool() | 注册工具。函数签名(参数名 + 类型注解 + docstring)会自动变成工具的 JSON Schema |
def add(a: int, b: int) -> int | 参数类型注解就是协议声明,客户端据此生成表单/校验 |
server.run(transport="stdio") | 通过标准输入输出通信——MCP 最常用的传输方式,客户端用子进程拉起它 |
mcp==2.0.0 里类是 mcp.server.mcpserver.MCPServer;
1.x 里叫 mcp.server.fastmcp.FastMCP,用法几乎一样。
网上大量教程写的是 from mcp.server.fastmcp import FastMCP,
如果你装的是 2.x 会直接 ModuleNotFoundError。
兼容写法:try: from mcp.server.mcpserver import MCPServer as Server
except ImportError: from mcp.server.fastmcp import FastMCP as Server
04客户端怎么调:15 行代码连上它
用官方 SDK 写一个临时客户端,验证 server 是否工作(这也是我每次改完 server 必跑的冒烟测试形态):
import asyncio, sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 用子进程方式拉起 server(stdio 传输)
params = StdioServerParameters(command=sys.executable, args=["hello_mcp.py"])
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize() # 握手
tools = await session.list_tools() # 发现工具
print("tools:", [t.name for t in tools.tools])
r1 = await session.call_tool("add", {"a": 2, "b": 3})
print("add(2,3):", r1.content[0].text)
r2 = await session.call_tool("greet", {"name": "丞相"})
print("greet:", r2.content[0].text)
asyncio.run(main())
tools: ['add', 'greet']
add(2,3): 5
greet: 你好,丞相!我是通过 MCP 被调用的。
print() 会把协议报文冲乱,客户端直接崩。
调试信息一律走 stderr(print(..., file=sys.stderr)),
或写日志文件。这个坑几乎每个 MCP 新手都会踩一次。
05常见 MCP 代码分析:生产级 server 的五个模式
极简示例能跑,但距离「敢把它接到生产服务器」还差得远。 下面五个模式来自我为成都生产服务器写的 chengdu-mcp 项目(23 工具 / 81 单测), 每个都是踩过坑才固化的。
模式一:业务逻辑与 MCP 层分离
工具函数里只做「参数声明 + 调用业务函数 + 把结果转成协议格式」。
业务逻辑写成签名统一的纯函数,第一个参数是 runner(执行器):
# ops.py —— 业务层,完全不 import mcp
def disk_usage(runner, path) -> tuple[bool, str]:
"""统计磁盘占用。第一个参数是执行器,测试时注入假执行器。"""
res = runner.run(["df", "-h", path])
...
return True, clipped_text
# server.py —— MCP 层,只做声明和转换
@server.tool()
def disk_usage(path: str) -> str:
ok, text = ops.disk_usage(CONFIG.runner(), path)
if not ok:
raise ToolError(text) # 抛错 = 客户端看到 is_error
return text
模式二:执行器注入(Runner 模式)
# ssh.py —— 真实执行器与测试替身
class Result:
exit_code: int
stdout: str
stderr: str
transport_error: str | None = None # SSH 自身失败,区别于命令失败
class SSHRunner:
def run(self, argv, timeout=60, stdin=None) -> Result:
# subprocess 以 argv 数组形式执行,不经本地 shell,杜绝注入
proc = subprocess.run(["ssh", "chengdu", *argv], ...)
...
class FakeRunner: # 测试替身
def __init__(self, default=Result(0, "", "")):
self.default = default
self.calls = []
def run(self, argv, timeout=None, stdin=None) -> Result:
self.calls.append(argv) # 记录调用,测试断言用
return self.default
transport_error 探测逻辑,才把这类 bug 锁死在测试里。
模式三:路径白名单 + 逐段深度校验
服务器运维 MCP 最怕模型被提示词注入骗去读写任意路径。
光用 normpath 不够——posixpath.normpath("/../../etc/shadow")
会归一成 /etc/shadow 溜过前缀检查。正确做法是逐段推演深度:
def normalize(root: str, p: str) -> str:
parts = p.split("/")
depth = 0
for part in parts:
if part in ("", "."):
continue
if part == "..":
depth -= 1
if depth < 0: # 试图逃出根目录
raise GuardError(f"路径越界(试图逃出根目录):{p}")
else:
depth += 1
...
| 防护 | 做法 |
|---|---|
| 读白名单 | 只允许 /var/www、/root/workspace、/etc/nginx 等 |
| 写白名单 | 只允许站点根、nginx conf.d、备份目录三个根 |
| 危险操作 | 回滚、重载 nginx、任意命令执行必须显式 confirm=True |
| 任意命令 | 默认关闭,需环境变量 CHENGDU_ALLOW_EXEC=1 才启用(双闸门) |
模式四:写操作先备份 + MD5 双边校验
# 每次写文件的标准流水线
1. 探测原文件是否存在(stat)
2. 存在 → 先 cp 到 /root/backups/<label>-<时间戳>(备份失败即中止)
3. 新内容 scp 到远端 /tmp 暂存
4. 原子 mv 落地(避免写一半留下残缺文件)
5. 本地 md5 与远端 md5sum 比对,不一致直接报错
read_remote_file 竟然返回了成功
——因为代码用「MISSING 是否出现在 stdout」判断文件是否存在,而连接失败时 stdout
是空的,被误判成「文件存在且内容为空」。更危险的是 deploy 流程里 stat 探测失败
会被当成「新增文件」从而静默跳过备份。修复:transport_error 独立字段,
所有业务判断前先拦截连接失败。这条写进了 20 项回归测试。
模式五:给工具打「危险」标注
# SDK 2.0 的 ToolAnnotations:客户端可据此在 UI 上提示用户
from mcp.types import ToolAnnotations
def _ann(*, read_only=False, destructive=False):
return ToolAnnotations(
read_only_hint=read_only,
destructive_hint=destructive,
idempotent_hint=False,
open_world_hint=False,
title=None,
)
@server.tool(annotations=_ann(destructive=True))
def restore_backup(backup_path: str, target_path: str, confirm: bool = False) -> str:
"""从备份回滚覆盖目标路径。危险操作,需要 confirm=True。"""
...
06部署与挂载:三个常见坑
以挂到 Hermes Agent 为例(Claude Desktop / Cursor 同理,只是配置文件位置不同):
坑一:args 数组被存成字符串
hermes config set mcp_servers.xxx.args '["-m","my_server"]'
会把 JSON 数组存成字符串,启动时 pydantic 直接报
args Input should be a valid list。args 必须手工写成 YAML 列表:
mcp_servers:
chengdu:
command: /home/damo/workspace/chengdu/mcp-chengdu/.venv/bin/python
args:
- -m
- chengdu_mcp.server
enabled: true
坑二:用绝对路径的 venv python,别用裸 python
MCP server 由客户端以子进程方式拉起,PATH 往往和你的 shell 不一样。
写 python 可能拉起系统 Python,找不到你装的依赖。
command 用 venv 里的绝对路径最稳。
坑三:改完配置必须真的测一遍连接
hermes mcp test chengdu
# ✓ Connected (1266ms)
# ✓ Tools discovered: 23
配置「看起来对」和「真的能连」是两回事。每次改 server 或配置,跑一次连接测试, 再调用一两个工具验证,才算部署完成。
tests/smoke.py:用官方 SDK 的
StdioServerParameters 拉起 server,跑一遍所有只读工具 + 几个负例
(越权路径、缺 confirm 的危险操作),输出统一格式。发布前跑一次,比任何 review 都可靠。