MCP协议的设计原理和核心架构

MCP协议：AI连接世界的标准

MCP（Model Context Protocol，模型上下文协议）是Anthropic在2024年底发布的开放协议，旨在为AI模型与外部工具、数据源建立标准化的通信方式。本文深入解析MCP的设计原理、协议架构、以及在AI工程实践中的应用。

一、为什么需要MCP

1.1 AI应用的核心挑战

当前AI应用面临一个根本性问题：AI模型是孤立的——它只能看到训练时的知识，无法直接与外部世界交互。RAG系统解决了知识获取问题，但工具调用、数据写入、多系统协同仍然缺乏统一标准。

典型场景：

AI需要从数据库读取实时数据

AI需要调用外部API完成特定操作

AI需要向多个外部系统写入数据

AI需要订阅实时事件流

每个AI应用都在重复造轮子：自定义的工具调用格式、自定义的API适配层、自定义的错误处理。

1.2 MCP的核心价值

MCP的定位是成为"AI世界的USB接口"——提供一个标准化的方式让AI模型与任何外部系统交互。

 

USB接口解决的问题:
- 不需要为每个设备设计专门的接口
- 设备插上就能用
- 标准化意味着生态化

MCP解决的问题:
- 不需要为每个外部系统设计专门的适配
- 工具接入一次，任何AI模型都能调用
- 标准化促进生态繁荣

 

二、MCP协议架构

2.1 核心组件

 

┌─────────────────────────────────────────────────────────────┐
│                      AI Application                          │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                   MCP Client                         │    │
│  │  - 维护与Server的连接                                │    │
│  │  - 序列化/反序列化消息                                │    │
│  │  - 处理协议握手                                      │    │
│  └─────────────────────────────────────────────────────┘    │
└──────────────────────────┬──────────────────────────────────┘
                           │ stdio / HTTP + SSE
┌──────────────────────────▼──────────────────────────────────┐
│                      MCP Server (进程)                       │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              Protocol Engine                          │    │
│  │  - 消息路由                                           │    │
│  │  - 能力协商                                           │    │
│  │  - 错误处理                                           │    │
│  └─────────────────────────────────────────────────────┘    │
│  ┌─────────────────────────────────────────────────────┐    │
│  │               Tool Handler                           │    │
│  │  - 暴露工具清单                                       │    │
│  │  - 执行工具调用                                       │    │
│  │  - 返回结果格式化                                     │    │
│  └─────────────────────────────────────────────────────┘    │
│  ┌─────────────────────────────────────────────────────┐    │
│  │               Resource Handler                       │    │
│  │  - 管理可读取的资源                                   │    │
│  │  - 处理资源订阅                                      │    │
│  │  - 内容缓存                                          │    │
│  └─────────────────────────────────────────────────────┘    │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────┐
│                    External Systems                          │
│  - REST APIs    - Databases    - File Systems    - etc.    │
└─────────────────────────────────────────────────────────────┘

 

2.2 通信协议

MCP支持两种传输方式：

stdio模式（适合本地/CLI工具）：

 

AI App → MCP Client → stdin → MCP Server → External System
         ↑                      ↓
         └────── stdout ←───────┘

 

HTTP + SSE模式（适合远程服务）：

 

AI App → MCP Client ──── HTTP POST /message ──→ MCP Server
         ↑                                              ↓
         └──────── SSE (Server-Sent Events) ←──────────┘

 

2.3 消息类型

MCP定义了四种核心消息类型：

 

// 1. Initialize - 握手协商
{"jsonrpc": "2.0", "id": 1, "method": "initialize",
 "params": {
   "protocolVersion": "2024-11-05",
   "capabilities": {"roots": {}, "tools": {}},
   "clientInfo": {"name": "claude-desktop", "version": "1.0"}
 }}

{"jsonrpc": "2.0", "id": 1, "result": {
   "protocolVersion": "2024-11-05",
   "capabilities": {"tools": {}, "resources": {}},
   "serverInfo": {"name": "filesystem-server", "version": "1.0"}
 }}

// 2. Tools/List - 列出可用工具
{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}

{"jsonrpc": "2.0", "id": 2, "result": {
   "tools": [
     {"name": "read_file", "description": "读取文件内容",
      "inputSchema": {"type": "object", "properties": {
        "path": {"type": "string"}
      }}},
     {"name": "write_file", "description": "写入文件内容",
      "inputSchema": {"type": "object", "properties": {
        "path": {"type": "string"},
        "content": {"type": "string"}
      }}}
   ]
 }}

// 3. Tools/Call - 调用工具
{"jsonrpc": "2.0", "id": 3, "method": "tools/call",
 "params": {
   "name": "read_file",
   "arguments": {"path": "/etc/passwd"}
 }}

{"jsonrpc": "2.0", "id": 3, "result": {
   "content": [{"type": "text", "text": "root0root:/root:/bin/bash
..."}]
 }}

// 4. Resources - 资源访问
{"jsonrpc": "2.0", "id": 4, "method": "resources/list"}

{"jsonrpc": "2.0", "id": 4, "result": {
   "resources": [
     {"uri": "file:///project/config.yaml", "name": "Config", "mimeType": "application/yaml"}
   ]
 }}

 

三、MCP Server开发

3.1 Python SDK实现

 

from mcp.server.fastmcp import FastMCP

# 初始化MCP Server
mcp = FastMCP("filesystem-server")

@mcp.tool()
def read_file(path: str, encoding: str = "utf-8") -> str:
    """读取文件内容"""
    with open(path, 'r', encoding=encoding) as f:
        return f.read()

@mcp.tool()
def write_file(path: str, content: str) -> dict:
    """写入文件内容"""
    with open(path, 'w', encoding='utf-8') as f:
        f.write(content)
    return {"success": True, "path": path, "bytes": len(content)}

@mcp.tool()
def list_directory(path: str) -> list[dict]:
    """列出目录内容"""
    import os
    items = []
    for item in os.listdir(path):
        full_path = os.path.join(path, item)
        items.append({
            "name": item,
            "type": "dir" if os.path.isdir(full_path) else "file",
            "size": os.path.getsize(full_path) if os.path.isfile(full_path) else None
        })
    return items

@mcp.resource("file://{path}")
def file_resource(path: str) -> str:
    """动态资源访问"""
    with open(path, 'r') as f:
        return f.read()

# 启动服务器
if __name__ == "__main__":
    mcp.run()

 

3.2 带复杂参数的工具

 

@mcp.tool()
def search_codebase(
    query: str,
    file_pattern: str = "*.py",
    case_sensitive: bool = False,
    max_results: int = 50
) -> list[dict]:
    """
    在代码库中搜索代码

    Args:
        query: 搜索关键词
        file_pattern: 文件名模式 (glob格式)
        case_sensitive: 是否区分大小写
        max_results: 最大返回结果数
    """
    import glob
    import os

    results = []
    for file_path in glob.glob(file_pattern, recursive=True):
        if not os.path.isfile(file_path):
            continue

        try:
            with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                lines = f.readlines()

            for i, line in enumerate(lines):
                if case_sensitive:
                    match = query in line
                else:
                    match = query.lower() in line.lower()

                if match:
                    results.append({
                        "file": file_path,
                        "line": i + 1,
                        "content": line.strip(),
                        "context": lines[max(0, i-2):i] + lines[i+1:min(len(lines), i+3)]
                    })

                    if len(results) >= max_results:
                        return results
        except Exception:
            continue

    return results

 

3.3 流式响应工具

 

from typing import AsyncIterator

@mcp.tool()
async def stream_log_tail(
    path: str,
    lines: int = 100
) -> AsyncIterator[dict]:
    """流式读取日志文件最后N行"""
    import asyncio

    with open(path, 'r') as f:
        all_lines = f.readlines()

    last_lines = all_lines[-lines:]

    for line in last_lines:
        yield {"content": line.strip()}
        await asyncio.sleep(0.1)  # 模拟实时推送

 

四、MCP Client集成

4.1 Python Client实现

 

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

class MCPClient:
    def __init__(self, server_command: list[str]):
        self.server_command = server_command

    async def initialize(self):
        server_params = StdioServerParameters(
            command=self.server_command[0],
            args=self.server_command[1:]
        )

        async with stdio_client(server_params) as (read, write):
            self.session = ClientSession(read, write)
            await self.session.initialize()

            # 获取服务器能力
            response = await self.session.get_server_capabilities()
            print(f"Server capabilities: {response.capabilities}")

    async def list_tools(self) -> list[dict]:
        response = await self.session.list_tools()
        return [tool.model_dump() for tool in response.tools]

    async def call_tool(
        self,
        tool_name: str,
        arguments: dict
    ) -> dict:
        result = await self.session.call_tool(tool_name, arguments)
        return result.content[0].model_dump()

# 使用示例
async def main():
    client = MCPClient(["python", "filesystem_server.py"])
    await client.initialize()

    # 列出可用工具
    tools = await client.list_tools()
    print("Available tools:", [t["name"] for t in tools])

    # 调用工具
    result = await client.call_tool("read_file", {"path": "/tmp/test.txt"})
    print("File content:", result["text"])

asyncio.run(main())

 

4.2 与LangChain集成

 

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# 初始化MCP客户端，连接多个服务器
client = MultiServerMCPClient({
    "filesystem": {
        "command": "python",
        "args": ["/path/to/filesystem_server.py"]
    },
    "database": {
        "command": "python",
        "args": ["/path/to/database_server.py"]
    }
})

# 获取所有工具
tools = client.get_tools()

# 创建Agent
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, tools)

# 运行Agent
result = agent.invoke({
    "messages": [{"role": "user", "content": "读取/tmp/config.yaml文件并总结关键配置"}]
})

 

五、生产环境最佳实践

5.1 Server健康检查

 

import asyncio
from mcp import ClientSession
from mcp.client.stdio import stdio_client

class MCPClientWithHealthCheck:
    def __init__(self, server_params, health_check_timeout: float = 5.0):
        self.server_params = server_params
        self.health_check_timeout = health_check_timeout
        self._session = None

    async def health_check(self) -> bool:
        """检查MCP Server是否可用"""
        try:
            async with stdio_client(self.server_params) as (read, write):
                session = ClientSession(read, write)
                await asyncio.wait_for(
                    session.initialize(),
                    timeout=self.health_check_timeout
                )

                # 尝试列出工具
                await session.list_tools()
                return True
        except asyncio.TimeoutError:
            return False
        except Exception as e:
            print(f"Health check failed: {e}")
            return False

    async def get_session(self) -> ClientSession:
        if self._session is None:
            async with stdio_client(self.server_params) as (read, write):
                self._session = ClientSession(read, write)
                await self._session.initialize()
        return self._session

 

5.2 错误处理与重试

 

import asyncio
from mcp.error import MCPError

class ResilientMCPClient:
    def __init__(self, server_params, max_retries: int = 3):
        self.server_params = server_params
        self.max_retries = max_retries

    async def call_with_retry(
        self,
        tool_name: str,
        arguments: dict
    ) -> dict:
        last_error = None

        for attempt in range(self.max_retries):
            try:
                async with stdio_client(self.server_params) as (read, write):
                    session = ClientSession(read, write)
                    await session.initialize()

                    result = await session.call_tool(tool_name, arguments)
                    return result.content[0].model_dump()

            except MCPError as e:
                # 协议级错误，不重试
                raise(f"MCP protocol error: {e}")
            except Exception as e:
                last_error = e
                if attempt < self.max_retries - 1:
                    await asyncio.sleep(2 ** attempt)  # 指数退避
                continue

        raise RuntimeError(f"Failed after {self.max_retries} attempts: {last_error}")

 

5.3 安全考虑

MCP的工具调用执行在Server进程中，需要注意：

 

# Server端：输入验证
@mcp.tool()
def delete_file(path: str) -> dict:
    """删除文件（需要严格的安全检查）"""
    import os

    # 禁止删除系统关键路径
    forbidden_paths = ["/", "/etc", "/usr", "/bin", "/sbin", "/var"]
    abs_path = os.path.abspath(path)

    for forbidden in forbidden_paths:
        if abs_path.startswith(forbidden):
            raise ValueError(f"Cannot delete files in {forbidden}")

    # 禁止跨目录删除
    if ".." in path:
        raise ValueError("Path traversal not allowed")

    # 检查文件是否存在
    if not os.path.exists(abs_path):
        return {"success": False, "error": "File not found"}

    os.remove(abs_path)
    return {"success": True, "path": path}

 

六、主流MCP Server生态

6.1 官方及社区Server

Server	功能	维护方
filesystem	本地文件系统访问	Anthropic官方
slack	Slack消息读写	Anthropic官方
github	GitHub API操作	Anthropic官方
puppeteer	浏览器自动化	社区
postgresql	PostgreSQL查询	社区
Brave Search	网页搜索	社区

6.2 Server发现与配置

 

// claude_desktop_config.json
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-server-filesystem"],
      "config": {
        "allowedDirectories": ["/Users/me/projects", "/tmp"]
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxx"
      }
    },
    "postgresql": {
      "command": "uvx",
      "args": ["mcp-server-postgres", "--host", "localhost", "--port", "5432"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost/mydb"
      }
    }
  }
}

 

七、协议演进与未来

MCP协议仍处于活跃开发中，以下是值得关注的方向：

7.1 采样能力（Prompt Caching的未来）

当前协议支持Server向Client"推送"数据的能力有限。未来版本可能会增强双向通信能力。

7.2 多Agent协作

MCP的设计天然支持多Agent场景：

多个AI Agent通过各自的MCP Client连接同一Server

Server作为协调者管理资源访问冲突

支持Agent间的消息传递

 

# 多Agent共享MCP Server
class SharedMCPBridge:
    def __init__(self, server):
        self.server = server
        self.agents = {}

    def register_agent(self, agent_id: str, client_session):
        self.agents[agent_id] = client_session

    async def broadcast(self, message: dict, from_agent: str):
        """向所有Agent广播消息"""
        for agent_id, session in self.agents.items():
            if agent_id != from_agent:
                await session.send_notification("agent/message", message)

 

总结

MCP协议代表了AI应用架构的一个重要演进方向：

核心价值：

标准化：统一的工具调用协议

生态化：一次实现，到处可用

安全性：清晰的权限边界

工程实践要点：

Server端做好输入验证和权限控制

Client端实现健康检查和重试机制

生产环境注意监控工具调用延迟和错误率

合理设计工具粒度，避免过度抽象

随着MCP生态的成熟，我们可以期待看到更多标准化的工具Server出现，AI应用的开发效率将显著提升。