Agent 开发系列 · 01

MCP 是什么?写给开发者的最小入门

MCP(Model Context Protocol,模型上下文协议)是 Anthropic 在 2024 年底开源的 「AI 应用与外部工具之间的标准接口」。本教程从概念讲到能跑的最小示例, 最后拆解一个生产级 MCP server 的代码模式——文中所有代码都在本机真实运行过, 输出即实测输出。

难度 ★☆☆☆☆约 20 分钟2026.08

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…
   │ 结构化结果(文本 + 资源 + 图片)
   ▼
模型组织回答 → 用户看到结果
▎本教程配套的真东西 为了写这篇教程,我把日常对一台生产服务器(Ubuntu + nginx,托管三个站点)的 SSH 运维动作, 封装成了一个 MCP server:23 个工具、81 项单测、写操作强制备份 + MD5 双边校验。 本文第 5 节的「代码分析」全部来自这个真实项目,而不是 PPT 里的伪代码。

02为什么需要它:没有 MCP 的世界

想象你要给 AI 助手接上「查数据库」的能力。没有 MCP 时,每一步都是手工活:

没有 MCP有了 MCP
为每个客户端写一套集成:Claude Desktop 一套、Cursor 一套、自己的应用再一套 写一个 MCP server,所有客户端通用
工具越多,prompt 越长,模型越容易「忘了」有哪些工具 工具按 server 分组发现,按需加载
每个工具自己定义 JSON 格式,错误处理各写各的 协议统一:工具声明、调用、结果、错误都有标准形态
给工具加鉴权、限流、审计要重新发明轮子 协议层自带结构化能力,可以统一做安全层
▎一句话判断要不要用 MCP 你的工具只给自己一个应用用、且不打算换客户端 → 直接写函数调用就行,不必上 MCP。 你的工具可能被多个 AI 客户端使用,或者想沉淀成可复用资产 → MCP 是当前事实标准。

03一个极简 MCP 示例(可运行)

依赖只有一个:pip install mcp(本教程验证版本 mcp==2.0.0)。 下面这个文件就是一个完整的 MCP server:

hello_mcp.py
"""极简 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 最常用的传输方式,客户端用子进程拉起它
▎版本坑:SDK 1.x 与 2.x 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 必跑的冒烟测试形态):

client_probe.py
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 被调用的。
▎调试纪律:stdio server 里禁止 print 到 stdout stdio 传输下,stdout 是协议通道,print() 会把协议报文冲乱,客户端直接崩。 调试信息一律走 stderrprint(..., 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
▎为什么这么分 业务函数完全不依赖 MCP SDK → 单测里注入一个「假 runner」就能全离线跑, 81 项测试 0.1 秒跑完,不碰真实服务器。改协议版本、换传输方式,业务层一行不动。

模式二:执行器注入(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
▎替身必须和真身一样严格 我最初让 FakeRunner 直接返回预设值,结果一个「SSH 超时被误判成业务成功」的 bug 在真实环境炸了、单测却全绿——因为替身比真身宽松。后来让替身也走同一套 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 比对,不一致直接报错
▎真实事故驱动的设计 我在验证写路径时遇到一次 ssh 超时,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 或配置,跑一次连接测试, 再调用一两个工具验证,才算部署完成。

▎冒烟测试模板 我每个 MCP server 都带一个 tests/smoke.py:用官方 SDK 的 StdioServerParameters 拉起 server,跑一遍所有只读工具 + 几个负例 (越权路径、缺 confirm 的危险操作),输出统一格式。发布前跑一次,比任何 review 都可靠。