一、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协议一样:

核心层(所有实现必须支持):

• JSON-RPC 2.0消息格式

• 初始化握手流程(initialize/initialized)

• 基本错误处理

功能层(按需选择):

• Resources、Prompts、Tools(服务器端)

• Roots、Sampling、Elicitation(客户端)

这样设计的好处:

统一的基础协议 → 保证互操作性
     +
灵活的功能选择 → 满足不同场景需求
     ↓
既标准化又可扩展

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规范定义了标准方法名:

功能	方法名	说明
列出资源	resources/list	固定方法名
读取资源	resources/read	固定方法名
列出工具	tools/list	固定方法名
调用工具	tools/call	固定方法名
列出提示	prompts/list	固定方法名
获取提示	prompts/get	固定方法名

步骤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的标准错误码:

错误码	含义	说明
-32700	Parse error	JSON解析错误
-32600	Invalid Request	无效的请求格式
-32601	Method not found	方法不存在
-32602	Invalid params	参数无效
-32603	Internal error	服务器内部错误
-32002	Resource not found	资源未找到(MCP扩展)

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传输详解

优点:

• ✅ 简单直接,适合本地开发

• ✅ 进程隔离,安全性好

• ✅ 自动管理生命周期

• ✅ 无需网络配置

缺点:

• ❌ 只能本地使用

• ❌ 不支持多客户端

• ❌ 调试相对困难

消息格式:

消息1\n
消息2\n
消息3\n

每个JSON对象占一行,以\n分隔。

HTTP传输详解

架构:

┌─────────┐         HTTP POST         ┌─────────┐
│         ├──────────────────────────>│         │
│ 客户端  │  请求/通知/响应(JSON-RPC) │ 服务器  │
│         │<──────────────────────────┤         │
└─────────┘     HTTP 响应/SSE流       └─────────┘
             (application/json 或
              text/event-stream)

发送消息(POST):

POST /mcp HTTP/1.1
Host: localhost:8080
Content-Type: application/json
Accept: application/json, text/event-stream
Mcp-Session-Id: abc123

{"jsonrpc":"2.0","id":1,"method":"tools/list"}

立即响应(JSON):

HTTP/1.1 200 OK
Content-Type: application/json

{"jsonrpc":"2.0","id":1,"result":{"tools":[...]}}

流式响应(SSE):

HTTP/1.1 200 OK
Content-Type: text/event-stream
Mcp-Session-Id: abc123

id: 1
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":25}}

id: 2  
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":50}}

id: 3
data: {"jsonrpc":"2.0","id":1,"result":{"content":[...]}}

接收服务器消息(GET):

GET /mcp HTTP/1.1
Host: localhost:8080
Accept: text/event-stream
Mcp-Session-Id: abc123
Last-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.request
async 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}
                }
            ]
        }

# 客户端:使用Prompt
async 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

pip install mcp

第二步: 创建服务器 (weather_server.py)

from mcp.server import Server
from mcp.types import Resource, Tool, Prompt, TextContent
import mcp.server.stdio
import 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 ClientSession
from mcp.client.stdio import stdio_client
import 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系统提供了坚实基础。本文全部内容基于提示编写，欢迎交流讨论！

参考文献

1. MCP官方规范: https://modelcontextprotocol.io/specification/2025-06-18

2. JSON-RPC 2.0: https://www.jsonrpc.org/