一、MCP 是什么? 为什么需要它?
想象一下，你正在开发一个 AI 编程助手，它需要:
读取和修改项目文件
查询数据库 Schema
搜索代码仓库
执行 Git 操作
传统做法是为每个数据源写一套专用代码，不同团队重复造轮子。Model Context Protocol(MCP) 就是为了解决这个问题而生的开放标准协议。
通俗理解: MCP 就像是「AI 应用的 USB 接口标准」。就像 USB 让不同设备都能接入电脑一样，MCP 让不同的数据源和工具都能以统一方式接入 AI 应用。
实际案例: 在 Claude Desktop 中，你可以配置多个官方 MCP 服务器:
Filesystem 服务器: 安全地读写本地文件，有权限控制
SQLite 服务器: 查询和分析 SQLite 数据库，自动生成 SQL
GitHub 服务器: 搜索仓库、创建 Issue、管理 PR
你的 AI 应用只需实现一个 MCP 客户端，就能连接所有服务器，无需为每个服务器写专用代码。
二、架构设计： 三个角色的分工
MCP 采用宿主-客户端-服务器三层架构，就像一家公司的组织结构:
宿主(Host) = 总经理
管理所有客户端
控制安全策略和权限
负责 AI 模型的调用
客户端(Client) = 部门经理
客户端负责连接服务器
负责双方的沟通协调
转发消息和通知
服务器(Server) = 业务专员
提供具体功能(资源、工具、提示模板)
可以是本地程序或远程服务
不知道其他服务器的存在
三、协议约定：统一规范与个性化扩展
每个 MCP 服务器提供的工具、资源都不一样，但它们都遵循相同的 MCP 协议规范。
3.1 协议的分层设计
MCP 采用 基础协议 + 功能扩展 的设计，就像 HTTP 协议一样:
核心层(所有实现必须支持) :
功能层(按需选择) :
这样设计的好处:
统一的基础协议 → 保证互操作性 +灵活的功能选择 → 满足不同场景需求 ↓既标准化又可扩展
复制代码
3.2 协议约定的过程
步骤 1: 基础协议是固定的
所有 MCP 服务器和客户端都遵循相同的 JSON-RPC 2.0 格式:
// 请求格式(固定){ "jsonrpc": "2.0", // 必须是2.0 "id": 1, // 唯一标识 "method": "方法名", // 要调用的方法 "params": {...} // 参数对象}
// 响应格式(固定){ "jsonrpc": "2.0", "id": 1, // 对应请求的ID "result": {...} // 成功结果 // 或 "error": {...} // 错误信息}
复制代码
步骤 2: 能力在初始化时协商
// 客户端发起初始化{ "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": { "sampling": {}, // 我支持LLM采样 "roots": {} // 我支持根目录 }, "clientInfo": {"name": "MyClient", "version": "1.0"} }}
// 服务器响应{ "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {}, // 我提供工具 "resources": {} // 我提供资源 }, "serverInfo": {"name": "SQLiteServer", "version": "2.0"} }}
复制代码
协商完成后，双方都知道对方支持什么功能，只使用交集部分。
步骤 3: 方法名称是标准化的
MCP 规范定义了标准方法名:
步骤 4: 具体内容是个性化的
虽然方法名固定，但每个服务器返回的具体数据不同:
// SQLite服务器的工具{ "tools": [ {"name": "query", "description": "执行SQL查询"}, {"name": "list_tables", "description": "列出所有表"} ]}
// Filesystem服务器的工具{ "tools": [ {"name": "read_file", "description": "读取文件"}, {"name": "write_file", "description": "写入文件"}, {"name": "search_files", "description": "搜索文件"} ]}
复制代码
3.3 协议发现机制
客户端如何知道服务器有哪些工具?
第一步：列举
客户端 → 服务器: {"method": "tools/list"}服务器 → 客户端: { "tools": [ { "name": "query", "description": "执行SQL查询", "inputSchema": { // JSON Schema定义输入格式 "type": "object", "properties": { "sql": {"type": "string"} } } } ]}
复制代码
第二步：调用
客户端 → 服务器: { "method": "tools/call", "params": { "name": "query", // 使用第一步获得的工具名 "arguments": {"sql": "SELECT * FROM users"} }}
复制代码
关键点:通过 JSON Schema，客户端知道如何正确调用工具，无需硬编码。
四、协议基础：如何通信?
MCP 基于 JSON-RPC 2.0 构建，这是一个成熟的远程过程调用协议。理解这一层对掌握 MCP 至关重要。
4.1 JSON-RPC 2.0 基础
消息类型
MCP 中有三种基本消息类型。
1. 请求(Request) - 期待响应
{ "jsonrpc": "2.0", // 协议版本,必须是"2.0" "id": 1, // 请求唯一标识(字符串或数字) "method": "tools/list", // 要调用的方法名 "params": { // 可选的参数对象 "cursor": "page2" }}
复制代码
2. 响应(Response) - 对请求的回复
// 成功响应{ "jsonrpc": "2.0", "id": 1, // 必须与请求的id相同 "result": { // 成功结果 "tools": [ {"name": "query", "description": "执行查询"} ] }}
// 错误响应{ "jsonrpc": "2.0", "id": 1, "error": { // 错误对象 "code": -32602, // 错误码(整数) "message": "参数无效", // 错误描述 "data": { // 可选的额外信息 "field": "cursor", "reason": "格式错误" } }}
复制代码
3. 通知(Notification) - 单向消息,无需响应
{ "jsonrpc": "2.0", "method": "notifications/resources/updated", // 通知方法名 "params": { // 通知参数 "uri": "file:///project/data.json" } // 注意:没有id字段}
复制代码
标准错误码
MCP 使用 JSON-RPC 2.0 的标准错误码:
4.2 能力协商详解
能力协商是 MCP 连接建立的第一步，决定了整个会话中可用的功能。
初始化流程详解
阶段 1: 客户端发起初始化
{ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", // 客户端支持的协议版本 "capabilities": { // 客户端能力声明 "roots": { // 支持根目录 "listChanged": true // 支持根目录变更通知 }, "sampling": {}, // 支持LLM采样 "elicitation": {}, // 支持用户询问 "experimental": { // 实验性功能 "customFeature": {} // 自定义功能 } }, "clientInfo": { // 客户端信息 "name": "MyAIApp", // 程序名(必填) "version": "1.2.0", // 版本号(必填) "title": "我的AI应用" // 显示名称(可选) } }}
复制代码
阶段 2: 服务器响应能力
{ "jsonrpc": "2.0", "id": 1, "result": { "protocolVersion": "2024-11-05", // 服务器选择的协议版本 "capabilities": { // 服务器能力声明 "resources": { // 支持资源 "subscribe": true, // 支持资源订阅 "listChanged": true // 支持资源列表变更通知 }, "tools": { // 支持工具 "listChanged": true }, "prompts": { // 支持提示模板 "listChanged": false // 不支持列表变更通知 }, "logging": {} // 支持日志输出 }, "serverInfo": { // 服务器信息 "name": "sqlite-mcp-server", "version": "2.1.0", "title": "SQLite MCP服务器" }, "instructions": "此服务器提供SQLite数据库访问能力" // 可选的使用说明 }}
复制代码
阶段 3: 客户端确认就绪
{ "jsonrpc": "2.0", "method": "notifications/initialized" // 无id,这是通知}
复制代码
协议版本协商规则
客户端请求版本: "2024-11-05" ↓ 服务器支持? ↙ ↘ 支持 不支持 ↓ ↓返回相同版本 返回服务器支持的最新版本 ↓ ↓协商成功 客户端检查是否支持 ↙ ↘ 支持 不支持 ↓ ↓ 协商成功 断开连接
复制代码
实际示例:
// 场景1: 版本匹配客户端: "protocolVersion": "2024-11-05"服务器: "protocolVersion": "2024-11-05" ✅ 成功
// 场景2: 服务器版本更新客户端: "protocolVersion": "2024-06-01"服务器: "protocolVersion": "2024-11-05" → 客户端检查是否支持2024-11-05 → 如果不支持则断开
// 场景3: 客户端版本更新客户端: "protocolVersion": "2025-01-01"服务器: "protocolVersion": "2024-11-05" → 客户端检查是否支持2024-11-05 → 如果支持则降级使用
复制代码
能力交集计算
初始化后,双方只能使用共同支持的能力:
客户端能力: {roots, sampling, elicitation}服务器能力: {resources, tools, prompts} ↓ 可用功能集合 ├─ 客户端 → 服务器: resources, tools, prompts └─ 服务器 → 客户端: roots, sampling, elicitation
复制代码
示例:
# 客户端代码示例if server_capabilities.get("tools"): # 服务器支持工具,可以调用 tools = await session.list_tools()else: # 服务器不支持工具,跳过 print("服务器不提供工具功能")
if client_capabilities.get("sampling"): # 客户端支持采样,服务器可以请求 # (服务器端会检查这个能力) pass
复制代码
4.3 连接生命周期深入
完整的消息时序图
客户端 服务器 │ │ │ 1. initialize (请求) │ ├──────────────────────────────────────>│ │ {protocolVersion, capabilities} │ │ │ │ 2. initialize (响应) │ │<──────────────────────────────────────┤ │ {protocolVersion, capabilities} │ │ │ │ 3. initialized (通知) │ ├──────────────────────────────────────>│ │ │ │═══════════ 正常操作阶段 ════════════ │ │ │ │ 4. tools/list (请求) │ ├──────────────────────────────────────>│ │ │ │ 5. tools/list (响应) │ │<──────────────────────────────────────┤ │ {tools: [...]} │ │ │ │ 6. tools/call (请求) │ ├──────────────────────────────────────>│ │ {name: "query", arguments: {...}} │ │ │ │ 7. notifications/progress (通知) │ │<──────────────────────────────────────┤ │ {progress: 50, total: 100} │ │ │ │ 8. tools/call (响应) │ │<──────────────────────────────────────┤ │ {content: [...]} │ │ │ │ 9. notifications/resources/updated │ │<──────────────────────────────────────┤ │ {uri: "file://..."} │ │ │ │═══════════ 关闭阶段 ═══════════ │ │ │ │ 10. 关闭stdin │ ├─────────────X │ │ │ │ 服务器退出
复制代码
初始化前的限制
在initialized通知发送前:
客户端只能发送:
✅ initialize请求
✅ ping请求(用于保活)
❌ 其他任何请求
服务器只能发送:
✅ initialize响应
✅ ping请求
✅ logging通知(日志)
❌ 其他任何消息
违反限制的后果:
// 客户端在初始化前调用tools/list请求: {"method": "tools/list"}响应: { "error": { "code": -32600, "message": "会话未初始化" }}
复制代码
超时和重试机制
请求超时:
import asyncio
# 设置30秒超时try: result = await asyncio.wait_for( session.call_tool("slow_operation", {}), timeout=30.0 )except asyncio.TimeoutError: # 发送取消通知 await session.send_notification( "notifications/cancelled", {"requestId": "123", "reason": "超时"} )
复制代码
进度通知重置超时:
# 当收到进度通知时,可以重置超时计时器timeout = 30 # 基础超时max_timeout = 300 # 最大超时(5分钟)
while True: try: msg = await wait_for_message(timeout) if msg.method == "notifications/progress": # 收到进度,重置超时 timeout = 30 except TimeoutError: # 超时处理 break
复制代码
4.4 传输方式对比
stdio 传输详解
优点:
✅ 简单直接,适合本地开发
✅ 进程隔离,安全性好
✅ 自动管理生命周期
✅ 无需网络配置
缺点:
❌ 只能本地使用
❌ 不支持多客户端
❌ 调试相对困难
消息格式:
每个 JSON 对象占一行,以\n分隔。
HTTP 传输详解
架构:
┌─────────┐ HTTP POST ┌─────────┐│ ├──────────────────────────>│ ││ 客户端 │ 请求/通知/响应(JSON-RPC) │ 服务器 ││ │<──────────────────────────┤ │└─────────┘ HTTP 响应/SSE流 └─────────┘ (application/json 或 text/event-stream)
复制代码
发送消息(POST) :
POST /mcp HTTP/1.1Host: localhost:8080Content-Type: application/jsonAccept: application/json, text/event-streamMcp-Session-Id: abc123
{"jsonrpc":"2.0","id":1,"method":"tools/list"}
复制代码
立即响应(JSON) :
HTTP/1.1 200 OKContent-Type: application/json
{"jsonrpc":"2.0","id":1,"result":{"tools":[...]}}
复制代码
流式响应(SSE) :
HTTP/1.1 200 OKContent-Type: text/event-streamMcp-Session-Id: abc123
id: 1data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":25}}
id: 2 data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":50}}
id: 3data: {"jsonrpc":"2.0","id":1,"result":{"content":[...]}}
复制代码
接收服务器消息(GET) :
GET /mcp HTTP/1.1Host: localhost:8080Accept: text/event-streamMcp-Session-Id: abc123Last-Event-ID: 42
复制代码
会话管理:
# 服务器端设置会话ID@app.post("/mcp")async def handle_mcp(request): if request.method == "initialize": session_id = generate_session_id() return Response( content=json.dumps(result), headers={"Mcp-Session-Id": session_id} )
# 客户端后续请求携带会话ID@client.requestasync def send_request(method, params): headers = {} if self.session_id: headers["Mcp-Session-Id"] = self.session_id return await http.post( "/mcp", json={"jsonrpc": "2.0", "method": method, "params": params}, headers=headers )
复制代码
断线重连:
async def connect_sse(last_event_id=None): headers = {"Accept": "text/event-stream"} if last_event_id: headers["Last-Event-ID"] = last_event_id async with httpx.stream("GET", "/mcp", headers=headers) as stream: async for line in stream.aiter_lines(): if line.startswith("id:"): last_event_id = line[3:].strip() elif line.startswith("data:"): data = json.loads(line[5:]) yield data, last_event_id
复制代码
4.5 实际通信示例
让我们看一个完整的 SQLite 查询场景:
// 1. 列出工具客户端 → 服务器:{ "jsonrpc": "2.0", "id": 1, "method": "tools/list"}
服务器 → 客户端:{ "jsonrpc": "2.0", "id": 1, "result": { "tools": [ { "name": "query", "description": "执行SQL查询", "inputSchema": { "type": "object", "properties": { "sql": {"type": "string"} }, "required": ["sql"] } } ] }}
// 2. 调用查询工具客户端 → 服务器:{ "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "query", "arguments": { "sql": "SELECT COUNT(*) FROM users WHERE active = 1" }, "_meta": { "progressToken": "query-123" // 请求进度通知 } }}
// 3. 服务器发送进度(异步通知)服务器 → 客户端:{ "jsonrpc": "2.0", "method": "notifications/progress", "params": { "progressToken": "query-123", "progress": 50, "total": 100, "message": "正在扫描users表..." }}
// 4. 返回查询结果服务器 → 客户端:{ "jsonrpc": "2.0", "id": 2, "result": { "content": [ { "type": "text", "text": "查询结果: 1,234个活跃用户" } ], "isError": false }}
// 5. 如果查询出错服务器 → 客户端(错误情况):{ "jsonrpc": "2.0", "id": 2, "error": { "code": -32603, "message": "SQL语法错误", "data": { "sql": "SELECT COUNT(*) FROM users WHERE active = 1", "error": "near "WHERE": syntax error", "position": 35 } }}
复制代码
这就是 MCP 通信的完整过程!通过 JSON-RPC 2.0，客户端和服务器可以进行结构化、类型安全的通信。
五、服务器能力：三种核心功能
MCP 服务器可以提供三种功能。
5.1 Resources(资源)：应用决定用什么
资源就是数据，比如文件内容、数据库记录、API 响应。
谁控制: 应用程序决定把哪些资源提供给 AI
如何使用:
// 列出所有可用资源{"method": "resources/list"}
// 读取某个资源{ "method": "resources/read", "params": {"uri": "file:///project/main.py"}}
复制代码
资源 URI 示例:
file:///project/src/main.py - 文件
db://schema/users - 数据库表结构
git://commits/main - Git 提交历史
https://api.example.com/data - Web API
订阅变更: 可以订阅资源,当它变化时自动收到通知。
实际案例: Filesystem 服务器暴露资源
{ "uri": "file:///Users/alice/project/src/main.py", // Python源文件 "name": "main.py", // 文件名 "mimeType": "text/x-python", // 文件类型 "text": "import os\ndef main()..." // 文件内容}
复制代码
客户端 AI 可以读取这个资源，理解代码结构后提供重构建议或生成测试。
5.2 Prompts(提示模板)：用户选择用什么
什么是 Prompt?
Prompt 就像是「对话模板」或「快捷指令」，把常用的复杂指令预设好，用户一键调用。用生活中的例子类比，就像微信的「快捷回复」或 IDE 中的「代码片段(Snippet)」。
为什么需要 Prompt?
场景 1:没有 Prompt 时
用户每次都要输入:"请分析这个Git仓库最近一周的提交,统计:1. 总提交次数2. 每个作者的贡献3. 修改的主要文件4. 是否有破坏性变更请用表格格式输出"
复制代码
场景 2:有 Prompt 后
用户只需:1. 点击 "/analyze-commits" 命令2. 选择分支 "main"3. AI自动执行完整分析
复制代码
Prompt 的数据结构
定义一个 Prompt:
{ "name": "analyze_commits", // Prompt的唯一标识 "title": "提交历史分析", // 用户界面显示的名称 "description": "分析Git提交并生成报告", // 功能说明 "arguments": [ // 需要的参数列表 { "name": "branch", // 参数名 "description": "要分析的分支名", // 参数说明 "required": true // 是否必填 }, { "name": "since", // 时间范围 "description": "起始日期(如:7 days ago)", "required": false // 可选参数 }, { "name": "author", // 作者过滤 "description": "只看某个作者的提交", "required": false } ]}
复制代码
实际使用示例
步骤 1: 列出所有可用的 Prompt
// 客户端请求{ "jsonrpc": "2.0", "id": 1, "method": "prompts/list"}
// 服务器响应{ "jsonrpc": "2.0", "id": 1, "result": { "prompts": [ { "name": "analyze_commits", "title": "📊 提交历史分析", "description": "分析指定分支的提交历史,生成统计报告", "arguments": [ {"name": "branch", "required": true}, {"name": "since", "required": false} ] }, { "name": "review_code", "title": "🔍 代码审查", "description": "对代码进行质量审查和改进建议", "arguments": [ {"name": "file_path", "required": true}, {"name": "focus", "required": false} ] }, { "name": "explain_error", "title": "🐛 错误诊断", "description": "解释错误信息并提供修复建议", "arguments": [ {"name": "error_message", "required": true} ] } ] }}
复制代码
步骤 2: 用户在界面上看到这些选项
━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 可用命令: 📊 /analyze-commits 分析指定分支的提交历史 🔍 /review-code 对代码进行质量审查 🐛 /explain-error 解释错误信息并修复━━━━━━━━━━━━━━━━━━━━━━━━━━━━
复制代码
步骤 3: 用户选择并填写参数
用户输入: /analyze-commits
系统弹窗:┌─────────────────────────┐│ 提交历史分析 │├─────────────────────────┤│ 分支名 *: [main ] ││ 时间范围: [7 days ago] ││ 作者: [ ] ││ ││ [取消] [确定] │└─────────────────────────┘
复制代码
步骤 4: 获取完整的 Prompt 内容
// 客户端请求{ "jsonrpc": "2.0", "id": 2, "method": "prompts/get", "params": { "name": "analyze_commits", // 使用哪个模板 "arguments": { // 用户填写的参数 "branch": "main", "since": "7 days ago" } }}
// 服务器响应 - 返回完整的对话消息{ "jsonrpc": "2.0", "id": 2, "result": { "description": "分析main分支最近7天的提交", "messages": [ // 发送给AI的完整对话 { "role": "user", // 用户角色 "content": { "type": "text", "text": "请分析main分支在过去7天的Git提交历史。\n\n需要统计:\n1. 总提交次数\n2. 每个作者的贡献次数和代码行数\n3. 主要修改的文件列表\n4. 是否包含breaking changes\n5. 提交消息的规范性\n\n请用表格格式输出结果,并在最后给出改进建议。" } } ] }}
复制代码
步骤 5: 客户端将消息发送给 AI
用户看到AI正在分析...
AI返回:━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━📊 main分支提交分析报告(最近7天)
总提交次数: 23
作者贡献统计:┌────────────┬────────┬──────────┐│ 作者 │ 提交数 │ 代码行数 │├────────────┼────────┼──────────┤│ Alice │ 12 │ +543/-89││ Bob │ 8 │ +234/-45││ Charlie │ 3 │ +123/-12│└────────────┴────────┴──────────┘
主要修改文件:- src/api/users.py (8次修改)- src/models/user.py (5次修改)- tests/test_user.py (4次修改)
Breaking Changes: 无
提交规范性: 良好 (91%符合Conventional Commits)
改进建议:1. 建议增加单元测试覆盖率2. 部分提交消息过于简短━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
复制代码
Prompt 的内容类型
Prompt 消息中可以包含多种内容:
1. 纯文本
{ "role": "user", "content": { "type": "text", "text": "请审查这段代码..." }}
复制代码
2. 嵌入图片
{ "role": "user", "content": { "type": "image", "data": "base64-encoded-image-data", // 图片数据 "mimeType": "image/png" // 图片类型 }}
复制代码
3. 嵌入资源(引用 MCP 资源)
{ "role": "user", "content": { "type": "resource", "resource": { "uri": "file:///project/src/user.py", // 资源URI "mimeType": "text/x-python", "text": "class User:\n def __init__..." // 资源内容 } }}
复制代码
4. 多轮对话
{ "messages": [ { "role": "user", "content": {"type": "text", "text": "我想优化这段代码"} }, { "role": "assistant", // AI的回复 "content": {"type": "text", "text": "请提供代码内容"} }, { "role": "user", "content": { "type": "resource", // 用户提供代码 "resource": {...} } } ]}
复制代码
Prompt vs Tool vs Resource 对比
特性 Prompt Tool Resource━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━控制者 用户主动选择 AI自动决定 应用程序控制
触发方式 用户点击命令 AI判断需要调用 应用自动附加 /analyze 或用户选择 返回内容 对话消息 执行结果 数据内容 (给AI的指令) (函数返回值) (上下文信息) 典型用途 工作流模板 执行操作 提供背景信息 快捷指令 查询数据 文件内容 示例 代码审查模板 执行SQL查询 项目README 错误诊断向导 发送邮件 数据库Schema 用户感知 ✅ 明显 ❓ 可能不知道 ❓ 透明的 (用户点击) (AI决定) (自动加载)
复制代码
Prompt 是预设的对话模板，通过参数化实现灵活应用，提升用户体验，并能与 MCP 其他能力组合形成完整工作流。
代码实现示例
# 服务器端:注册Prompt@server.list_prompts()async def list_prompts(): return [ Prompt( name="analyze_commits", title="📊 提交历史分析", description="分析Git提交历史并生成统计报告", arguments=[ {"name": "branch", "description": "分支名", "required": True}, {"name": "since", "description": "时间范围", "required": False} ] ) ]
@server.get_prompt()async def get_prompt(name: str, arguments: dict): if name == "analyze_commits": branch = arguments["branch"] since = arguments.get("since", "7 days ago") # 构建完整的提示消息 prompt_text = f"""请分析{branch}分支在{since}的Git提交历史。
需要统计:1. 总提交次数2. 每个作者的贡献3. 主要修改的文件4. 是否有breaking changes
请用表格格式输出。 """ return { "messages": [ { "role": "user", "content": {"type": "text", "text": prompt_text} } ] }
# 客户端:使用Promptasync def use_prompt(session, prompt_name, arguments): # 获取Prompt内容 prompt = await session.get_prompt( name=prompt_name, arguments=arguments ) # 将消息发送给AI for message in prompt.messages: ai_response = await send_to_ai(message) print(ai_response)
复制代码
5.3 Tools(工具)：AI 自己决定用什么
Tool 就是可执行的函数，比如查询数据库、调用 API、写文件。
谁控制：AI 模型根据对话内容自己决定调用哪个工具
如何使用:
// 列出可用工具{"method": "tools/list"}
// AI调用工具{ "method": "tools/call", "params": { "name": "get_weather", "arguments": {"city": "北京"} }}
复制代码
返回结果:
{ "content": [{ "type": "text", "text": "北京天气:晴,温度22°C" }], "isError": false}
复制代码
5.4 其他功能
补全
MCP 提供标准化的参数自动补全功能，支持为提示和资源 URI 提供上下文相关的建议，实现类似 IDE 的交互体验。服务器通过声明completions能力，支持对ref/prompt和ref/resource两种引用类型的补全，每次最多返回 100 个按相关性排序的建议值，并可通过completion/complete请求获取补全结果。
日志
MCP 提供结构化日志消息传递机制，允许服务器向客户端发送包含严重性级别、可选记录器名称和任意 JSON 可序列化数据的日志通知。服务器需声明logging能力，支持遵循 RFC 5424 标准的日志级别（从 debug 到 emergency），客户端可通过logging/setLevel请求配置最低日志级别，服务器通过notifications/message通知发送日志消息。
分页
MCP 支持对可能返回大量结果集的列表操作进行分页处理，使用基于不透明游标的分页模型而非数字页码。服务器在响应中包含当前页结果和可选的nextCursor字段（表示更多结果存在），客户端可通过在请求中包含游标继续分页。支持分页的操作包括resources/list、resources/templates/list、prompts/list和tools/list，客户端必须将游标视为不透明令牌。
六、客户端能力：反向请求
客户端不仅接收服务器的数据，也可以提供能力给服务器使用:
6.1 Sampling(采样)：服务器请求客户端调用 AI
场景: 服务器在处理任务时，需要 AI 帮忙分析中间结果。
如何使用:
{ "method": "sampling/createMessage", "params": { "messages": [{ "role": "user", "content": {"type": "text", "text": "这个数据正常吗?"} }], "modelPreferences": { "hints": [{"name": "claude-3-sonnet"}], // 建议用的模型 "intelligencePriority": 0.8, // 要求智能程度 "speedPriority": 0.5 // 速度要求 } }}
复制代码
实际案例:Filesystem 服务器在搜索大量文件时,请求 AI 判断哪些文件最相关。
6.2 Roots(目录)：告诉服务器工作范围
场景: 让服务器知道可以访问哪些目录。
如何使用:
{ "method": "roots/list"}
复制代码
返回:
{ "roots": [{ "uri": "file:///home/user/project", "name": "我的项目" }]}
复制代码
服务器知道只能在这个目录里操作，保护其他文件安全。
6.3 Elicitation(引导)：服务器向用户询问信息
场景: 服务器需要用户提供额外信息才能继续。
如何使用:
{ "method": "elicitation/create", "params": { "message": "请提供您的GitHub用户名", "requestedSchema": { "type": "object", "properties": { "username": {"type": "string"} } } }}
复制代码
用户响应:
{ "action": "accept", // 或"decline"拒绝, "cancel"取消 "content": { "username": "octocat" }}
复制代码
实际案例: Git 服务器需要知道提交信息格式，弹窗问用户:"请选择提交规范:Conventional Commits/Angular/Custom?"
七、完整实战：从零构建天气查询 MCP
下面让我们从头到尾构建一个完整的 MCP 系统，包含服务器和客户端。
7.1 需求分析
目标: 构建一个天气查询 MCP 服务器,提供:
资源: 城市列表
工具: 查询天气、获取预报
提示: 天气分析模板
7.2 服务器实现(Python)
第一步: 安装 MCP SDK
第二步: 创建服务器 (weather_server.py)
from mcp.server import Serverfrom mcp.types import Resource, Tool, Prompt, TextContentimport mcp.server.stdioimport httpx
# 创建MCP服务器实例server = Server("weather-server")
# 1. 定义资源:支持的城市列表@server.list_resources()async def list_resources(): """返回可用的资源列表""" return [ Resource( uri="weather://cities", name="支持的城市列表", mimeType="application/json", description="查询天气支持的所有城市" ) ]
@server.read_resource()async def read_resource(uri: str): """读取具体资源内容""" if uri == "weather://cities": cities = ["北京", "上海", "广州", "深圳", "杭州"] return { "contents": [{ "uri": uri, "mimeType": "application/json", "text": str(cities) }] }
# 2. 定义工具:天气查询@server.list_tools()async def list_tools(): """返回可用的工具列表""" return [ Tool( name="get_current_weather", description="获取指定城市的当前天气", inputSchema={ "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,如'北京'" } }, "required": ["city"] } ), Tool( name="get_forecast", description="获取未来3天天气预报", inputSchema={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"}, "days": {"type": "number", "description": "预报天数(1-3)", "default": 3} }, "required": ["city"] } ) ]
@server.call_tool()async def call_tool(name: str, arguments: dict): """执行工具调用""" if name == "get_current_weather": city = arguments["city"] # 实际项目中这里调用真实的天气API # 示例:使用模拟数据 weather_data = { "city": city, "temperature": 22, "condition": "晴", "humidity": 45 } return { "content": [{ "type": "text", "text": f"{city}当前天气:\n温度: {weather_data['temperature']}°C\n天气: {weather_data['condition']}\n湿度: {weather_data['humidity']}%" }] } elif name == "get_forecast": city = arguments["city"] days = arguments.get("days", 3) # 模拟预报数据 forecast = f"{city}未来{days}天预报:\n第1天: 晴,20-25°C\n第2天: 多云,18-23°C\n第3天: 小雨,16-20°C" return { "content": [{"type": "text", "text": forecast}] }
# 3. 定义提示模板:天气分析@server.list_prompts()async def list_prompts(): """返回可用的提示模板""" return [ Prompt( name="analyze_weather", description="分析天气趋势并给出建议", arguments=[ {"name": "city", "description": "城市名称", "required": True} ] ) ]
@server.get_prompt()async def get_prompt(name: str, arguments: dict): """获取提示模板内容""" if name == "analyze_weather": city = arguments["city"] return { "messages": [ { "role": "user", "content": { "type": "text", "text": f"请分析{city}的天气情况,并给出出行建议。包括:\n1. 温度是否适宜\n2. 是否需要带伞\n3. 穿衣建议" } } ] }
# 启动服务器if __name__ == "__main__": # 使用stdio传输(本地) mcp.server.stdio.run_stdio_server(server)
复制代码
7.3 配置服务器(Claude Desktop)
创建配置文件 ~/Library/Application Support/Claude/claude_desktop_config.json:
{ "mcpServers": { "weather": { "command": "python", "args": ["/path/to/weather_server.py"] } }}
复制代码
7.4 客户端实现(Python)
如果要自己实现客户端:
from mcp import ClientSessionfrom mcp.client.stdio import stdio_clientimport asyncio
async def main(): # 连接到服务器 async with stdio_client( command="python", args=["/path/to/weather_server.py"] ) as (read, write): async with ClientSession(read, write) as session: # 1. 初始化连接 await session.initialize() print("✅ 连接成功!") # 2. 列出可用资源 resources = await session.list_resources() print(f"\n📁 可用资源: {len(resources.resources)}") for r in resources.resources: print(f" - {r.name}: {r.uri}") # 3. 读取城市列表资源 cities_resource = await session.read_resource( uri="weather://cities" ) print(f"\n🌍 城市列表: {cities_resource.contents[0].text}") # 4. 列出可用工具 tools = await session.list_tools() print(f"\n🔧 可用工具: {len(tools.tools)}") for t in tools.tools: print(f" - {t.name}: {t.description}") # 5. 调用工具查询天气 result = await session.call_tool( name="get_current_weather", arguments={"city": "北京"} ) print(f"\n🌤️ 查询结果:\n{result.content[0].text}") # 6. 获取预报 forecast = await session.call_tool( name="get_forecast", arguments={"city": "上海", "days": 3} ) print(f"\n📅 天气预报:\n{forecast.content[0].text}") # 7. 列出提示模板 prompts = await session.list_prompts() print(f"\n💡 提示模板: {len(prompts.prompts)}") for p in prompts.prompts: print(f" - {p.name}: {p.description}") # 8. 获取提示内容 prompt = await session.get_prompt( name="analyze_weather", arguments={"city": "广州"} ) print(f"\n📝 生成的提示:\n{prompt.messages[0]['content']['text']}")
if __name__ == "__main__": asyncio.run(main())
复制代码
7.5 运行效果
$ python weather_client.py
✅ 连接成功!
📁 可用资源: 1 - 支持的城市列表: weather://cities
🌍 城市列表: ['北京', '上海', '广州', '深圳', '杭州']
🔧 可用工具: 2 - get_current_weather: 获取指定城市的当前天气 - get_forecast: 获取未来3天天气预报
🌤️ 查询结果:北京当前天气:温度: 22°C天气: 晴湿度: 45%
📅 天气预报:上海未来3天预报:第1天: 晴,20-25°C第2天: 多云,18-23°C第3天: 小雨,16-20°C
💡 提示模板: 1 - analyze_weather: 分析天气趋势并给出建议
📝 生成的提示:请分析广州的天气情况,并给出出行建议。包括:1. 温度是否适宜2. 是否需要带伞3. 穿衣建议
复制代码
八、其他部分：MCP 基础协议的另一半
8.1 授权（Authorization）
MCP 授权规范定义了基于 HTTP 传输的安全授权机制，使 MCP 客户端能够代表资源所有者向受限制的 MCP 服务器发出请求。该规范基于 OAuth 2.1 及相关标准，实现了授权服务器发现、动态客户端注册和访问令牌管理。例如，客户端通过resource参数明确指定目标 MCP 服务器（如https://mcp.example.com），服务器则验证令牌是否专门为其颁发，确保令牌不会被误用于其他服务，从而防止"令牌传递"安全漏洞。
8.2 取消（Cancellation）
MCP 取消机制允许通过通知消息中止正在进行的请求，任何一方都可以发送notifications/cancelled通知来终止先前发出的请求。例如，当用户取消长时间运行的操作时，客户端可以发送包含请求 ID 和可选原因的取消通知，接收方应停止处理、释放资源且不发送响应。该机制考虑了网络延迟导致的竞态条件，允许接收方在请求已完成或无法取消时忽略通知，同时建议双方记录取消原因以便调试。
{ "method": "notifications/cancelled", "params": { "requestId": "123", "reason": "用户取消" }}
复制代码
8.3 Ping 机制
MCP 提供了可选的 ping 机制，允许任何一方验证对方是否仍然响应且连接存活。该机制通过简单的请求/响应模式实现，例如客户端发送{"jsonrpc":"2.0","id":"123","method":"ping"}，服务器必须立即响应{"jsonrpc":"2.0","id":"123","result":{}}。如果在合理超时时间内未收到响应，发送方可以将连接视为陈旧并终止连接或尝试重新连接。实现应定期发送 ping 以检测连接健康状况，但应避免过度 ping 以减少网络开销。
8.4 进度跟踪（Progress）
MCP 支持通过通知消息对长时间运行的操作进行可选的进度跟踪。请求方可以在请求元数据中包含唯一的progressToken（如字符串"task123"）来接收进度更新，接收方则可以发送包含进度值、可选总值和消息的notifications/progress通知。例如，文件上传操作可以发送{"progress":50,"total":100,"message":"正在上传文件..."}来指示完成百分比。进度值必须随每个通知递增，双方应实现速率限制以防止消息泛滥，并在操作完成后停止发送进度通知。
{ "method": "notifications/progress", "params": { "progressToken": "task123", "progress": 50, // 当前进度 "total": 100, // 总量 "message": "正在上传文件..." }}
复制代码
九、安全实践：必须重视
9.1 核心原则
1. 用户同意优先
所有数据访问必须经用户明确同意
所有工具调用前必须让用户确认
2. 数据隐私保护
服务器只能看到必要的信息
完整对话历史保留在宿主,不发给服务器
3. 工具安全
工具代表代码执行,必须谨慎
显示工具要做什么,让用户批准
4. 输入验证
服务器必须验证所有输入
客户端必须验证工具返回的结果
9.2 实际建议
服务器开发者:
验证所有输入参数
实现访问控制和速率限制
记录操作日志供审计
客户端开发者:
显示清晰的权限请求界面
在调用工具前展示参数
实现工具调用超时机制
十、MCP 生态：谁开发客户端?
关键认知: 在 MCP 生态中，客户端通常不是由下游开发者开发的，而是内置在 AI 应用平台中。
开发者开发MCP服务器 ↓ 配置到AI平台(Claude/Cursor等) ↓ AI平台内置的MCP客户端自动连接
复制代码
对于软件开发者来说，在 MCP 生态中的位置如下。
角色定位:┌─────────────────────────────────────────┐│ AI平台开发者(Anthropic, Cursor等) ││ ──────────────────────────────── ││ 职责: ││ ✅ 开发MCP客户端SDK ││ ✅ 集成到自己的AI应用中 ││ ✅ 提供配置界面 ││ ✅ 管理MCP服务器生命周期 ││ ✅ 处理AI与MCP的交互逻辑 │└─────────────────────────────────────────┘ ↓ 提供平台┌─────────────────────────────────────────┐│ MCP服务器开发者(你、我、社区) ││ ──────────────────────────────── ││ 职责: ││ ✅ 开发MCP服务器 ││ ✅ 实现Resources/Tools/Prompts ││ ✅ 编写使用文档 ││ ✅ 发布到npm/PyPI ││ ❌ 不需要开发客户端 ││ ❌ 不需要关心AI如何调用 │└─────────────────────────────────────────┘ ↓ 使用服务┌─────────────────────────────────────────┐│ 最终用户(开发者、分析师等) ││ ──────────────────────────────── ││ 职责: ││ ✅ 安装需要的MCP服务器 ││ ✅ 配置到AI平台 ││ ✅ 使用AI完成任务 ││ ❌ 不需要写代码 │└─────────────────────────────────────────┘
复制代码
后记
MCP 让 AI 应用开发变得更简单、更安全、更强大。它不是银弹，但为构建可靠的 AI 系统提供了坚实基础。本文全部内容基于提示编写，欢迎交流讨论！
参考文献
MCP 官方规范: https://modelcontextprotocol.io/specification/2025-06-18
JSON-RPC 2.0: https://www.jsonrpc.org/
评论