7.3 MCP 协议详解
在上一节中，我们深入探讨了工具调用的基础机制——如何定义工具的 JSON Schema、如何处理 LLM 返回的函数调用请求、以及如何在应用层执行工具并返回结果。这套机制在单个应用内部运行良好，但当我们需要让 AI 模型连接越来越多的外部系统时，一个新问题浮现出来：每个工具的集成都需要从零开始编写胶水代码。不同 LLM 提供商的调用格式各异，工具的注册、发现、调用逻辑散落在应用代码的各个角落，缺乏统一的标准。
MCP（Model Context Protocol，模型上下文协议）正是为解决这一问题而生。它提供了一套标准化的协议，让 AI 模型与外部工具、数据源之间的连接变得像插入 USB-C 接口一样简单。本节将全面解析 MCP 协议的设计理念、架构模型和开发实践。
7.3.1 为什么需要 MCP：AI 的 USB-C 接口
USB-C 的启示
回顾 USB-C 接口普及之前的年代：每台设备都配有自己的专属充电线和数据线——老式手机用 Micro-USB，苹果设备用 Lightning，相机用专用接口，耳机用 3.5mm 耳机口。用户出门需要带一堆线缆，厂商需要为每款产品设计专用接口。USB-C 的出现统一了这一切：一个接口，既能充电、又能传数据、还能传输视频信号，所有设备共享同一种物理连接标准。
MCP 对 AI 工具集成做的事情，和 USB-C 对电子设备做的事情如出一辙：
┌──────────────────────────────────────────────────────────────────┐
│ MCP：AI 的 USB-C 接口 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ USB-C 之前 MCP 之前 │
│ ───────── ────── │
│ · 每个设备独有接口 · 每个 LLM 有不同的工具调用格式 │
│ · 线缆无法通用 · 工具代码绑定到特定 AI 平台 │
│ · 换设备 = 换全套配件 · 换 AI 平台 = 重写工具集成 │
│ │
│ USB-C 之后 MCP 之后 │
│ ───────── ────── │
│ · 统一物理接口 · 统一通信协议（JSON-RPC 2.0） │
│ · 一根线连所有设备 · 一个 Server 连所有 AI 客户端 │
│ · 即插即用 · 工具自动发现与调用 │
│ │
└──────────────────────────────────────────────────────────────────┘MCP 解决的核心痛点
在 MCP 出现之前，工具集成面临三个层次的痛点：
第一，格式碎片化。OpenAI 的 Function Calling、Anthropic 的 Tool Use、Google 的 Function Calling 虽然思路相似，但请求和响应的 JSON 结构各不相同。一个为 OpenAI 编写的工具集成，迁移到 Claude 上需要修改大量代码。
第二，耦合度高。工具的定义和调用逻辑直接嵌入在应用代码中。如果你想给 Claude Desktop 添加一个文件搜索功能，需要编写特定于 Claude Desktop 的插件代码；同样的功能用在 Cursor 中又要重新实现。
第三，缺乏发现机制。工具的注册和发现没有统一标准。应用启动时需要硬编码加载哪些工具，动态增减工具的能力有限。
MCP 通过引入标准化的中间层解决了这些问题。工具提供方只需实现一次 MCP Server，就可以被任何支持 MCP 的 AI 客户端使用——无论是 Claude Desktop、Cursor IDE，还是自行开发的 AI 应用。
MCP 的诞生背景
MCP 由 Anthropic 于 2024 年 11 月正式开源发布。它的设计灵感来源于 Language Server Protocol（LSP）——后者统一了编程语言和代码编辑器之间的通信标准，使得任何语言服务器可以被任何编辑器使用。MCP 希望在 AI 领域实现类似的标准化：让工具和数据源只需实现一次，就能被所有 AI 模型使用。
截至本书写作时，MCP 已获得广泛的社区支持。Anthropic、OpenAI、Microsoft 等公司均已在其产品中集成或宣布支持 MCP，生态系统中已有数百个开源 MCP Server，覆盖文件系统、数据库、浏览器自动化、搜索引擎、代码仓库等场景。
MCP 的设计原则
MCP 在设计上遵循以下原则：
- 开放标准：协议规范完全开源，任何组织和个人都可以实现 MCP Server 或 Client，无需授权。
- 传输无关：同一套协议可以运行在 stdio、SSE、HTTP 等不同传输层之上，开发者根据部署场景灵活选择。
- 安全优先：工具调用由 Host 应用决定是否执行，Server 无法主动发起请求。用户可以审查和批准每一次工具调用。
- 渐进增强：Server 可以声明自己支持哪些能力（tools、resources、prompts），Client 根据协商结果使用相应的功能，不支持的特性不会导致错误。 渐进增强：Server 可以声明自己支持哪些能力（tools、resources、prompts），Client 根据协商结果使用相应的功能，不支持的特性不会导致错误。
7.3.2 MCP 架构全景
三层角色模型
MCP 定义了三个核心角色，它们之间的关系可以用 USB-C 的类比来理解：
| 角色 | USB-C 类比 | 说明 |
|---|---|---|
| MCP Host | 你的电脑 | AI 应用程序本身，如 Claude Desktop、Cursor IDE、自定义 AI 助手。Host 决定何时调用工具、如何展示结果给用户。 |
| MCP Client | 电脑上的 USB-C 端口 | 协议客户端，运行在 Host 内部，负责与 Server 建立一对一的连接。每个 Server 对应一个 Client 实例。 |
| MCP Server | USB-C 设备（U盘、显示器等） | 暴露工具、资源和提示的独立进程。Server 是"被动的"——它只响应 Client 的请求，不主动发起调用。 |
理解三者的关系至关重要：Host 可以同时连接多个 Server（就像电脑有多个 USB-C 端口），每个 Server 通过独立的 Client 实例进行通信。Claude Desktop 可以同时连接文件系统 Server、数据库 Server 和搜索引擎 Server，AI 模型可以在一次对话中调用来自不同 Server 的工具。
┌─────────────────────────────────────────────────────────────┐
│ MCP 架构全景图 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────┐ │
│ │ MCP Host │ (Claude Desktop / Cursor / 自定义应用) │
│ │ │ │
│ │ ┌───────────┐ │ ┌───────────┐ ┌───────────┐ │
│ │ │Client A │─┼──│ Server A │ │ Server B │ │
│ │ ├───────────┤ │ │ (文件系统) │ │ (数据库) │ │
│ │ │Client B │─┼──│ │ │ │ │
│ │ ├───────────┤ │ └───────────┘ └───────────┘ │
│ │ │Client C │─┼──┌───────────────────────────┐ │
│ │ └───────────┘ │ │ Server C (搜索引擎) │ │
│ └───────────────┘ └───────────────────────────┘ │
│ │
│ 传输层: stdio / SSE / Streamable HTTP │
│ 协议层: JSON-RPC 2.0 │
│ │
└─────────────────────────────────────────────────────────────┘图 7-3：MCP 协议的三层架构
三种核心原语
MCP Server 可以对外暴露三种类型的能力，称为"原语"（Primitives）。继续用 USB-C 来类比：USB-C 接口可以传输电力（供电）、传输数据（文件传输）、传输视频（外接显示器），而 MCP 的三种原语同样对应不同类型的能力：
| 原语 | USB-C 类比 | 对应 Web 概念 | 特征 |
|---|---|---|---|
| Tools | 数据传输 | POST 端点 | 可执行的操作，可能产生副作用（如修改文件、发送邮件）。由 AI 模型决定何时调用。 |
| Resources | 只读存储 | GET 端点 | 可读取的数据，不产生副作用。由应用决定何时加载到上下文中。 |
| Prompts | 预设模式 | 模板引擎 | 可复用的交互模板，支持参数化。引导用户或 AI 进行特定工作流。 |
三者的区别需要特别注意：Tools 由 AI 模型自主决定调用，模型根据对话内容判断是否需要使用某个工具；Resources 由应用程序决定加载，通常由用户点击或系统自动触发；Prompts 由用户主动选择，用于引导特定的交互模式。这种分工设计确保了不同能力在权限和触发方式上的合理划分。
协议四层架构
从技术视角来看，MCP 协议可以划分为四层：
┌─────────────────────────────────────────────┐
│ 第 4 层：应用层 │
│ · 工具执行逻辑、资源读取、Prompt 生成 │
│ · 开发者主要关注这一层 │
├─────────────────────────────────────────────┤
│ 第 3 层：协议层（JSON-RPC 2.0） │
│ · 消息格式定义、请求/响应配对 │
│ · 能力协商、方法路由 │
├─────────────────────────────────────────────┤
│ 第 2 层：会话层 │
│ · 初始化握手、生命周期管理 │
│ · 连接状态维护 │
├─────────────────────────────────────────────┤
│ 第 1 层：传输层 │
│ · stdio / SSE / Streamable HTTP │
│ · 负责底层消息传递 │
└─────────────────────────────────────────────┘图 7-4：MCP 协议的四层技术架构
作为开发者，你通常只需关注第 4 层（应用层）和选择第 1 层（传输方式），中间两层由 SDK 自动处理。FastMCP 框架更是将第 3 层和第 4 层封装到了装饰器级别，让开发者几乎只需编写普通的 Python 函数即可。 作为开发者，你通常只需关注第 4 层（应用层）和选择第 1 层（传输方式），中间两层由 SDK 自动处理。FastMCP 框架更是将第 3 层和第 4 层封装到了装饰器级别，让开发者几乎只需编写普通的 Python 函数即可。
7.3.3 通信协议详解
JSON-RPC 2.0：MCP 的通信基础
MCP 协议的通信层基于 JSON-RPC 2.0，这是一个轻量级的远程过程调用协议。选择 JSON-RPC 而非 REST 或 GraphQL 的原因是：MCP 的通信是双向的——Client 向 Server 发送请求，Server 也可以向 Client 发送通知。JSON-RPC 2.0 天然支持这种双向通信模式，且消息格式简单、解析高效。
一条 JSON-RPC 消息可以是以下三种之一：
// 1. 请求（Request）：期望收到响应
{
"jsonrpc": "2.0", // 固定版本号
"id": 1, // 请求 ID，用于匹配响应
"method": "tools/call", // 要调用的方法名
"params": { // 方法参数
"name": "get_weather",
"arguments": {"city": "北京"}
}
}
// 2. 响应（Response）：对请求的回复
{
"jsonrpc": "2.0",
"id": 1, // 对应请求的 ID
"result": { // 成功时的结果
"content": [
{"type": "text", "text": "晴，28°C，湿度45%"}
]
}
}
// 3. 通知（Notification）：不需要响应的单向消息
{
"jsonrpc": "2.0",
"method": "notifications/initialized", // 通知方法名
"params": {} // 注意：通知没有 id 字段
}完整的交互流程
一次完整的 MCP 会话包含三个阶段：初始化、能力协商和正常操作。下面的时序图展示了一个典型的交互全过程：
Client Server
│ │
│═════════ 阶段1：初始化 ═════════════════ │
│ │
│──── initialize ────────────────────────▶│ 发起握手
│ {protocolVersion: "2024-11-05", │
│ capabilities: {tools: {}}, │
│ clientInfo: {name, version}} │
│ │
│◀─── Result ─────────────────────────────│ 返回服务器能力
│ {protocolVersion: "2024-11-05", │
│ capabilities: {tools, resources, │
│ prompts}, │
│ serverInfo: {name, version}} │
│ │
│──── notifications/initialized ─────────▶│ 确认初始化完成
│ (通知，无响应) │
│ │
│═════════ 阶段2：工具发现 ═══════════════ │
│ │
│──── tools/list ────────────────────────▶│ 获取工具列表
│◀─── [tool1, tool2, tool3, ...] ─────────│ 返回工具清单
│ │
│═════════ 阶段3：工具调用 ═══════════════ │
│ │
│──── tools/call ────────────────────────▶│ 调用工具
│ {name: "get_weather", │
│ arguments: {city: "北京"}} │
│ │
│◀─── Result ─────────────────────────────│ 返回执行结果
│ {content: [{type: "text", │
│ text: "晴，28°C"}], │
│ isError: false} │
│ │图 7-5：MCP 协议完整交互时序
关键 JSON-RPC 方法
MCP 定义了一套标准化的方法名，所有方法名使用斜杠分隔的命名空间格式：
| 方法 | 方向 | 阶段 | 说明 |
|---|---|---|---|
initialize | Client → Server | 初始化 | 握手，发送协议版本和客户端能力，协商共同支持的功能集 |
notifications/initialized | Client → Server | 初始化 | 通知 Server 初始化完成，可以开始正常操作。这是通知而非请求，不期望响应 |
tools/list | Client → Server | 操作 | 获取 Server 暴露的所有工具列表，包含名称、描述和参数 Schema |
tools/call | Client → Server | 操作 | 调用指定工具，传入工具名称和参数。返回执行结果或错误信息 |
resources/list | Client → Server | 操作 | 获取 Server 暴露的所有资源 URI 列表 |
resources/read | Client → Server | 操作 | 读取指定 URI 的资源内容 |
resources/subscribe | Client → Server | 操作 | 订阅资源变更通知，当资源内容变化时 Server 会主动推送 |
prompts/list | Client → Server | 操作 | 获取所有可用的 Prompt 模板列表 |
prompts/get | Client → Server | 操作 | 获取指定 Prompt 模板的内容，支持传入参数 |
ping | Client → Server | 操作 | 心跳检测，验证连接是否仍然活跃 |
能力协商机制
能力协商是 MCP 设计中一个精妙的细节。在 initialize 阶段，Client 和 Server 各自声明自己支持的能力：
// Client 声明自己支持的能力
{
"capabilities": {
"roots": {"listChanged": true}, // 支持文件系统根目录列表
"sampling": {} // 支持 Server 请求 Client 进行 LLM 采样
}
}
// Server 声明自己支持的能力
{
"capabilities": {
"tools": {"listChanged": true}, // 支持工具列表动态变化
"resources": {"subscribe": true, "listChanged": true},
"prompts": {"listChanged": true},
"logging": {} // 支持日志输出
}
}协商完成后，双方只使用对方声明支持的功能。如果 Server 没有声明 resources 能力，Client 就不会向它发送 resources/list 请求。这种设计确保了向前兼容性——当未来协议增加新能力时，旧版本的实现不会因为遇到未知能力而崩溃。
传输层详解
MCP 目前支持三种传输方式，各有适用场景：
| 传输方式 | 适用场景 | 连接方式 | 特点 |
|---|---|---|---|
| stdio | 本地工具、桌面应用集成 | Host 启动 Server 子进程 | 最简单，无需网络配置。Claude Desktop 默认使用此方式。 |
| SSE | 远程服务、Web 部署 | HTTP 长连接 | Server 可以主动推送消息。适合需要实时通知的场景。 |
| Streamable HTTP | 现代 Web 服务 | HTTP 请求/响应 + 流式传输 | 兼顾灵活性和效率，是未来推荐的标准传输方式。 |
选择建议：如果是为本地桌面应用（如 Claude Desktop）开发工具，使用 stdio 即可；如果是构建远程服务供团队共享，使用 Streamable HTTP；只有在特殊需要 Server 主动推送的场景下才考虑 SSE。 选择建议：如果是为本地桌面应用（如 Claude Desktop）开发工具，使用 stdio 即可；如果是构建远程服务供团队共享，使用 Streamable HTTP；只有在特殊需要 Server 主动推送的场景下才考虑 SSE。
7.3.4 使用 FastMCP 开发 MCP Server
FastMCP 框架简介
FastMCP 是 Anthropic 官方 Python SDK（mcp 包）提供的高层 API。它的设计理念与 FastAPI 相似——用装饰器声明工具、资源和提示，由框架自动处理协议层细节。开发者只需编写普通的 Python 函数，FastMCP 自动将其转换为符合 MCP 规范的工具。
FastMCP 的核心优势在于：它从函数的类型注解和 docstring 中自动提取参数 Schema 和工具描述，无需手动编写 JSON Schema。函数签名的参数名变成工具参数名，类型注解变成参数类型，docstring 变成工具描述。
安装 MCP SDK
# 安装 MCP Python SDK（含 CLI 工具）
pip install "mcp[cli]"
# 验证安装
mcp --version第一个 MCP Server：基础工具集
下面是一个功能完整的基础 MCP Server，我们逐行注释每一行代码的含义：
# server.py - 一个功能完整的 MCP Server
# 从 mcp 包中导入 FastMCP 类
# FastMCP 是高层 API，自动处理 JSON-RPC 协议和传输细节
from mcp.server.fastmcp import FastMCP
import httpx # HTTP 客户端库，用于异步网络请求
import json # JSON 编解码，用于资源数据序列化
from typing import Optional # 可选类型注解
# 创建 MCP Server 实例
# 参数 "我的工具服务器" 是 Server 名称，在工具发现时会展示给 Client
mcp = FastMCP("我的工具服务器")
# ==================== 定义工具（Tools）====================
# 使用 @mcp.tool() 装饰器将普通函数注册为 MCP 工具
# FastMCP 自动从以下位置提取元信息：
# - 函数名 -> 工具名称（add）
# - docstring -> 工具描述（"计算两个整数的和"）
# - 参数名 + 类型注解 -> 参数 Schema（a: int, b: int）
@mcp.tool()
def add(a: int, b: int) -> int:
"""计算两个整数的和"""
return a + b # 返回值会自动包装为 MCP 响应格式
@mcp.tool()
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
# 模拟天气数据（实际应用中应调用真实天气 API）
weather_data = {
"北京": "晴，28°C，湿度45%",
"上海": "多云，32°C，湿度70%",
"深圳": "阵雨，30°C，湿度85%",
}
# dict.get() 在找不到城市时返回默认提示，避免 KeyError
return weather_data.get(city, f"未找到{city}的天气数据")
@mcp.tool()
async def search_web(query: str, limit: int = 5) -> str:
"""搜索网页内容（模拟）"""
# 异步函数也受支持，FastMCP 会自动处理 async/sync 的差异
# limit=5 是默认值，Client 调用时可以省略此参数
return f"关于 '{query}' 的搜索结果（共 {limit} 条）：\n1. 相关文章一\n2. 相关文章二"
# ==================== 定义资源（Resources）====================
# 使用 @mcp.resource() 装饰器注册资源
# 参数 "config://app" 是资源的 URI，Client 通过此 URI 读取资源
# URI 使用自定义协议（config://），不限于 http:// 或 file://
@mcp.resource("config://app")
def get_app_config() -> str:
"""获取应用配置"""
# 返回 JSON 格式的配置信息
# ensure_ascii=False 保留中文，indent=2 增加可读性
return json.dumps({
"app_name": "我的工具服务器",
"version": "1.0.0",
"features": ["天气查询", "网页搜索", "数学计算"]
}, ensure_ascii=False, indent=2)
# URI 中可以使用 {path} 作为路径参数模板
# Client 读取 "file:///etc/hosts" 时，path 变量为 "/etc/hosts"
@mcp.resource("file://{path}")
def read_file(path: str) -> str:
"""读取文件内容（模拟）"""
# 实际应用中应读取真实文件并做安全校验
return f"[文件内容] {path} 的内容..."
# ==================== 定义提示（Prompts）====================
# 使用 @mcp.prompt() 装饰器注册可复用的提示模板
# Prompt 可以接收参数，返回格式化的提示文本
@mcp.prompt()
def code_review_prompt(code: str, language: str = "python") -> str:
"""生成代码审查提示"""
# 返回一段格式化的 Prompt 文本
# Client 调用 prompts/get 时获得此文本，可插入到对话中
return f"""请对以下 {language} 代码进行审查，关注：
1. 代码质量和可读性
2. 潜在的性能问题
3. 安全漏洞
4. 改进建议
```{language}
{code}
```"""
# ==================== 启动服务器 ====================
if __name__ == "__main__":
# mcp.run() 默认使用 stdio 传输
# Host（如 Claude Desktop）会作为父进程启动此脚本
# 通过标准输入/输出进行 JSON-RPC 通信
mcp.run()开发调试：MCP Inspector
MCP SDK 自带一个可视化调试工具——MCP Inspector。它是一个 Web 界面，让你在浏览器中测试 Server 的所有功能：
# 启动开发模式，自动打开浏览器调试界面
mcp dev server.py
# 调试界面提供以下功能：
# · 查看所有工具、资源、Prompt 的列表
# · 手动调用工具并查看返回结果
# · 读取资源内容
# · 查看 JSON-RPC 原始消息
# · 测试能力协商过程
# 确认调试通过后，可以直接安装到 Claude Desktop
mcp install server.py代码与 Schema 的对应关系
FastMCP 的魔法在于它如何从 Python 代码自动生成 MCP 工具定义。理解这种对应关系对调试和优化非常重要：
| Python 代码元素 | MCP 工具定义字段 | 示例 |
|---|---|---|
| 函数名 | name | def get_weather → "name": "get_weather" |
| docstring | description | """获取天气信息""" → "description": "获取天气信息" |
| 参数名 | properties 的 key | city: str → "city": {} |
| 类型注解 | type | str → "type": "string" |
| 默认值 | default | limit: int = 5 → "default": 5 |
Optional | 可选参数 | Optional[str] → 参数不在 required 列表中 |
比如上面的 search_web 函数，FastMCP 自动生成如下工具定义：
{
"name": "search_web",
"description": "搜索网页内容（模拟）",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"limit": {"type": "integer", "default": 5}
},
"required": ["query"]
}
}注意 limit 参数有默认值，所以不在 required 数组中。这种自动推导让开发者完全专注于业务逻辑，无需手动维护 Schema。 注意 limit 参数有默认值，所以不在 required 数组中。这种自动推导让开发者完全专注于业务逻辑，无需手动维护 Schema。
7.3.5 高级实践：生命周期与上下文管理
当 MCP Server 需要维护数据库连接、缓存等有状态资源时，需要使用生命周期管理机制。FastMCP 通过 lifespan 参数支持 Server 级别的上下文管理。
下面的示例展示了如何构建一个带数据库的 MCP Server，每行代码都配有详细注释：
# advanced_server.py - 带数据库和生命周期管理的高级 MCP Server
from mcp.server.fastmcp import FastMCP
from contextlib import asynccontextmanager # 异步上下文管理器装饰器
from collections.abc import AsyncIterator # 异步迭代器类型注解
from dataclasses import dataclass # 数据类装饰器，自动生成 __init__ 等
import sqlite3 # Python 内置 SQLite3 库
import os
from typing import Optional
# ==================== 应用上下文 ====================
# @dataclass 自动生成 __init__、__repr__ 等方法
# AppContext 在整个 Server 生命周期中共享，所有工具都可以访问
@dataclass
class AppContext:
"""应用级上下文，在整个 Server 生命周期中共享"""
db: sqlite3.Connection # 数据库连接，在 lifespan 中初始化
# @asynccontextmanager 将异步函数变成异步上下文管理器
# FastMCP 在 Server 启动时调用此函数，在关闭时执行 finally 块
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""管理 Server 启动和关闭时的资源"""
# -------- 启动阶段 --------
# 创建/打开 SQLite 数据库文件
db = sqlite3.connect("my_app.db")
# 建表（IF NOT EXISTS 避免重复创建）
db.execute("""
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
content TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.commit() # 提交 DDL 语句
try:
# yield 之前：启动阶段，初始化资源
# yield 的值会被传递给所有工具作为 ctx 参数
yield AppContext(db=db)
finally:
# yield 之后：关闭阶段，清理资源
# 无论正常退出还是异常退出都会执行
db.close()
# 创建带生命周期的 Server
# lifespan 参数告诉 FastMCP 使用 app_lifespan 管理上下文
mcp = FastMCP("高级工具服务器", lifespan=app_lifespan)
# ==================== 数据库工具 ====================
# ctx: AppContext 参数是特殊的——FastMCP 会自动注入上下文
# Client 调用时不需要（也不能）传递 ctx 参数
@mcp.tool()
def search_documents(ctx: AppContext, query: str, limit: int = 10) -> str:
"""在数据库中搜索文档"""
try:
# 使用参数化查询防止 SQL 注入（? 占位符）
# LIKE 操作符配合 % 通配符进行模糊搜索
cursor = ctx.db.execute(
"SELECT id, title, content FROM documents WHERE title LIKE ? OR content LIKE ? LIMIT ?",
(f"%{query}%", f"%{query}%", limit)
)
results = cursor.fetchall() # 获取所有匹配的行
if not results:
return f"未找到与 '{query}' 相关的文档"
# 格式化输出：每个文档一行标题 + 100 字预览
output = []
for doc_id, title, content in results:
output.append(f"[{doc_id}] {title}\n{content[:100]}...")
return "\n\n".join(output)
except Exception as e:
# 捕获异常并返回友好错误信息
# 不直接抛出异常，避免 Client 收到 500 错误
return f"搜索失败: {str(e)}"
@mcp.tool()
def add_document(ctx: AppContext, title: str, content: str) -> str:
"""添加新文档到数据库"""
try:
cursor = ctx.db.execute(
"INSERT INTO documents (title, content) VALUES (?, ?)",
(title, content)
)
ctx.db.commit() # INSERT 后必须 commit 才生效
# lastrowid 是新插入行的主键值
return f"文档已添加，ID: {cursor.lastrowid}"
except Exception as e:
return f"添加失败: {str(e)}"
# ==================== 系统工具 ====================
# 注意此工具没有 ctx 参数——它不需要访问应用上下文
# FastMCP 不要求每个工具都必须接收 ctx
@mcp.tool()
def get_system_info() -> str:
"""获取系统信息"""
import platform # 延迟导入，减少启动开销
import sys
return f"""
系统信息:
- 操作系统: {platform.system()} {platform.release()}
- Python 版本: {sys.version}
- 处理器: {platform.processor()}
"""
# ==================== 资源 ====================
# 资源也可以接收 ctx 参数，与工具的机制相同
@mcp.resource("stats://documents")
def get_document_stats(ctx: AppContext) -> str:
"""获取文档统计信息"""
cursor = ctx.db.execute("SELECT COUNT(*) FROM documents")
count = cursor.fetchone()[0] # COUNT(*) 返回单行单列
return f"文档总数: {count}"
if __name__ == "__main__":
mcp.run()lifespan 的工作原理
lifespan 机制是理解 FastMCP 高级用法的关键。其工作流程如下：
Server 启动
│
▼
调用 app_lifespan(server)
│
├─ 执行 yield 之前的代码（初始化数据库等）
│
├─ yield AppContext(db=db) ──→ 上下文被保存
│ │
│ ┌────────────────────────────┘
│ │
│ ▼
│ Server 进入正常运行状态
│ 每次工具调用时，ctx 参数被注入 AppContext
│ │
│ ◄────────────────────────────┘
│
▼
Server 收到关闭信号
│
▼
执行 finally 块（关闭数据库等）
│
▼
Server 退出图 7-6：lifespan 生命周期流程
这种设计的好处是：数据库连接只创建一次，在整个 Server 生命周期内复用，避免了每次工具调用都重新建立连接的开销。同时，资源清理代码集中在 finally 块中，确保不会因为异常导致资源泄漏。 这种设计的好处是：数据库连接只创建一次，在整个 Server 生命周期内复用，避免了每次工具调用都重新建立连接的开销。同时，资源清理代码集中在 finally 块中，确保不会因为异常导致资源泄漏。
7.3.6 MCP Client 开发
前面的示例中，Server 端的代码是核心，但理解 Client 端如何连接和调用同样重要。在实际应用中，你可能需要自己编写 MCP Client——比如构建自定义的 AI 助手应用，或将 MCP Server 集成到自己的工作流中。
下面是一个完整的 MCP Client 示例，连接到前面定义的 Todo Server，执行完整的工具发现和调用流程：
# mcp_client_demo.py - MCP Client 开发示例
import asyncio
# ClientSession：MCP 客户端会话，封装了 JSON-RPC 通信
# StdioServerParameters：配置 stdio 传输所需的参数
from mcp import ClientSession, StdioServerParameters
# stdio_client：创建 stdio 传输的上下文管理器
from mcp.client.stdio import stdio_client
async def main():
# -------- 1. 配置 Server 连接参数 --------
# StdioServerParameters 描述如何启动和连接 Server
# command：启动 Server 的命令
# args：命令行参数，指向 Server 脚本路径
server_params = StdioServerParameters(
command="python",
args=["todo_server.py"]
)
# -------- 2. 建立连接 --------
# stdio_client() 会启动 Server 子进程
# 返回 (read, write) 一对流，用于 JSON-RPC 通信
async with stdio_client(server_params) as (read, write):
# ClientSession 封装了完整的 MCP 协议逻辑
# 包括初始化、能力协商、方法调用等
async with ClientSession(read, write) as session:
# -------- 3. 初始化连接 --------
# initialize() 发送 JSON-RPC initialize 请求
# 完成协议版本协商和能力交换
await session.initialize()
print("✅ 已连接到 MCP Server")
# -------- 4. 工具发现 --------
# list_tools() 发送 tools/list 请求
# 返回 Server 暴露的所有工具的元数据
tools_result = await session.list_tools()
print("\n📦 可用工具:")
for tool in tools_result.tools:
# tool.name: 工具名称
# tool.description: 工具描述
# tool.inputSchema: 参数 Schema（JSON Schema 格式）
print(f" - {tool.name}: {tool.description}")
# -------- 5. 调用工具 --------
# call_tool() 发送 tools/call 请求
# 参数1：工具名称
# 参数2：参数字典（key 必须与工具定义的参数名匹配）
print("\n🔧 调用 add_todo...")
result = await session.call_tool(
"add_todo",
arguments={"title": "学习 MCP 协议", "priority": "high"}
)
# result.content 是一个内容块列表
# 每个内容块有 type 和 text 字段
print(f" 结果: {result.content[0].text}")
# 再次调用，添加第二个待办
result = await session.call_tool(
"add_todo",
arguments={"title": "完成课程练习", "priority": "medium"}
)
print(f" 结果: {result.content[0].text}")
# -------- 6. 列出所有待办 --------
print("\n📋 列出待办:")
result = await session.call_tool("list_todos", arguments={"status": "all"})
print(result.content[0].text)
# -------- 7. 标记完成 --------
print("\n✅ 完成待办 #1:")
result = await session.call_tool("complete_todo", arguments={"todo_id": 1})
print(f" 结果: {result.content[0].text}")
# -------- 8. 读取资源 --------
# 资源使用 list_resources() 和 read_resource() 方法
print("\n📄 读取资源:")
resources = await session.list_resources()
for resource in resources.resources:
# resource.uri 是资源的 URI 标识符
content = await session.read_resource(resource.uri)
print(f" {resource.uri}: {content.contents[0].text}")
# 运行异步主函数
asyncio.run(main())Client 开发的关键要点
编写 MCP Client 时，有几个容易出错的细节：
第一，必须调用 initialize()。在调用任何工具之前，必须先完成初始化握手。跳过此步骤会导致后续所有请求被拒绝。
第二，参数名必须精确匹配。call_tool 的 arguments 字典中，key 必须与 Server 端工具函数的参数名完全一致（区分大小写）。如果 Server 定义了 city: str，Client 传 {"City": "北京"} 会导致参数不匹配。
第三，正确处理返回值。call_tool 返回的对象中，content 是一个列表而非单个值。这是因为 MCP 工具可以返回多种类型的内容（文本、图片、嵌入资源等），即使只返回文本，它也是一个包含单个文本块的列表。
第四，异步上下文管理。连接和会话都使用 async with 上下文管理器，确保在退出时正确关闭连接和清理资源。不要手动管理连接的生命周期。 第四，异步上下文管理。连接和会话都使用 async with 上下文管理器，确保在退出时正确关闭连接和清理资源。不要手动管理连接的生命周期。
7.3.7 MCP 生态系统与集成
官方与社区 Server
MCP 的价值不仅在于协议本身，更在于围绕协议生长出的生态系统。正如 USB-C 的成功依赖于大量支持该接口的设备一样，MCP 的成功也需要大量预构建的 Server。截至本书写作时，生态中已有数百个开源 Server，覆盖常见的外部系统集成场景：
| Server 名称 | 功能描述 | 语言 | 安装方式 |
|---|---|---|---|
server-filesystem | 安全的文件系统读写操作 | Node.js | npx @modelcontextprotocol/server-filesystem /path |
server-puppeteer | 浏览器自动化（截图、点击、填写） | Node.js | npx @modelcontextprotocol/server-puppeteer |
server-github | GitHub API 操作（仓库、Issue、PR） | Node.js | 配置 GITHUB_TOKEN 环境变量 |
server-postgres | PostgreSQL 数据库只读查询 | Node.js | 配置数据库连接字符串 |
server-brave-search | Brave 搜索引擎网页搜索 | Node.js | 配置 BRAVE_API_KEY |
server-sqlite | SQLite 数据库操作 | Python | uvx mcp-server-sqlite --db-path path |
server-git | Git 仓库操作（日志、diff、blame） | Python | uvx mcp-server-git |
server-memory | 基于知识图谱的持久化记忆 | Node.js | npx @modelcontextprotocol/server-memory |
这些 Server 大多是即装即用的——配置好环境变量或路径，就能在 AI 客户端中使用。你不需要编写任何代码，就能让 AI 拥有读取 GitHub 仓库、查询数据库、操作浏览器的能力。
Claude Desktop 集成配置
Claude Desktop 是目前最成熟的 MCP Host 应用之一。集成 MCP Server 只需编辑一个 JSON 配置文件：
// 文件路径（macOS）：
// ~/Library/Application Support/Claude/claude_desktop_config.json
// 文件路径（Windows）：
// %APPDATA%\Claude\claude_desktop_config.json
{
"mcpServers": {
// 自定义 Server：指向你自己的 Python 脚本
"my-tools": {
"command": "python",
"args": ["/path/to/server.py"]
},
// 文件系统 Server：限制可访问的目录
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/Users/username/Documents",
"/Users/username/Projects"
]
},
// GitHub Server：需要 API Token
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {
"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
}
}
}
}配置完成后重启 Claude Desktop，它会在启动时自动连接所有配置的 MCP Server。你可以在对话中直接让 Claude 使用这些工具——比如"在 Documents 文件夹中搜索包含 '报告' 的文件"，Claude 会自动调用文件系统 Server 的搜索工具。
构建自定义 MCP Host
如果你在开发自己的 AI 应用，可以将 MCP Client 集成到你的应用中。以下是一个简化的集成框架：
# custom_host.py - 将 MCP 集成到自定义 AI 应用中的简化示例
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI # 以 OpenAI 为例，也可替换为其他 LLM
class SimpleMCPHost:
"""一个简化的 MCP Host 实现"""
def __init__(self):
self.servers = {} # 存储所有已连接的 Server 会话
self.all_tools = [] # 汇总所有 Server 的工具
async def connect_server(self, name: str, command: str, args: list):
"""连接一个 MCP Server 并收集其工具"""
params = StdioServerParameters(command=command, args=args)
read, write = await stdio_client(params).__aenter__()
session = await ClientSession(read, write).__aenter__()
await session.initialize()
# 获取该 Server 的工具列表
result = await session.list_tools()
self.servers[name] = session
# 将工具名加上 Server 前缀以避免命名冲突
for tool in result.tools:
tool.full_name = f"{name}__{tool.name}"
self.all_tools.append((tool, session))
async def call_tool(self, full_name: str, arguments: dict):
"""根据完整工具名调用对应 Server 的工具"""
for tool, session in self.all_tools:
if tool.full_name == full_name:
result = await session.call_tool(tool.name, arguments)
return result.content[0].text
return f"未找到工具: {full_name}"
def get_openai_tools(self):
"""将 MCP 工具转换为 OpenAI Function Calling 格式"""
tools = []
for tool, _ in self.all_tools:
tools.append({
"type": "function",
"function": {
"name": tool.full_name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
return tools这个简化示例展示了 MCP Host 的核心逻辑：连接多个 Server、汇总工具列表、将工具定义转换为 LLM 所需的格式、根据 LLM 的调用请求路由到正确的 Server。实际的 Host 实现（如 Claude Desktop）还需要处理权限管理、用户确认、错误恢复等更多细节。 这个简化示例展示了 MCP Host 的核心逻辑：连接多个 Server、汇总工具列表、将工具定义转换为 LLM 所需的格式、根据 LLM 的调用请求路由到正确的 Server。实际的 Host 实现（如 Claude Desktop）还需要处理权限管理、用户确认、错误恢复等更多细节。
7.3.8 常见误区
在学习和使用 MCP 的过程中，有几个常见的理解偏差和实践陷阱值得特别注意。
误区一：MCP 就是 Function Calling 的换个说法
这是一个非常普遍的误解。MCP 和 Function Calling 是不同层次的东西：Function Calling 是 LLM 的一种能力——模型能识别何时需要调用函数，并生成结构化的调用参数；MCP 是一个通信协议——它定义了工具如何被注册、发现和调用的标准化方式。
打个比方：Function Calling 是大脑决定"我要用计算器算一下"，MCP 是计算器与大脑之间通信的"接口标准"。你可以用 Function Calling 配合 MCP，也可以用 Function Calling 不用 MCP（就像前面 7.2 节中那样直接在代码中定义工具），还可以用 MCP 配合其他工具调用机制。
MCP 的真正价值在于解耦：它让工具的实现与 AI 应用的实现相互独立，一次编写、到处使用。
误区二：MCP Server 可以主动调用 AI 模型
MCP 的架构设计中，Server 是完全被动的。它不能主动发起请求，只能响应 Client 的调用。这个设计是刻意的——它简化了安全模型，让 Host 应用始终保持对工具调用的控制权。
如果你需要 Server 端主动触发 AI 操作（比如数据库变化时自动让 AI 分析），应该使用 MCP 的 sampling 能力：Server 向 Client 发送 sampling/createMessage 请求，由 Client 决定是否执行 LLM 调用并将结果返回给 Server。但即使在这种情况下，最终的决定权仍在 Client 端。
误区三：所有外部数据都应该通过 MCP 暴露
并非所有数据都适合作为 MCP Resource 暴露。Resources 的设计初衷是提供"上下文补充"——当 AI 需要理解某个文档、查看某项配置时，通过资源读取获取这些信息。但以下场景不适合用 Resources：
- 频繁变化的数据（如实时股价）：每次读取都返回不同值，容易导致上下文混乱
- 大量结构化数据（如整个数据库表）：应通过 Tools 做查询过滤，而非一次性全部加载
- 敏感信息（如 API 密钥）：不应通过 Resources 暴露给 AI 上下文
对于查询类需求，使用 Tools（如 query_database）比 Resources 更合适。Tools 可以接收参数做精准查询，返回精简结果，避免将大量无关数据塞入上下文。
误区四：FastMCP 装饰器中的 docstring 不重要
在 FastMCP 中，函数的 docstring 不是可有可无的注释——它是 AI 模型理解工具用途的关键信息。docstring 会被自动提取为工具的 description 字段，发送给 LLM。如果 docstring 写得不清楚或缺失，模型可能不知道何时该用这个工具，或者误解工具的功能。
好的工具 docstring 应该回答三个问题：这个工具做什么？参数是什么含义？返回什么结果？
# ❌ 不好的 docstring
@mcp.tool()
def search(query: str) -> str:
"""搜索""" # 太简略，模型不知道搜什么、返回什么
...
# ✅ 好的 docstring
@mcp.tool()
def search(query: str, limit: int = 10) -> str:
"""在文档库中搜索匹配关键词的文档，返回标题和摘要列表。
Args:
query: 搜索关键词，支持模糊匹配
limit: 返回结果数量上限，默认10
Returns:
匹配的文档列表，每条包含 ID、标题和100字摘要
"""
...误区五：MCP 已经成熟到可以直接用于生产环境
虽然 MCP 的核心协议已经稳定，但生态仍在快速演进中。协议版本在持续更新，部分 SDK API 可能发生变化，某些高级特性（如 sampling、roots）的实现还不够完善。在生产环境中使用 MCP 时，建议：
- 锁定特定版本的 SDK，避免自动升级引入 breaking change
- 对关键工具准备降级方案（当 MCP 连接失败时，回退到直接函数调用）
- 充分测试错误处理和重连逻辑，网络环境下的稳定性需要特别关注
- 关注协议版本的兼容性，Server 和 Client 的协议版本需要匹配
7.3.9 本节小结
MCP（Model Context Protocol）是连接 AI 模型与外部工具、数据源的标准化协议。它被形象地比喻为"AI 的 USB-C 接口"——正如 USB-C 统一了各种电子设备的物理接口，MCP 统一了 AI 应用与工具服务之间的通信接口。
本节从以下几个维度全面解析了 MCP：
架构层面，MCP 采用三层角色模型——Host（AI 应用）、Client（协议客户端）和 Server（工具服务端），基于 JSON-RPC 2.0 协议进行通信。协议支持 stdio、SSE、Streamable HTTP 三种传输方式，通过能力协商机制实现渐进式增强。
原语层面，MCP 定义了三种核心能力：Tools（可执行的操作，由 AI 模型决定调用）、Resources（可读取的数据，由应用决定加载）、Prompts（可复用的交互模板，由用户主动选择）。这种分工确保了不同能力在权限和触发方式上的合理划分。
开发层面，FastMCP 框架通过装饰器将普通 Python 函数转化为 MCP 工具，自动从类型注解和 docstring 提取参数 Schema，极大降低了开发门槛。lifespan 机制支持数据库连接等有状态资源的管理。MCP Inspector 提供了可视化的调试体验。
生态层面，已有数百个开源 MCP Server 覆盖文件系统、数据库、浏览器、代码仓库等常见场景。Claude Desktop 等 Host 应用通过简单的 JSON 配置即可集成多个 Server。
MCP 的意义远不止于技术协议本身。它代表了一种趋势：AI 工具生态正在从碎片化的"每家自己做"走向标准化的"一次编写、处处可用"。正如 USB-C 让用户摆脱了线缆混乱的困扰，MCP 正在让 AI 开发者从重复的工具集成工作中解放出来，专注于更有价值的业务逻辑。
回顾本节，我们从协议层面理解了 MCP 如何标准化工具的发现与调用。但掌握协议只是第一步——在实际的 AI Agent 系统中，一个更复杂的问题摆在面前：当 Agent 可以使用数十甚至上百个工具时，如何决定在什么场景下选择哪些工具？如何编排多个工具的调用顺序？当多个工具需要组合使用时，如何设计高效的编排策略？
这些问题将在下一节 7.4 工具选择与编排 中深入讨论。我们将探讨工具数量增长带来的"选择困境"、基于语义相似度的工具检索机制、以及多步骤工具编排的常见模式，帮助你构建在复杂场景下仍能高效运作的 AI Agent。
延伸阅读
- MCP 官方文档与规范：https://modelcontextprotocol.io/
- MCP Python SDK 文档：https://pypi.org/project/mcp/
- MCP 协议规范（JSON-RPC 消息定义）：https://spec.modelcontextprotocol.io/
- Anthropic Tool Use 文档：https://docs.anthropic.com/en/docs/build-with-claude/tool-use