深度解析 MCP (Model Context Protocol)

一、MCP 是什么？

Model Context Protocol (MCP) 是 Anthropic 推出的开源标准协议，用于连接 AI 应用与外部系统。如果用一个形象的比喻，MCP 就像是 AI 应用的 USB-C 接口——提供了一个标准化的连接方式，让 AI 应用能够与各种数据源、工具和工作流无缝集成。

1.1 为什么需要 MCP？

当前 AI 应用面临的核心挑战：

信息孤岛问题：每个数据源都需要单独的集成方案
开发复杂度高：为每个 AI 应用构建定制化的数据连接器
缺乏标准化：不同厂商的集成方案互不兼容
上下文丢失：AI 在不同工具间切换时难以保持上下文连续性

MCP 通过提供统一的协议标准，将这些碎片化的集成方案替换为单一的标准协议，从根本上解决了这些问题。

二、核心架构设计

2.1 架构组件

MCP 采用经典的客户端-服务器架构：

┌─────────────────┐
│   MCP Host      │ ← AI 应用（如 Claude Desktop）
│   (协调者)      │
└────────┬────────┘
         │
         ├─── MCP Client 1 ──→ MCP Server A (数据库)
         │
         ├─── MCP Client 2 ──→ MCP Server B (文件系统)
         │
         └─── MCP Client 3 ──→ MCP Server C (API 服务)

各组件职责：

MCP Host：AI 应用本身，负责协调多个 MCP 客户端
MCP Client：维护与 MCP 服务器的连接，处理协议通信
MCP Server：提供具体的上下文数据和能力（工具、资源、提示）

2.2 协议设计

MCP 基于 JSON-RPC 2.0 构建，包含两个核心层：

数据层 (Protocol Layer)

定义通信协议格式
管理连接生命周期
处理能力协商机制

传输层 (Transport Layer)

管理通信信道
支持 stdio（本地进程）和 HTTP with SSE（远程服务器）

三、三大核心原语

3.1 Resources（资源）

类似文件系统的只读数据源，供客户端读取。

{
  "uri": "file:///path/to/document.txt",
  "name": "项目文档",
  "mimeType": "text/plain",
  "description": "项目需求文档"
}

3.2 Tools（工具）

LLM 可调用的可执行函数，实现具体操作。

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather")

@mcp.tool()
async def get_weather(city: str, state: str) -> str:
    """获取指定城市的天气信息"""
    return f"{city}, {state} 的天气是晴天，温度 25°C"

3.3 Prompts（提示模板）

预编写的任务模板，用于标准化常见操作。

{
  "name": "code_review",
  "description": "代码审查提示模板",
  "arguments": [
    {
      "name": "language",
      "description": "编程语言",
      "required": true
    }
  ]
}

四、实战：构建 MCP 服务器

4.1 Python 实现（FastMCP）

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("database-server")

@mcp.tool()
async def query_users(limit: int = 10) -> str:
    """查询用户列表"""
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users LIMIT ?', (limit,))
    results = cursor.fetchall()
    conn.close()
    return str(results)

@mcp.tool()
async def create_user(name: str, email: str) -> str:
    """创建新用户"""
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute(
        'INSERT INTO users (name, email) VALUES (?, ?)',
        (name, email)
    )
    conn.commit()
    user_id = cursor.lastrowid
    conn.close()
    return f"用户创建成功，ID: {user_id}"

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

4.2 TypeScript/Node.js 实现

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
  { name: "file-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler("tools/call", async (request) => {
  if (request.params.name === "read_file") {
    const fs = await import("fs/promises");
    const content = await fs.readFile(request.params.arguments.path, "utf-8");
    return { content: [{ type: "text", text: content }] };
  }
  throw new Error("未知工具");
});

const transport = new StdioServerTransport();
await server.connect(transport);

4.3 最佳实践

日志处理：stdio 服务器永远不要写入 stdout，使用 stderr 或文件日志
错误处理：实现完善的异常处理和错误信息返回
类型安全：使用类型提示和输入验证
安全控制：实现适当的访问控制和参数验证

五、构建 MCP 客户端

from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPClient:
    def __init__(self):
        self.anthropic = Anthropic()
        self.session = None

    async def connect_to_server(self, server_script_path: str):
        server_params = StdioServerParameters(
            command="python",
            args=[server_script_path],
        )
        stdio_transport = await stdio_client(server_params)
        self.stdio, self.write = stdio_transport
        self.session = ClientSession(self.stdio, self.write)
        await self.session.initialize()

    async def process_query(self, query: str) -> str:
        tools = await self.session.list_tools()
        response = self.anthropic.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            messages=[{"role": "user", "content": query}],
            tools=[{
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            } for tool in tools.tools]
        )
        # 处理工具调用并返回结果
        return response.content[0].text

六、生态系统

6.1 官方参考服务器

服务器	功能描述
filesystem	安全的文件系统操作
git	Git 仓库读取、搜索和操作
github	GitHub API 集成
fetch	Web 内容抓取和转换
memory	基于知识图谱的持久化记忆

6.2 企业级应用

已集成 MCP 的公司：Block、Apollo、Zed、Replit、Codeium、Sourcegraph、Microsoft 等。

6.3 官方 SDK

Python、TypeScript、Kotlin、Java、C#、Ruby

七、MCP vs 传统方案

维度	传统方式	MCP 方式
开发成本	每个数据源需单独实现	一次实现，到处复用
维护难度	多套代码分别维护	统一协议标准
扩展性	添加新数据源需重写	实现 MCP 服务器即可
互操作性	厂商方案互不兼容	开放标准，跨平台

八、安全与最佳实践

访问控制

@mcp.tool()
async def read_file(path: str) -> str:
    if not is_path_allowed(path):
        raise PermissionError("路径访问被拒绝")
    return read_file_content(path)

参数验证

from pydantic import BaseModel, validator

class QueryParams(BaseModel):
    limit: int

    @validator('limit')
    def validate_limit(cls, v):
        if v > 1000:
            raise ValueError('limit 不能超过 1000')
        return v