MCP协议实战：构建AI智能体的万能工具箱

2026-06-16 9101 words 19 minutes

Contents

引言：AI Agent 的"USB-C 时刻"

如果你经历过移动设备接口统一前的混乱时代——Lightning、Micro-USB、Mini-USB 各自为政——你一定理解统一接口的价值。USB-C 出现后，一根线解决了所有问题。

MCP（Model Context Protocol）就是 AI Agent 世界的 USB-C。

2024年11月，Anthropic 发布了 MCP 协议，为大语言模型（LLM）连接外部工具和数据源提供了一个标准化的、开放的通信协议。在此之前，每个 AI 应用都要为每个外部服务单独编写集成代码，形成了 N 个 AI 应用 × M 个外部服务的 N×M 复杂度。MCP 将这个复杂度降低为 N+M——每个 AI 应用实现一次 MCP 客户端，每个外部服务实现一次 MCP Server。

本文将从架构原理到实战代码，带你完整走一遍 MCP 的开发全流程。

一、MCP 是什么？

1.1 核心定义

MCP（Model Context Protocol）是一个基于 JSON-RPC 2.0 的开放协议，定义了 AI 模型（或 AI 应用）与外部工具、数据源之间的标准通信方式。它由 Anthropic 主导开发，但设计为厂商无关（vendor-neutral）。

一句话概括：MCP 让 AI Agent 能够以标准化的方式发现、调用和管理外部工具与数据。

1.2 它解决了什么问题？

在 MCP 之前，AI 应用连接外部工具的方式大致如下：

Function Calling：OpenAI、Anthropic 等模型厂商各自定义了函数调用格式，开发者需要针对每个模型 API 适配
自定义集成：每个 AI 应用自己编写与外部服务的对接代码，重复造轮子
缺乏状态管理：一次性的函数调用无法维护与外部服务的持久连接
无法共享上下文：多个 AI 应用无法共享对同一工具的访问和上下文

MCP 通过统一协议解决所有这些问题。

1.3 一个类比帮助理解

概念	类比
LLM	你的电脑
MCP 协议	USB 协议
MCP Client	USB 控制器（电脑端）
MCP Server	USB 设备（如键盘、摄像头）
Tools	设备的功能（打字、拍照）
Resources	设备提供的数据（文件、数据库记录）
Prompts	设备的操作模板（预设的使用方式）

二、MCP 架构深度解析

2.1 三层架构

MCP 采用经典的三层架构：Host → Client → Server

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


┌─────────────────────────────────────────┐
│                 Host                     │
│  (AI 应用，如 Claude Desktop)            │
│                                         │
│  ┌──────────┐  ┌──────────┐            │
│  │ MCP      │  │ MCP      │            │
│  │ Client A │  │ Client B │            │
│  └────┬─────┘  └────┬─────┘            │
│       │              │                  │
└───────┼──────────────┼──────────────────┘
        │              │
   ┌────▼─────┐  ┌─────▼────┐
   │ MCP      │  │ MCP      │
   │ Server 1 │  │ Server 2 │
   │(文件系统) │  │(数据库)  │
   └──────────┘  └──────────┘

Host（宿主）：运行 AI 模型的应用程序，如 Claude Desktop、VS Code + Continue.dev 等。Host 负责管理多个 MCP Client 实例。
Client（客户端）：MCP 协议的客户端实例，与特定的 MCP Server 维持一对一连接。Client 负责处理协议握手、能力协商和消息路由。
Server（服务端）：暴露工具、资源和提示模板的服务进程。每个 MCP Server 专注于一个特定领域（如文件操作、数据库查询、API 调用等）。

2.2 通信协议

MCP 使用 JSON-RPC 2.0 作为消息格式，支持两种传输机制：

stdio（标准输入/输出）

最常用的本地通信方式。Client 通过启动一个子进程来运行 MCP Server，通过 stdin/stdout 进行通信：

1
2
3
4
5


┌────────────┐   stdin    ┌────────────┐
│  MCP       │ ──────────>│  MCP       │
│  Client    │<────────── │  Server    │
│            │   stdout   │  (子进程)   │
└────────────┘            └────────────┘

优点：无需网络配置，安全性好，适合本地工具。

SSE/HTTP（Server-Sent Events）

用于远程 MCP Server 的通信方式，基于 HTTP 协议：

1
2
3
4
5


┌────────────┐  HTTP POST  ┌────────────┐
│  MCP       │ ──────────>│  MCP       │
│  Client    │<────────── │  Server    │
│            │  SSE 流     │  (远程)     │
└────────────┘            └────────────┘

优点：支持远程部署，可跨越网络边界，适合云服务场景。

2.3 协议握手流程

MCP 连接建立时，Client 和 Server 会进行能力协商：

⚠️ 注：以下示例中的 protocolVersion: "2025-03-26" 为 MCP 规范中的日期格式版本号。实际版本号请以官方 GitHub 仓库为准，规范可能随时间更新。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41


// 1. Client → Server: 初始化请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "roots": { "listChanged": true },
      "sampling": {}
    },
    "clientInfo": {
      "name": "my-ai-app",
      "version": "1.0.0"
    }
  }
}

// 2. Server → Client: 初始化响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "tools": { "listChanged": true },
      "resources": { "subscribe": true },
      "prompts": { "listChanged": true }
    },
    "serverInfo": {
      "name": "my-mcp-server",
      "version": "1.0.0"
    }
  }
}

// 3. Client → Server: 确认初始化完成
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

握手完成后，Client 可以查询 Server 提供的工具列表，然后 AI 模型就可以根据对话内容决定调用哪些工具。

三、构建你的第一个 MCP Server（Python）

3.1 环境准备

确保你已安装 Python 3.10+，然后安装官方 MCP SDK：

1
2
3
4
5
6
7


# 创建项目
mkdir my-mcp-server && cd my-mcp-server
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装 MCP SDK
pip install mcp

3.2 一个最小的 MCP Server

让我们创建一个简单的天气查询 MCP Server：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122


# weather_server.py
import json
import httpx
from mcp.server.fastmcp import FastMCP

# 创建 MCP Server 实例
mcp = FastMCP(
    name="weather-server",
    version="1.0.0",
    description="一个查询天气信息的 MCP Server"
)


@mcp.tool()
async def get_weather(city: str) -> str:
    """查询指定城市的当前天气信息。
    
    Args:
        city: 城市名称，如 "北京"、"Shanghai"
    
    Returns:
        天气信息的 JSON 字符串
    """
    # 这里使用一个示例 API，实际项目中替换为真实天气 API
    async with httpx.AsyncClient() as client:
        # 示例：使用 wttr.in 天气服务
        response = await client.get(
            f"https://wttr.in/{city}?format=j1",
            timeout=10.0
        )
        if response.status_code == 200:
            data = response.json()
            current = data.get("current_condition", [{}])[0]
            result = {
                "city": city,
                "temperature": current.get("temp_C", "N/A"),
                "description": current.get("weatherDesc", [{}])[0].get("value", "N/A"),
                "humidity": current.get("humidity", "N/A"),
                "wind_speed": current.get("windspeedKmph", "N/A"),
            }
            return json.dumps(result, ensure_ascii=False)
        else:
            return json.dumps({"error": f"无法获取 {city} 的天气信息"}, ensure_ascii=False)


@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
    """查询指定城市未来几天的天气预报。
    
    Args:
        city: 城市名称
        days: 预报天数，默认3天，最多7天
    
    Returns:
        天气预报的 JSON 字符串
    """
    days = min(max(days, 1), 7)  # 限制在1-7天之间
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}?format=j1",
            timeout=10.0
        )
        if response.status_code == 200:
            data = response.json()
            forecasts = []
            for day_data in data.get("weather", [])[:days]:
                forecasts.append({
                    "date": day_data.get("date"),
                    "max_temp": day_data.get("maxtempC"),
                    "min_temp": day_data.get("mintempC"),
                    "description": day_data.get("hourly", [{}])[4].get(
                        "weatherDesc", [{}]
                    )[0].get("value", "N/A"),
                })
            return json.dumps({
                "city": city,
                "forecasts": forecasts
            }, ensure_ascii=False)
        else:
            return json.dumps({"error": f"无法获取 {city} 的预报信息"}, ensure_ascii=False)


# 定义资源：提供天气 API 文档
@mcp.resource("weather://docs")
def get_api_docs() -> str:
    """获取天气 API 的使用文档。"""
    return """
天气查询 MCP Server API 文档
============================

可用工具：
1. get_weather(city) - 查询当前天气
2. get_forecast(city, days) - 查询天气预报

支持的城市名称格式：
- 中文：北京、上海、广州
- 英文：Beijing, Shanghai, Guangzhou
- 也可以用拼音：Beijing
"""


# 定义提示模板
@mcp.prompt()
def weather_report(city: str, language: str = "中文") -> str:
    """生成天气报告的提示模板。
    
    Args:
        city: 城市名称
        language: 报告语言
    """
    return f"""请使用 get_weather 和 get_forecast 工具获取 {city} 的天气信息，
然后用{language}为我生成一份详细的天气报告，包括：
1. 当前天气概况
2. 未来几天的天气趋势
3. 穿衣和出行建议


请确保报告内容准确、实用。"""


if __name__ == "__main__":
    mcp.run(transport="stdio")

3.3 运行与测试

首先确保安装了所有依赖：

1

pip install mcp httpx

手动测试 MCP Server：

1
2


# 直接运行（会监听 stdin/stdout）
python weather_server.py

四、构建一个更复杂的 MCP Server（TypeScript）

在实际项目中，TypeScript 的 MCP SDK 同样成熟。让我们构建一个数据库查询 MCP Server：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218


// db-server/src/index.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { Pool } from "pg";

// 数据库连接池
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 10,
  idleTimeoutMillis: 30000,
});

// 创建 MCP Server
const server = new McpServer({
  name: "postgres-query-server",
  version: "1.0.0",
  capabilities: {
    tools: {},
    resources: {},
  },
});

// 工具1：执行 SQL 查询
server.tool(
  "query",
  "执行一个只读的 SQL SELECT 查询",
  {
    sql: z.string().describe("要执行的 SQL SELECT 语句"),
    params: z
      .array(z.string())
      .optional()
      .describe("查询参数（防止 SQL 注入）"),
  },
  async ({ sql, params }) => {
    // 安全检查：只允许 SELECT 语句
    const trimmedSql = sql.trim().toUpperCase();
    if (!trimmedSql.startsWith("SELECT")) {
      return {
        content: [
          {
            type: "text",
            text: "错误：只允许执行 SELECT 查询，禁止修改数据。",
          },
        ],
        isError: true,
      };
    }

    try {
      const result = await pool.query(sql, params || []);
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(
              {
                rowCount: result.rowCount,
                fields: result.fields.map((f) => f.name),
                rows: result.rows.slice(0, 100), // 限制返回行数
              },
              null,
              2
            ),
          },
        ],
      };
    } catch (error: any) {
      return {
        content: [
          {
            type: "text",
            text: `查询错误: ${error.message}`,
          },
        ],
        isError: true,
      };
    }
  }
);

// 工具2：获取表结构
server.tool(
  "describe_table",
  "获取指定表的结构信息（列名、类型、约束等）",
  {
    table_name: z.string().describe("要查询的表名"),
  },
  async ({ table_name }) => {
    try {
      const result = await pool.query(
        `
        SELECT 
          column_name,
          data_type,
          is_nullable,
          column_default,
          character_maximum_length
        FROM information_schema.columns
        WHERE table_name = $1
        ORDER BY ordinal_position
      `,
        [table_name]
      );

      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(
              {
                table: table_name,
                columns: result.rows,
              },
              null,
              2
            ),
          },
        ],
      };
    } catch (error: any) {
      return {
        content: [
          {
            type: "text",
            text: `错误: ${error.message}`,
          },
        ],
        isError: true,
      };
    }
  }
);

// 工具3：数据聚合统计
server.tool(
  "aggregate",
  "对指定表的数值列进行聚合统计（count, sum, avg, min, max）",
  {
    table_name: z.string().describe("表名"),
    column: z.string().describe("要聚合的列名"),
    operation: z
      .enum(["count", "sum", "avg", "min", "max"])
      .describe("聚合操作"),
    group_by: z.string().optional().describe("分组列名（可选）"),
  },
  async ({ table_name, column, operation, group_by }) => {
    try {
      let sql: string;
      if (group_by) {
        sql = `SELECT ${group_by}, ${operation}(${column}) as result 
               FROM ${table_name} 
               GROUP BY ${group_by} 
               ORDER BY result DESC 
               LIMIT 50`;
      } else {
        sql = `SELECT ${operation}(${column}) as result FROM ${table_name}`;
      }

      const result = await pool.query(sql);
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(
              {
                table: table_name,
                column,
                operation,
                group_by: group_by || null,
                result: result.rows,
              },
              null,
              2
            ),
          },
        ],
      };
    } catch (error: any) {
      return {
        content: [
          {
            type: "text",
            text: `聚合错误: ${error.message}`,
          },
        ],
        isError: true,
      };
    }
  }
);

// 资源：数据库概览
server.resource(
  "db-schema",
  "database://schema/overview",
  async (uri) => ({
    contents: [
      {
        uri: uri.href,
        mimeType: "text/plain",
        text: `数据库概览信息，请使用 describe_table 工具查看具体表结构。`,
      },
    ],
  })
);

// 启动服务器
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("PostgreSQL MCP Server 已启动");
}

main().catch((error) => {
  console.error("启动失败:", error);
  process.exit(1);
});

对应的 package.json：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


{
  "name": "postgres-mcp-server",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/index.js"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.0.0",
    "pg": "^8.13.0",
    "zod": "^3.23.0"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "@types/pg": "^8.11.0",
    "typescript": "^5.6.0"
  }
}

五、MCP 客户端配置

5.1 Claude Desktop 配置

在 Claude Desktop 的配置文件中注册你的 MCP Server：

macOS：~/Library/Application Support/Claude/claude_desktop_config.json Windows：%APPDATA%\Claude\claude_desktop_config.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/weather_server.py"],
      "env": {
        "API_KEY": "your-api-key"
      }
    },
    "database": {
      "command": "node",
      "args": ["/path/to/dist/index.js"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost:5432/mydb"
      }
    }
  }
}

5.2 Continue.dev 配置

在 ~/.continue/config.json 中添加：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


{
  "mcpServers": [
    {
      "name": "filesystem",
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/you/projects"
      ]
    },
    {
      "name": "github",
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxx"
      }
    }
  ]
}

5.3 自定义 MCP Client（Python）

如果你要构建自己的 AI 应用并集成 MCP：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59


# custom_client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
    # 定义 Server 参数
    server_params = StdioServerParameters(
        command="python",
        args=["/path/to/weather_server.py"],
        env={"API_KEY": "your-key"}
    )
    
    # 连接到 MCP Server
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 初始化连接
            await session.initialize()
            
            # 列出可用工具
            tools = await session.list_tools()
            print("可用工具:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            
            # 列出可用资源
            resources = await session.list_resources()
            print("\n可用资源:")
            for resource in resources.resources:
                print(f"  - {resource.uri}: {resource.name}")
            
            # 列出可用提示
            prompts = await session.list_prompts()
            print("\n可用提示:")
            for prompt in prompts.prompts:
                print(f"  - {prompt.name}: {prompt.description}")
            
            # 调用工具
            result = await session.call_tool(
                "get_weather",
                arguments={"city": "北京"}
            )
            print(f"\n天气查询结果: {result.content[0].text}")
            
            # 读取资源
            weather_docs = await session.read_resource("weather://docs")
            print(f"\nAPI 文档:\n{weather_docs.contents[0].text}")
            
            # 获取提示模板
            prompt_result = await session.get_prompt(
                "weather_report",
                arguments={"city": "上海", "language": "中文"}
            )
            print(f"\n提示模板:\n{prompt_result.messages[0].content.text}")


if __name__ == "__main__":
    asyncio.run(main())

六、MCP 资源和提示模板

6.1 Resources（资源）

Resources 是 MCP 提供的只读数据源，类似于 REST API 中的 GET 端点。它们让 AI 模型能够主动获取上下文信息。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31


from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

# 静态资源
@mcp.resource("config://app")
def get_app_config() -> str:
    """应用配置信息"""
    return json.dumps({
        "version": "1.0.0",
        "environment": "production",
        "features": ["chat", "search", "analysis"]
    }, indent=2)

# 动态资源（带参数）
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """获取用户资料"""
    # 从数据库获取用户信息...
    return json.dumps({
        "id": user_id,
        "name": "张三",
        "role": "admin"
    }, indent=2)

# 目录型资源
@mcp.resource("logs://recent")
def get_recent_logs() -> str:
    """获取最近的日志"""
    return "2026-01-15 10:30:00 - INFO - 服务启动成功\n" \
           "2026-01-15 10:30:01 - DEBUG - 加载配置完成"

6.2 Prompts（提示模板）

Prompts 让 MCP Server 可以向 AI 提供预定义的交互模板：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


@mcp.prompt()
def code_review(language: str, code: str) -> str:
    """代码审查提示模板。
    
    Args:
        language: 编程语言
        code: 要审查的代码
    """
    return f"""请对以下 {language} 代码进行详细审查：

```{language}
{code}

请从以下方面进行审查：

代码质量和可读性
潜在的 bug 和边界情况
性能优化建议
安全漏洞检查
最佳实践遵循情况

请提供具体的改进建议和修改后的代码示例。"""

@mcp.prompt() def data_analysis(dataset_path: str, question: str) -> str: “““数据分析提示模板。””” return f"““数据分析任务：

数据集路径：{dataset_path}
分析问题：{question}

请使用可用的工具完成以下步骤：

首先查看数据集的结构（使用 describe_table 工具）
执行必要的查询获取相关数据（使用 query 工具）
进行聚合统计（使用 aggregate 工具）
根据数据回答分析问题并提供可视化建议””"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22



## 七、主流 MCP Server 生态

MCP 生态系统正在快速发展，以下是目前最流行的官方和社区 MCP Server：

### 7.1 官方参考实现

| MCP Server | 功能 | 安装方式 |
|-----------|------|---------|
| `@modelcontextprotocol/server-filesystem` | 文件系统操作 | `npx -y @modelcontextprotocol/server-filesystem` |
| `@modelcontextprotocol/server-postgres` | PostgreSQL 数据库 | `npx -y @modelcontextprotocol/server-postgres` |
| `@modelcontextprotocol/server-github` | GitHub API 操作 | `npx -y @modelcontextprotocol/server-github` |
| `@modelcontextprotocol/server-puppeteer` | 浏览器自动化 | `npx -y @modelcontextprotocol/server-puppeteer` |
| `@modelcontextprotocol/server-brave-search` | Brave 搜索引擎 | `npx -y @modelcontextprotocol/server-brave-search` |
| `@modelcontextprotocol/server-memory` | 知识图谱记忆 | `npx -y @modelcontextprotocol/server-memory` |
| `@modelcontextprotocol/server-fetch` | URL 内容抓取 | `npx -y @modelcontextprotocol/server-fetch` |

### 7.2 快速体验：文件系统 MCP Server

```bash
# 直接运行文件系统 MCP Server（限制在指定目录）
npx -y @modelcontextprotocol/server-filesystem /Users/you/projects

在 Claude Desktop 配置中添加：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/you/projects"
      ]
    }
  }
}

配置完成后重启 Claude Desktop，你就可以在对话中让 Claude 直接读写 /Users/you/projects 目录下的文件了。

7.3 第三方 MCP Server

社区还提供了大量第三方 MCP Server：

Slack MCP Server：在 Slack 中搜索消息和频道
Google Drive MCP Server：访问 Google Drive 文件
Notion MCP Server：读写 Notion 页面和数据库
Sentry MCP Server：查看和分析 Sentry 错误
Docker MCP Server：管理 Docker 容器和镜像
Kubernetes MCP Server：管理 K8s 集群资源

八、构建生产级 MCP Server

8.1 错误处理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51


import traceback
import logging
from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-server")

mcp = FastMCP("production-server")


class MCPError(Exception):
    """自定义 MCP 错误"""
    def __init__(self, message: str, code: str = "UNKNOWN_ERROR"):
        self.message = message
        self.code = code
        super().__init__(message)


@mcp.tool()
async def safe_operation(param: str) -> str:
    """一个具有完整错误处理的操作。"""
    try:
        # 业务逻辑
        if not param:
            raise MCPError("参数不能为空", "INVALID_PARAM")
        
        result = await _do_operation(param)
        return json.dumps({"success": True, "data": result}, ensure_ascii=False)
    
    except MCPError as e:
        logger.warning(f"业务错误: {e.code} - {e.message}")
        return json.dumps({
            "success": False,
            "error": {"code": e.code, "message": e.message}
        }, ensure_ascii=False)
    
    except Exception as e:
        logger.error(f"未预期的错误: {traceback.format_exc()}")
        return json.dumps({
            "success": False,
            "error": {
                "code": "INTERNAL_ERROR",
                "message": "服务器内部错误，请稍后重试"
            }
        }, ensure_ascii=False)


async def _do_operation(param: str) -> dict:
    """模拟业务操作"""
    # 实际实现...
    return {"result": f"处理了参数: {param}"}

8.2 认证模式

对于远程 MCP Server（SSE/HTTP 传输），认证至关重要：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49


# auth_server.py - 带认证的 MCP Server
from mcp.server.fastmcp import FastMCP
from mcp.server.auth import OAuthServerProvider
import httpx

mcp = FastMCP("authenticated-server")


class TokenVerifier:
    """验证 Bearer Token 的中间件"""
    
    def __init__(self, introspection_url: str):
        self.introspection_url = introspection_url
    
    async def verify(self, token: str) -> dict:
        """验证 token 并返回用户信息"""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.introspection_url,
                data={"token": token},
                timeout=5.0
            )
            if response.status_code == 200:
                data = response.json()
                if data.get("active"):
                    return data
            raise Exception("Invalid token")


# 使用中间件
verifier = TokenVerifier("https://auth.example.com/introspect")


@mcp.tool()
async def authenticated_action(query: str, token: str = None) -> str:
    """需要认证的操作。"""
    if not token:
        return json.dumps({"error": "需要提供认证 token"})
    
    try:
        user_info = await verifier.verify(token)
        # 执行需要认证的操作...
        return json.dumps({
            "success": True,
            "user": user_info.get("sub"),
            "result": f"用户 {user_info.get('sub')} 的查询结果"
        }, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"error": f"认证失败: {str(e)}"})

8.3 日志和可观测性

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37


import structlog
from contextlib import contextmanager
import time

# 结构化日志
logger = structlog.get_logger("mcp-server")


@mcp.tool()
async def observed_tool(param: str) -> str:
    """带完整日志记录的工具。"""
    start_time = time.time()
    
    logger.info("tool_called", 
                tool="observed_tool", 
                param=param)
    
    try:
        result = await _process(param)
        duration = time.time() - start_time
        
        logger.info("tool_completed",
                     tool="observed_tool",
                     duration_ms=round(duration * 1000, 2),
                     success=True)
        
        return json.dumps(result, ensure_ascii=False)
    
    except Exception as e:
        duration = time.time() - start_time
        
        logger.error("tool_failed",
                     tool="observed_tool",
                     duration_ms=round(duration * 1000, 2),
                     error=str(e))
        
        return json.dumps({"error": str(e)})

8.4 测试 MCP Server

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56


# test_mcp_server.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


@pytest.fixture
async def mcp_session():
    """创建测试用的 MCP 会话"""
    server_params = StdioServerParameters(
        command="python",
        args=["/path/to/weather_server.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session


@pytest.mark.asyncio
async def test_get_weather(mcp_session: ClientSession):
    """测试天气查询工具"""
    result = await mcp_session.call_tool(
        "get_weather",
        arguments={"city": "北京"}
    )
    
    assert result.content
    data = json.loads(result.content[0].text)
    assert "temperature" in data
    assert "description" in data


@pytest.mark.asyncio
async def test_list_tools(mcp_session: ClientSession):
    """测试列出工具"""
    tools = await mcp_session.list_tools()
    tool_names = [t.name for t in tools.tools]
    
    assert "get_weather" in tool_names
    assert "get_forecast" in tool_names


@pytest.mark.asyncio
async def test_invalid_city(mcp_session: ClientSession):
    """测试无效城市名称"""
    result = await mcp_session.call_tool(
        "get_weather",
        arguments={"city": "这是一个不存在的城市名称12345"}
    )
    
    data = json.loads(result.content[0].text)
    # 应该返回错误或默认值
    assert data is not None

九、MCP vs 原生 Function Calling

9.1 对比分析

维度	MCP	原生 Function Calling
标准化	✅ 开放协议，厂商无关	❌ 各厂商格式不同
状态管理	✅ 持久连接，维护上下文	❌ 无状态，每次调用独立
工具发现	✅ 自动发现 Server 提供的工具	❌ 开发者手动定义工具 schema
资源管理	✅ 内置 Resources 和 Prompts	❌ 需要自行实现
安全性	✅ 协议层安全，权限控制	⚠️ 依赖应用层实现
复用性	✅ Server 可被多个 Client 复用	❌ 工具定义绑定在应用代码中
复杂度	⚠️ 需要理解协议和搭建 Server	✅ 直接在 API 调用中定义
延迟	⚠️ 子进程启动或网络通信开销	✅ 无额外通信开销
生态	🔄 快速增长中	✅ 已经非常成熟

9.2 什么时候用 MCP？

推荐使用 MCP 的场景：

工具需要被多个 AI 应用共享（如公司内部的工具平台）
需要持久连接和状态管理（如数据库连接池、浏览器会话）
需要标准化的工具发现和注册机制
构建可插拔的 Agent 系统
需要精细的权限和安全控制

推荐使用原生 Function Calling 的场景：

简单的单次工具调用，无状态需求
快速原型开发，不想搭建额外基础设施
工具逻辑简单，不需要复杂的协议支持
对延迟极其敏感的场景

9.3 混合使用策略

实际上，MCP 和 Function Calling 可以共存。一个典型的 AI Agent 架构可能是：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


┌──────────────────────────────────────────────┐
│                  AI Agent                     │
│                                              │
│   LLM ──┬── Function Calling (简单工具)       │
│         │   • 字符串格式化                    │
│         │   • 简单数学计算                    │
│         │   • JSON 数据转换                   │
│         │                                    │
│         └── MCP (复杂工具)                    │
│             • 数据库查询                      │
│             • 文件系统操作                    │
│             • 外部 API 调用                   │
│             • 浏览器自动化                    │
└──────────────────────────────────────────────┘

十、MCP 生态系统

10.1 MCP Server 注册中心

Smithery（smithery.ai）：最大的 MCP Server 注册中心，提供搜索、评分和一键安装
Glama MCP Directory（glama.ai/mcp/servers）：MCP Server 目录，提供详细的文档和示例
mcp.run：MCP Server 的云托管平台，支持一键部署远程 MCP Server

10.2 开发工具链

MCP Inspector：官方调试工具，用于测试和调试 MCP Server
MCP CLI：命令行工具，用于创建、测试和发布 MCP Server
Cursor / Windsurf / VS Code：越来越多的 IDE 开始原生支持 MCP

10.3 MCP Inspector：调试利器

MCP Inspector 是 Anthropic 提供的官方调试工具：

1
2


# 启动 MCP Inspector
npx @modelcontextprotocol/inspector python weather_server.py

Inspector 提供了一个 Web 界面，你可以：

查看 Server 暴露的所有工具、资源和提示
手动调用工具并查看返回结果
查看 JSON-RPC 消息流
测试错误场景

十一、MCP 的未来展望

11.1 Agent-to-Agent 通信

MCP 的下一个重要方向是 Agent-to-Agent 协议。当多个 AI Agent 需要协作时，MCP 可以作为它们之间的通信桥梁：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


# 未来的 Agent-to-Agent MCP 模式
agent_a_mcp_config = {
    "mcpServers": {
        "agent_b": {
            "command": "python",
            "args": ["agent_b_server.py"],
            "capabilities": {
                "agent_id": "research-agent",
                "supported_tasks": ["search", "analysis"]
            }
        }
    }
}

11.2 远程 MCP Server

SSE/HTTP 传输机制让远程 MCP Server 成为可能。未来的趋势：

MCP Server as a Service：云厂商提供托管的 MCP Server
MCP Gateway：网关模式，统一管理多个 MCP Server 的访问
MCP Marketplace：商业化市场，开发者可以出售自己的 MCP Server

11.3 标准化与互操作

MCP 协议仍在快速演进中，未来可能的发展方向：

与其他 Agent 协议（如 LangChain、AutoGPT）的互操作
标准化的 MCP Server 认证和授权框架
性能优化：支持流式传输和批量操作
更丰富的错误码和状态码体系

总结

MCP 协议为 AI Agent 的工具生态提供了一个优雅的标准化方案。它将 N×M 的集成复杂度降低为 N+M，通过客户端-服务器架构实现了工具的可发现性、可复用性和可管理性。

关键要点回顾：

MCP = AI 的 USB-C：统一的工具连接协议
三层架构：Host（AI 应用）→ Client（协议客户端）→ Server（工具服务）
三种原语：Tools（工具调用）、Resources（数据源）、Prompts（提示模板）
两种传输：stdio（本地）和 SSE/HTTP（远程）
两种语言 SDK：Python 和 TypeScript 都有一流支持

现在就开始构建你的第一个 MCP Server 吧！从简单的文件操作工具开始，逐步扩展到数据库、API 集成等复杂场景。MCP 生态正在快速发展，越早参与，越能在这个新兴领域占据先机。

相关资源：