MCP 协议详解

MCP（Model Context Protocol）是连接 AI 模型与外部资源的标准化协议，实现工具调用的统一接口。

一、核心概念

1.1 什么是 MCP

MCP = Model Context Protocol

目标：
- 标准化 AI 模型与外部资源的连接
- 统一的工具调用接口
- 支持多种资源类型（文件、数据库、API）

1.2 架构组件

MCP 架构
├── MCP Client（客户端）
│   └── AI 模型/应用
├── MCP Server（服务器）
│   ├── 资源管理器
│   ├── 工具注册表
│   └── 协议处理器
└── Resources（资源）
    ├── 文件系统
    ├── 数据库
    └── API 服务

二、协议规范

2.1 基础协议

// MCP 消息格式
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "resources/list",
  "params": {}
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "resources": [
      {
        "uri": "file:///data/report.pdf",
        "name": "Report PDF",
        "mimeType": "application/pdf"
      }
    ]
  }
}

2.2 资源定义

# 资源规范
class Resource:
    def __init__(self, uri: str, name: str, mime_type: str):
        self.uri = uri  # 统一资源标识符
        self.name = name  # 人类可读名称
        self.mimeType = mime_type  # MIME 类型
        self.description = ""  # 可选描述

2.3 工具定义

# 工具规范
class Tool:
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description
        self.inputSchema = {
            "type": "object",
            "properties": {
                "param1": {"type": "string"}
            },
            "required": ["param1"]
        }

三、实现模式

3.1 MCP Server 实现

from mcp.server import Server
from mcp.types import Resource, Tool

server = Server("my-mcp-server")

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="file:///data/doc.pdf",
            name="Document",
            mimeType="application/pdf"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    # 读取资源内容
    with open(uri.replace("file://", ""), "r") as f:
        return f.read()

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search",
            description="Search documents"
        )
    ]

@server.call_tool()
async def call_tool(name: str, args: dict) -> str:
    if name == "search":
        return search_documents(args["query"])

3.2 MCP Client 实现

from mcp.client import Client

client = Client("ws://localhost:8080")

# 连接服务器
await client.connect()

# 列出资源
resources = await client.list_resources()

# 读取资源
content = await client.read_resource("file:///data/doc.pdf")

# 调用工具
result = await client.call_tool("search", {"query": "AI"})

四、资源连接

4.1 文件系统资源

class FileSystemResource:
    def __init__(self, root_path: str):
        self.root = Path(root_path)
    
    async def read(self, uri: str) -> str:
        path = self.root / uri.replace("file://", "")
        
        # 安全检查
        if not path.is_relative_to(self.root):
            raise SecurityError("Path traversal detected")
        
        return path.read_text()
    
    async def list(self) -> list[Resource]:
        resources = []
        for file in self.root.rglob("*"):
            if file.is_file():
                resources.append(Resource(
                    uri=f"file://{file}",
                    name=file.name,
                    mimeType=self.get_mime_type(file)
                ))
        return resources

4.2 数据库资源

class DatabaseResource:
    def __init__(self, connection_string: str):
        self.conn = await asyncpg.connect(connection_string)
    
    async def query(self, sql: str, params: dict) -> list:
        # 参数化查询，防止 SQL 注入
        rows = await self.conn.fetch(sql, **params)
        return [dict(row) for row in rows]
    
    async def list_tables(self) -> list[str]:
        tables = await self.conn.fetch(
            "SELECT table_name FROM information_schema.tables"
        )
        return [t["table_name"] for t in tables]

4.3 API 资源

class APIResource:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {api_key}"}
        )
    
    async def get(self, endpoint: str, params: dict = None) -> dict:
        async with self.session.get(
            f"{self.base_url}/{endpoint}",
            params=params
        ) as response:
            return await response.json()
    
    async def post(self, endpoint: str, data: dict) -> dict:
        async with self.session.post(
            f"{self.base_url}/{endpoint}",
            json=data
        ) as response:
            return await response.json()

五、安全控制

5.1 访问控制

class AccessControl:
    def __init__(self):
        self.policies = {}
    
    def check_permission(self, user: str, resource: str, action: str) -> bool:
        policy = self.policies.get(resource, {})
        allowed_users = policy.get(action, [])
        return user in allowed_users
    
    def add_policy(self, resource: str, action: str, users: list):
        if resource not in self.policies:
            self.policies[resource] = {}
        self.policies[resource][action] = users

5.2 审计日志

class AuditLogger:
    async def log(self, user: str, action: str, resource: str, result: str):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "user": user,
            "action": action,
            "resource": resource,
            "result": result
        }
        await self.save_log(log_entry)
    
    async def get_logs(self, user: str = None, start_time: str = None) -> list:
        # 查询审计日志
        pass

六、最佳实践

6.1 错误处理

class MCPErrorHandler:
    async def handle_error(self, error: Exception) -> dict:
        if isinstance(error, SecurityError):
            return {"error": "Access denied", "code": 403}
        elif isinstance(error, ResourceNotFoundError):
            return {"error": "Resource not found", "code": 404}
        else:
            return {"error": "Internal error", "code": 500}

6.2 性能优化

# 连接池
class ConnectionPool:
    def __init__(self, max_size: int = 10):
        self.pool = await asyncpg.create_pool(
            max_size=max_size
        )
    
    async def execute(self, query: str, params: dict):
        async with self.pool.acquire() as conn:
            return await conn.fetch(query, **params)

七、总结

MCP 协议核心要点：

MCP 是 AI 应用与外部资源连接的标准协议，理解其原理有助于构建可扩展的 AI 系统。