MCP 规范与最佳实践

Model Context Protocol (MCP) 是连接 AI 模型与外部资源的标准化协议。如何开发 MCP 服务器？如何定义资源？本文详解 MCP 规范与最佳实践。

一、MCP 概述

1.1 协议架构

MCP 协议架构：

┌─────────────────────────────────────┐
│              Client                  │
│         (AI Application)             │
├─────────────────────────────────────┤
│           MCP Protocol               │
│    (JSON-RPC based protocol)         │
├─────────────────────────────────────┤
│              Server                  │
│    (Resource/Tool Provider)          │
│  ┌──────────┬──────────┬──────────┐ │
│  │Resources │  Tools   │Prompts   │ │
│  └──────────┴──────────┴──────────┘ │
└─────────────────────────────────────┘

1.2 核心概念

二、服务器开发

2.1 基础服务器

# mcp_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, Prompt
import json

class MCPServer:
    """MCP 服务器"""
    
    def __init__(self, name: str):
        self.server = Server(name)
        self._register_handlers()
    
    def _register_handlers(self):
        """注册处理器"""
        @self.server.list_resources()
        async def handle_list_resources() -> list[Resource]:
            """列出可用资源"""
            return [
                Resource(
                    uri="file:///documents/report.pdf",
                    name="Report",
                    description="Quarterly report",
                    mimeType="application/pdf"
                )
            ]
        
        @self.server.read_resource()
        async def handle_read_resource(uri: str) -> str:
            """读取资源"""
            # 实现资源读取逻辑
            return "Resource content"
        
        @self.server.list_tools()
        async def handle_list_tools() -> list[Tool]:
            """列出可用工具"""
            return [
                Tool(
                    name="search",
                    description="Search documents",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "Search query"
                            }
                        },
                        "required": ["query"]
                    }
                )
            ]
        
        @self.server.call_tool()
        async def handle_call_tool(
            name: str,
            arguments: dict
        ) -> list:
            """调用工具"""
            if name == "search":
                query = arguments.get("query", "")
                results = self._search_documents(query)
                return [{"type": "text", "text": json.dumps(results)}]
            return []
        
        @self.server.list_prompts()
        async def handle_list_prompts() -> list[Prompt]:
            """列出可用提示"""
            return [
                Prompt(
                    name="analyze",
                    description="Analyze document",
                    arguments=[
                        {
                            "name": "document",
                            "description": "Document to analyze",
                            "required": True
                        }
                    ]
                )
            ]
        
        @self.server.get_prompt()
        async def handle_get_prompt(
            name: str,
            arguments: dict
        ) -> str:
            """获取提示"""
            if name == "analyze":
                document = arguments.get("document", "")
                return f"Please analyze: {document}"
            return ""
    
    def _search_documents(self, query: str) -> list:
        """搜索文档"""
        # 实现搜索逻辑
        return []
    
    async def run(self):
        """运行服务器"""
        async with stdio_server() as streams:
            await self.server.run(
                streams[0],
                streams[1],
                self.server.create_initialization_options()
            )

# 使用示例
if __name__ == "__main__":
    server = MCPServer("my-mcp-server")
    import asyncio
    asyncio.run(server.run())

2.2 资源定义

# mcp_resources.py
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class ResourceMIMEType(Enum):
    """资源 MIME 类型"""
    TEXT = "text/plain"
    JSON = "application/json"
    PDF = "application/pdf"
    IMAGE_PNG = "image/png"
    IMAGE_JPEG = "image/jpeg"

@dataclass
class ResourceDefinition:
    """资源定义"""
    uri: str
    name: str
    description: str
    mime_type: ResourceMIMEType
    metadata: Dict = None

class ResourceManager:
    """资源管理器"""
    
    def __init__(self):
        self.resources: Dict[str, ResourceDefinition] = {}
    
    def register_resource(self, resource: ResourceDefinition):
        """注册资源"""
        self.resources[resource.uri] = resource
    
    def unregister_resource(self, uri: str):
        """注销资源"""
        if uri in self.resources:
            del self.resources[uri]
    
    def get_resource(self, uri: str) -> Optional[ResourceDefinition]:
        """获取资源"""
        return self.resources.get(uri)
    
    def list_resources(self) -> List[ResourceDefinition]:
        """列出所有资源"""
        return list(self.resources.values())
    
    def search_resources(self, query: str) -> List[ResourceDefinition]:
        """搜索资源"""
        results = []
        for resource in self.resources.values():
            if (
                query.lower() in resource.name.lower() or
                query.lower() in resource.description.lower()
            ):
                results.append(resource)
        return results

# 使用示例
manager = ResourceManager()

# 注册文件资源
manager.register_resource(ResourceDefinition(
    uri="file:///documents/report.pdf",
    name="Quarterly Report",
    description="Q4 2026 financial report",
    mime_type=ResourceMIMEType.PDF
))

# 注册数据库资源
manager.register_resource(ResourceDefinition(
    uri="db://users/table",
    name="Users Table",
    description="User database table",
    mime_type=ResourceMIMEType.JSON,
    metadata={"table": "users", "columns": ["id", "name", "email"]}
))

三、工具开发

3.1 工具定义

# mcp_tools.py
from typing import Dict, List, Any, Callable
from dataclasses import dataclass
import json

@dataclass
class ToolParameter:
    """工具参数"""
    name: str
    type: str
    description: str
    required: bool = True
    default: Any = None
    enum: List = None

@dataclass
class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    parameters: List[ToolParameter]
    handler: Callable

class ToolManager:
    """工具管理器"""
    
    def __init__(self):
        self.tools: Dict[str, ToolDefinition] = {}
    
    def register_tool(self, tool: ToolDefinition):
        """注册工具"""
        self.tools[tool.name] = tool
    
    def execute_tool(
        self,
        name: str,
        arguments: Dict
    ) -> Dict:
        """执行工具"""
        if name not in self.tools:
            return {
                "success": False,
                "error": f"Tool '{name}' not found"
            }
        
        tool = self.tools[name]
        
        # 验证参数
        validation = self._validate_arguments(
            tool.parameters,
            arguments
        )
        
        if not validation["valid"]:
            return validation
        
        # 执行工具
        try:
            result = tool.handler(**arguments)
            return {
                "success": True,
                "result": result
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def _validate_arguments(
        self,
        parameters: List[ToolParameter],
        arguments: Dict
    ) -> Dict:
        """验证参数"""
        errors = []
        
        for param in parameters:
            # 检查必填参数
            if param.required and param.name not in arguments:
                errors.append(f"Missing required parameter: {param.name}")
                continue
            
            # 检查参数类型
            if param.name in arguments:
                value = arguments[param.name]
                if not self._check_type(value, param.type):
                    errors.append(
                        f"Parameter '{param.name}' should be {param.type}"
                    )
                
                # 检查枚举值
                if param.enum and value not in param.enum:
                    errors.append(
                        f"Parameter '{param.name}' should be one of {param.enum}"
                    )
        
        return {
            "valid": len(errors) == 0,
            "errors": errors
        }
    
    def _check_type(self, value: Any, expected_type: str) -> bool:
        """检查类型"""
        type_map = {
            "string": str,
            "integer": int,
            "number": (int, float),
            "boolean": bool,
            "array": list,
            "object": dict
        }
        
        expected = type_map.get(expected_type)
        if expected:
            return isinstance(value, expected)
        return True
    
    def to_mcp_schema(self, name: str) -> Dict:
        """转换为 MCP Schema"""
        tool = self.tools.get(name)
        if not tool:
            return {}
        
        properties = {}
        required = []
        
        for param in tool.parameters:
            properties[param.name] = {
                "type": param.type,
                "description": param.description
            }
            
            if param.default is not None:
                properties[param.name]["default"] = param.default
            
            if param.enum:
                properties[param.name]["enum"] = param.enum
            
            if param.required:
                required.append(param.name)
        
        return {
            "name": tool.name,
            "description": tool.description,
            "inputSchema": {
                "type": "object",
                "properties": properties,
                "required": required
            }
        }

# 使用示例
tool_manager = ToolManager()

# 定义搜索工具
def search_handler(query: str, limit: int = 10) -> Dict:
    """搜索处理函数"""
    return {
        "results": [
            {"title": f"Result {i}", "url": f"https://example.com/{i}"}
            for i in range(limit)
        ]
    }

tool_manager.register_tool(ToolDefinition(
    name="search",
    description="Search documents",
    parameters=[
        ToolParameter(
            name="query",
            type="string",
            description="Search query",
            required=True
        ),
        ToolParameter(
            name="limit",
            type="integer",
            description="Max results",
            required=False,
            default=10
        )
    ],
    handler=search_handler
))

# 执行工具
result = tool_manager.execute_tool("search", {"query": "AI", "limit": 5})

四、安全控制

4.1 认证授权

# mcp_security.py
from typing import Dict, List, Optional
from enum import Enum
import hashlib
import secrets

class AuthType(Enum):
    """认证类型"""
    API_KEY = "api_key"
    JWT = "jwt"
    OAUTH2 = "oauth2"

class Permission(Enum):
    """权限"""
    READ = "read"
    WRITE = "write"
    EXECUTE = "execute"
    ADMIN = "admin"

class SecurityManager:
    """安全管理器"""
    
    def __init__(self):
        self.api_keys: Dict[str, Dict] = {}
        self.permissions: Dict[str, List[Permission]] = {}
    
    def create_api_key(
        self,
        client_id: str,
        permissions: List[Permission]
    ) -> str:
        """创建 API Key"""
        api_key = secrets.token_urlsafe(32)
        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
        
        self.api_keys[key_hash] = {
            "client_id": client_id,
            "created_at": datetime.now().isoformat(),
            "last_used": None
        }
        
        self.permissions[key_hash] = permissions
        
        return api_key
    
    def validate_api_key(self, api_key: str) -> bool:
        """验证 API Key"""
        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
        
        if key_hash not in self.api_keys:
            return False
        
        # 更新最后使用时间
        self.api_keys[key_hash]["last_used"] = datetime.now().isoformat()
        
        return True
    
    def check_permission(
        self,
        api_key: str,
        required_permission: Permission
    ) -> bool:
        """检查权限"""
        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
        
        if key_hash not in self.permissions:
            return False
        
        return required_permission in self.permissions[key_hash]
    
    def revoke_api_key(self, api_key: str):
        """撤销 API Key"""
        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
        
        if key_hash in self.api_keys:
            del self.api_keys[key_hash]
        
        if key_hash in self.permissions:
            del self.permissions[key_hash]

# 使用示例
security = SecurityManager()

# 创建 API Key
api_key = security.create_api_key(
    client_id="client_123",
    permissions=[Permission.READ, Permission.EXECUTE]
)

# 验证 API Key
if security.validate_api_key(api_key):
    print("API Key valid")

# 检查权限
if security.check_permission(api_key, Permission.READ):
    print("Has read permission")

4.2 资源访问控制

# mcp_access_control.py
from typing import Dict, List, Set

class AccessControlList:
    """访问控制列表"""
    
    def __init__(self):
        self.rules: List[Dict] = []
    
    def add_rule(
        self,
        resource_pattern: str,
        client_id: str,
        permissions: List[str],
        effect: str = "allow"
    ):
        """添加规则"""
        self.rules.append({
            "resource_pattern": resource_pattern,
            "client_id": client_id,
            "permissions": permissions,
            "effect": effect
        })
    
    def check_access(
        self,
        resource_uri: str,
        client_id: str,
        action: str
    ) -> bool:
        """检查访问权限"""
        for rule in self.rules:
            if self._match_resource(resource_uri, rule["resource_pattern"]):
                if rule["client_id"] == client_id or rule["client_id"] == "*":
                    if action in rule["permissions"]:
                        return rule["effect"] == "allow"
        
        # 默认拒绝
        return False
    
    def _match_resource(self, uri: str, pattern: str) -> bool:
        """匹配资源"""
        import fnmatch
        return fnmatch.fnmatch(uri, pattern)

# 使用示例
acl = AccessControlList()

# 允许 client_123 读取所有文件
acl.add_rule(
    resource_pattern="file:///*",
    client_id="client_123",
    permissions=["read"],
    effect="allow"
)

# 禁止 client_456 访问敏感文件
acl.add_rule(
    resource_pattern="file:///sensitive/*",
    client_id="client_456",
    permissions=["read", "write"],
    effect="deny"
)

# 检查访问权限
if acl.check_access("file:///documents/report.pdf", "client_123", "read"):
    print("Access granted")

五、最佳实践

5.1 设计原则

MCP 设计原则：

1. 单一职责
   - 每个工具只做一件事
   - 资源定义清晰
   - 提示模板专注

2. 最小权限
   - 只请求必要权限
   - 限制资源访问范围
   - 定期审查权限

3. 错误处理
   - 明确的错误信息
   - 优雅的错误恢复
   - 完整的错误日志

4. 性能优化
   - 资源缓存
   - 工具执行超时
   - 并发控制

5.2 实施建议

# mcp_best_practices.py

class MCPBestPractices:
    """MCP 最佳实践"""
    
    @staticmethod
    def tool_design_tips() -> List[str]:
        """工具设计建议"""
        return [
            "保持工具简单专注",
            "提供清晰的描述",
            "定义完整的参数 schema",
            "处理所有可能的错误",
            "设置合理的超时时间"
        ]
    
    @staticmethod
    def resource_design_tips() -> List[str]:
        """资源设计建议"""
        return [
            "使用清晰的 URI 格式",
            "提供详细的元数据",
            "实现资源缓存",
            "限制资源大小",
            "支持分页访问"
        ]
    
    @staticmethod
    def security_tips() -> List[str]:
        """安全建议"""
        return [
            "始终验证 API Key",
            "实施访问控制",
            "记录所有操作",
            "定期轮换密钥",
            "监控异常行为"
        ]
    
    @staticmethod
    def performance_tips() -> List[str]:
        """性能建议"""
        return [
            "实现资源缓存",
            "使用连接池",
            "设置超时限制",
            "实现限流",
            "监控性能指标"
        ]

六、总结

6.1 核心要点

1. 服务器开发

   • 标准协议实现
   • 资源/工具/提示注册
   • 错误处理

2. 资源定义

   • 统一 URI 格式
   • 元数据完整
   • 访问控制

3. 安全控制

   • API Key 认证
   • 权限管理
   • 访问控制列表

6.2 最佳实践

1. 设计原则

   • 单一职责
   • 最小权限
   • 错误处理

2. 实施建议

   • 标准化
   • 文档化
   • 自动化测试

参考资料

• MCP Specification
• MCP SDK
• MCP Best Practices