深入理解MCP：Model Context Protocol架构解析与开发实践

1
2
3
4
5

🎯 MCP解决的核心问题：
1. 数据孤岛：AI模型被困在信息孤岛中
2. 集成复杂：每个数据源需要定制化实现
3. 扩展困难：连接系统难以规模化
4. 维护成本：碎片化集成难以维护

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

graph TB
    Host[MCP Host<br/>宿主应用] --> Client1[MCP Client 1]
    Host --> Client2[MCP Client 2]
    Host --> Client3[MCP Client 3]
    
    Client1 --> Server1[MCP Server 1<br/>文件系统]
    Client2 --> Server2[MCP Server 2<br/>数据库]
    Client3 --> Server3[MCP Server 3<br/>API服务]
    
    Server1 --> Data1[本地文件]
    Server2 --> Data2[PostgreSQL]
    Server3 --> Data3[外部API]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

🏠 定义：运行AI模型的应用程序
📋 功能：
- 管理与多个MCP服务器的连接
- 协调AI模型与外部数据的交互
- 处理用户请求和响应

💡 典型示例：
- Claude Desktop
- VS Code插件
- 自定义AI应用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

🔌 定义：协议客户端，维护与服务器的1:1连接
📋 功能：
- 处理协议通信
- 管理连接状态
- 数据序列化/反序列化

🛠️ 实现方式：
- 通常内嵌在Host应用中
- 使用官方SDK开发
- 支持多种传输协议

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

⚙️ 定义：轻量级程序，通过标准协议暴露特定能力
📋 功能：
- 提供Resources（资源）
- 暴露Tools（工具）
- 实现Prompts（提示模板）

🌟 特点：
- 单一职责原则
- 标准化接口
- 易于部署和维护

1
2
3
4
5
6

{
  "jsonrpc": "2.0",
  "method": "resources/list",
  "id": 1,
  "params": {}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 启动stdio传输的MCP服务器
import sys
import asyncio
from mcp.server.stdio import stdio_server

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await run_server(read_stream, write_stream)

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

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

# HTTP传输配置
from mcp.server.fastapi import create_mcp_app
from fastapi import FastAPI

app = FastAPI()
mcp_app = create_mcp_app(MyMcpServer())
app.mount("/mcp", mcp_app)

# 启动HTTP服务器
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# WebSocket实时通信
import websockets
from mcp.server.websocket import websocket_server

async def handle_websocket(websocket, path):
    async with websocket_server(websocket) as server:
        await server.run()

start_server = websockets.serve(handle_websocket, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_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

{
  "jsonrpc": "2.0",
  "method": "initialize",
  "id": 1,
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "resources": {
        "subscribe": true,
        "listChanged": true
      },
      "tools": {
        "listChanged": true
      },
      "prompts": {
        "listChanged": true
      },
      "logging": {},
      "sampling": {}
    },
    "clientInfo": {
      "name": "example-client",
      "version": "1.0.0"
    }
  }
}

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

sequenceDiagram
    participant Client
    participant Server
    
    Client->>Server: initialize request
    Server->>Client: initialize response
    Client->>Server: initialized notification
    Note over Client,Server: 连接建立完成
    
    Client->>Server: resources/list
    Server->>Client: resources list response
    
    Client->>Server: tools/list  
    Server->>Client: tools list response
    
    Note over Client,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

class ServerCapabilities:
    def __init__(self):
        self.resources = ResourceCapabilities(
            subscribe=True,        # 支持资源订阅
            listChanged=True      # 支持列表变更通知
        )
        self.tools = ToolCapabilities(
            listChanged=True      # 支持工具列表变更
        )
        self.prompts = PromptCapabilities(
            listChanged=True      # 支持提示模板变更
        )
        self.logging = LoggingCapabilities()      # 日志功能
        self.sampling = SamplingCapabilities()   # 采样功能

# 服务器能力声明
async def handle_initialize(self, params):
    return InitializeResult(
        protocolVersion="2024-11-05",
        capabilities=self.get_capabilities(),
        serverInfo=ServerInfo(
            name="advanced-mcp-server",
            version="2.0.0"
        )
    )

 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

from typing import Union, List, Dict, Any
from dataclasses import dataclass
from enum import Enum

class ResourceType(Enum):
    TEXT = "text"
    BINARY = "binary"
    JSON = "json"
    XML = "xml"
    CSV = "csv"
    IMAGE = "image"
    AUDIO = "audio"
    VIDEO = "video"

@dataclass
class ResourceMetadata:
    """资源元数据"""
    size: int
    lastModified: str
    encoding: str
    language: str = None
    tags: List[str] = None
    permissions: Dict[str, bool] = None

@dataclass
class AdvancedResource:
    uri: str
    name: str
    description: str
    mimeType: str
    resourceType: ResourceType
    metadata: ResourceMetadata
    annotations: Dict[str, Any] = None

 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

class AdvancedResourceServer(McpServer):
    def __init__(self):
        super().__init__("advanced-resource-server")
        self.resource_cache = {}
        self.resource_subscribers = set()
        
    async def list_resources(self) -> List[AdvancedResource]:
        """列出所有可用资源"""
        resources = []
        
        # 动态扫描文件系统
        for file_path in self.scan_directory("/data"):
            resource = await self.create_resource_from_file(file_path)
            resources.append(resource)
            
        # 数据库资源
        db_resources = await self.get_database_resources()
        resources.extend(db_resources)
        
        # API资源
        api_resources = await self.get_api_resources()
        resources.extend(api_resources)
        
        return resources
    
    async def read_resource(self, uri: str) -> Union[str, bytes]:
        """读取资源内容，支持缓存和压缩"""
        # 检查缓存
        if uri in self.resource_cache:
            cache_entry = self.resource_cache[uri]
            if not self.is_cache_expired(cache_entry):
                return cache_entry['content']
        
        # 根据URI类型读取资源
        if uri.startswith("file://"):
            content = await self.read_file_resource(uri)
        elif uri.startswith("db://"):
            content = await self.read_database_resource(uri)
        elif uri.startswith("api://"):
            content = await self.read_api_resource(uri)
        else:
            raise ValueError(f"不支持的资源URI: {uri}")
        
        # 更新缓存
        self.resource_cache[uri] = {
            'content': content,
            'timestamp': time.time(),
            'size': len(content) if isinstance(content, str) else len(content)
        }
        
        return content
    
    async def subscribe_to_resource(self, uri: str) -> None:
        """订阅资源变更通知"""
        self.resource_subscribers.add(uri)
        # 启动文件监控
        await self.start_file_watcher(uri)
    
    async def notify_resource_changed(self, uri: str):
        """通知资源变更"""
        if uri in self.resource_subscribers:
            await self.send_notification("notifications/resources/updated", {
                "uri": uri,
                "timestamp": time.time()
            })

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

# 文件系统资源服务器
from mcp import McpServer, Resource

class FileSystemServer(McpServer):
    async def list_resources(self):
        return [
            Resource(
                uri="file:///path/to/document.txt",
                name="项目文档",
                description="项目的详细说明文档",
                mimeType="text/plain"
            )
        ]
    
    async def read_resource(self, uri: str):
        if uri.startswith("file://"):
            path = uri[7:]  # 移除file://前缀
            with open(path, 'r', encoding='utf-8') as f:
                return f.read()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

📁 文件系统：
- 代码文件、文档、配置文件
- 支持多种文件格式（txt, md, json, etc.）

🗄️ 数据库：
- 表结构、查询结果
- 支持SQL和NoSQL数据库

🌐 Web资源：
- API响应、网页内容
- 实时数据抓取

 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

from pydantic import BaseModel, validator, Field
from typing import Optional, Union, List
import jsonschema

class AdvancedToolParameter(BaseModel):
    """高级工具参数定义"""
    name: str
    type: str
    description: str
    required: bool = True
    default: Optional[Any] = None
    enum: Optional[List[str]] = None
    pattern: Optional[str] = None
    minimum: Optional[Union[int, float]] = None
    maximum: Optional[Union[int, float]] = None
    
    @validator('type')
    def validate_type(cls, v):
        valid_types = ['string', 'number', 'integer', 'boolean', 'array', 'object']
        if v not in valid_types:
            raise ValueError(f'Invalid type: {v}')
        return v

class AdvancedTool(BaseModel):
    """高级工具定义"""
    name: str
    description: str
    parameters: List[AdvancedToolParameter]
    examples: Optional[List[Dict[str, Any]]] = None
    category: str = "general"
    tags: List[str] = []
    timeout: int = 30  # 超时时间（秒）
    retryable: bool = True
    permissions: List[str] = []
    
    def validate_input(self, arguments: Dict[str, Any]) -> None:
        """验证输入参数"""
        schema = self.generate_json_schema()
        try:
            jsonschema.validate(arguments, schema)
        except jsonschema.ValidationError as e:
            raise ValueError(f"参数验证失败: {e.message}")
    
    def generate_json_schema(self) -> Dict[str, Any]:
        """生成JSON Schema用于参数验证"""
        properties = {}
        required = []
        
        for param in self.parameters:
            prop = {
                "type": param.type,
                "description": param.description
            }
            
            if param.enum:
                prop["enum"] = param.enum
            if param.pattern:
                prop["pattern"] = param.pattern
            if param.minimum is not None:
                prop["minimum"] = param.minimum
            if param.maximum is not None:
                prop["maximum"] = param.maximum
            if param.default is not None:
                prop["default"] = param.default
                
            properties[param.name] = prop
            
            if param.required:
                required.append(param.name)
        
        return {
            "type": "object",
            "properties": properties,
            "required": required
        }

  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

import asyncio
from contextlib import asynccontextmanager
import time
from typing import AsyncGenerator

class ToolExecutionContext:
    """工具执行上下文"""
    def __init__(self, tool_name: str, client_id: str):
        self.tool_name = tool_name
        self.client_id = client_id
        self.start_time = time.time()
        self.execution_id = f"{tool_name}_{int(time.time())}"
        self.metadata = {}

class AdvancedToolServer(McpServer):
    def __init__(self):
        super().__init__("advanced-tool-server")
        self.tool_registry = {}
        self.execution_history = []
        self.concurrent_executions = {}
        
    def register_tool(self, tool: AdvancedTool, handler_func):
        """注册工具和处理函数"""
        self.tool_registry[tool.name] = {
            'tool': tool,
            'handler': handler_func
        }
    
    async def list_tools(self) -> List[AdvancedTool]:
        """列出所有注册的工具"""
        return [entry['tool'] for entry in self.tool_registry.values()]
    
    @asynccontextmanager
    async def execution_context(self, tool_name: str, client_id: str) -> AsyncGenerator[ToolExecutionContext, None]:
        """工具执行上下文管理器"""
        context = ToolExecutionContext(tool_name, client_id)
        self.concurrent_executions[context.execution_id] = context
        
        try:
            yield context
        finally:
            # 记录执行历史
            context.end_time = time.time()
            context.duration = context.end_time - context.start_time
            self.execution_history.append(context)
            
            # 清理并发执行记录
            del self.concurrent_executions[context.execution_id]
    
    async def call_tool(self, name: str, arguments: dict) -> ToolResult:
        """执行工具调用"""
        if name not in self.tool_registry:
            raise McpError(
                code=ErrorCode.METHOD_NOT_FOUND,
                message=f"工具 '{name}' 未找到"
            )
        
        tool_entry = self.tool_registry[name]
        tool = tool_entry['tool']
        handler = tool_entry['handler']
        
        # 参数验证
        try:
            tool.validate_input(arguments)
        except ValueError as e:
            raise McpError(
                code=ErrorCode.INVALID_PARAMS,
                message=str(e)
            )
        
        # 权限检查
        await self.check_tool_permissions(tool, self.get_current_client_id())
        
        # 执行工具
        async with self.execution_context(name, self.get_current_client_id()) as context:
            try:
                # 设置超时
                result = await asyncio.wait_for(
                    handler(arguments, context),
                    timeout=tool.timeout
                )
                return result
                
            except asyncio.TimeoutError:
                raise McpError(
                    code=ErrorCode.REQUEST_TIMEOUT,
                    message=f"工具执行超时 ({tool.timeout}秒)"
                )
            except Exception as e:
                if tool.retryable:
                    # 可重试的工具进行重试
                    return await self.retry_tool_execution(tool, handler, arguments, context)
                else:
                    raise McpError(
                        code=ErrorCode.INTERNAL_ERROR,
                        message=f"工具执行失败: {str(e)}"
                    )
    
    async def retry_tool_execution(self, tool: AdvancedTool, handler, arguments: dict, context: ToolExecutionContext, max_retries: int = 3) -> ToolResult:
        """重试工具执行"""
        for attempt in range(max_retries):
            try:
                await asyncio.sleep(2 ** attempt)  # 指数退避
                result = await asyncio.wait_for(
                    handler(arguments, context),
                    timeout=tool.timeout
                )
                return result
            except Exception as e:
                if attempt == max_retries - 1:
                    raise McpError(
                        code=ErrorCode.INTERNAL_ERROR,
                        message=f"工具执行失败（重试{max_retries}次后）: {str(e)}"
                    )
                continue

 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

class ToolWorkflow:
    """工具工作流编排"""
    def __init__(self, name: str):
        self.name = name
        self.steps = []
        self.conditions = {}
        
    def add_step(self, tool_name: str, arguments: dict, condition: Optional[callable] = None):
        """添加工作流步骤"""
        self.steps.append({
            'tool': tool_name,
            'arguments': arguments,
            'condition': condition
        })
    
    async def execute(self, tool_server: AdvancedToolServer, context: Dict[str, Any] = None) -> List[ToolResult]:
        """执行工作流"""
        results = []
        workflow_context = context or {}
        
        for step in self.steps:
            # 检查执行条件
            if step['condition'] and not step['condition'](workflow_context, results):
                continue
            
            # 动态参数替换
            arguments = self.resolve_arguments(step['arguments'], workflow_context, results)
            
            # 执行工具
            result = await tool_server.call_tool(step['tool'], arguments)
            results.append(result)
            
            # 更新上下文
            workflow_context[f"step_{len(results)}_result"] = result
        
        return results
    
    def resolve_arguments(self, arguments: dict, context: Dict[str, Any], results: List[ToolResult]) -> dict:
        """解析动态参数"""
        resolved = {}
        for key, value in arguments.items():
            if isinstance(value, str) and value.startswith("${"):
                # 动态参数解析
                var_name = value[2:-1]  # 移除 ${ 和 }
                if var_name in context:
                    resolved[key] = context[var_name]
                elif var_name.startswith("result_"):
                    # 引用之前的结果
                    index = int(var_name.split("_")[1]) - 1
                    if 0 <= index < len(results):
                        resolved[key] = results[index].content[0].text
                else:
                    resolved[key] = value
            else:
                resolved[key] = value
        return resolved

# 工作流使用示例
async def setup_code_analysis_workflow():
    workflow = ToolWorkflow("code_analysis")
    
    # 步骤1: 扫描代码文件
    workflow.add_step("scan_code_files", {
        "directory": "${target_directory}",
        "extensions": [".py", ".js", ".ts"]
    })
    
    # 步骤2: 分析代码复杂度
    workflow.add_step("analyze_complexity", {
        "files": "${result_1}"
    }, condition=lambda ctx, results: len(results) > 0 and results[0].isError == False)
    
    # 步骤3: 生成报告
    workflow.add_step("generate_report", {
        "analysis_data": "${result_2}",
        "format": "markdown"
    })
    
    return workflow

 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

from mcp import Tool, ToolResult

class DatabaseTool(McpServer):
    async def list_tools(self):
        return [
            Tool(
                name="execute_query",
                description="执行SQL查询",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "要执行的SQL查询"
                        }
                    },
                    "required": ["query"]
                }
            )
        ]
    
    async def call_tool(self, name: str, arguments: dict):
        if name == "execute_query":
            query = arguments["query"]
            # 执行数据库查询
            result = await self.db.execute(query)
            return ToolResult(
                content=[{"type": "text", "text": str(result)}],
                isError=False
            )

1
2
3
4
5
6
7
8
9

🔒 权限控制：
- 工具权限最小化原则
- 敏感操作需要确认
- 审计日志记录

⚠️ 输入验证：
- 参数类型检查
- SQL注入防护
- 文件路径验证

 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

from mcp import Prompt, PromptArgument

class CodeReviewServer(McpServer):
    async def list_prompts(self):
        return [
            Prompt(
                name="code_review",
                description="代码评审提示模板",
                arguments=[
                    PromptArgument(
                        name="code",
                        description="要评审的代码",
                        required=True
                    ),
                    PromptArgument(
                        name="language",
                        description="编程语言",
                        required=False
                    )
                ]
            )
        ]
    
    async def get_prompt(self, name: str, arguments: dict):
        if name == "code_review":
            code = arguments["code"]
            language = arguments.get("language", "未知")
            
            return f"""
请对以下{language}代码进行详细评审：

代码内容：
```{language}
{code}

 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

## 🔒 MCP安全与中间件系统

### 认证与授权中间件

#### JWT认证中间件
```python
import jwt
from functools import wraps
from typing import Dict, Any, Optional

class JWTAuthMiddleware:
    def __init__(self, secret_key: str, algorithm: str = "HS256"):
        self.secret_key = secret_key
        self.algorithm = algorithm
    
    def verify_token(self, token: str) -> Optional[Dict[str, Any]]:
        """验证JWT令牌"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except jwt.ExpiredSignatureError:
            raise McpError(code=ErrorCode.UNAUTHORIZED, message="令牌已过期")
        except jwt.InvalidTokenError:
            raise McpError(code=ErrorCode.UNAUTHORIZED, message="无效令牌")
    
    def require_permission(self, permission: str):
        """权限装饰器"""
        def decorator(func):
            @wraps(func)
            async def wrapper(self, *args, **kwargs):
                client_token = self.get_client_token()
                payload = self.verify_token(client_token)
                
                user_permissions = payload.get('permissions', [])
                if permission not in user_permissions:
                    raise McpError(
                        code=ErrorCode.PERMISSION_DENIED,
                        message=f"缺少权限: {permission}"
                    )
                
                return await func(self, *args, **kwargs)
            return wrapper
        return decorator

class SecureServer(McpServer):
    def __init__(self):
        super().__init__("secure-server")
        self.auth_middleware = JWTAuthMiddleware(os.getenv("JWT_SECRET"))
    
    @auth_middleware.require_permission("read:resources")
    async def read_resource(self, uri: str):
        return await super().read_resource(uri)
    
    @auth_middleware.require_permission("execute:tools")
    async def call_tool(self, name: str, arguments: dict):
        return await super().call_tool(name, arguments)

 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

import time
from collections import defaultdict, deque
from typing import DefaultDict

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: DefaultDict[str, deque] = defaultdict(deque)
    
    def is_allowed(self, client_id: str) -> bool:
        """检查是否允许请求"""
        now = time.time()
        client_requests = self.requests[client_id]
        
        # 清理过期请求
        while client_requests and client_requests[0] <= now - self.window_seconds:
            client_requests.popleft()
        
        # 检查请求数量
        if len(client_requests) >= self.max_requests:
            return False
        
        # 记录新请求
        client_requests.append(now)
        return True

class RateLimitedServer(McpServer):
    def __init__(self):
        super().__init__("rate-limited-server")
        self.rate_limiter = RateLimiter(max_requests=100, window_seconds=60)
    
    async def handle_request(self, method: str, params: dict):
        client_id = self.get_current_client_id()
        
        if not self.rate_limiter.is_allowed(client_id):
            raise McpError(
                code=ErrorCode.RATE_LIMITED,
                message="请求频率超限，请稍后重试"
            )
        
        return await super().handle_request(method, params)

 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

from cryptography.fernet import Fernet
import base64
import json

class FieldEncryption:
    def __init__(self, encryption_key: bytes):
        self.cipher_suite = Fernet(encryption_key)
    
    def encrypt_field(self, data: str) -> str:
        """加密字段"""
        encrypted_data = self.cipher_suite.encrypt(data.encode())
        return base64.b64encode(encrypted_data).decode()
    
    def decrypt_field(self, encrypted_data: str) -> str:
        """解密字段"""
        decoded_data = base64.b64decode(encrypted_data.encode())
        decrypted_data = self.cipher_suite.decrypt(decoded_data)
        return decrypted_data.decode()

class DataMasking:
    """数据脱敏工具"""
    
    @staticmethod
    def mask_email(email: str) -> str:
        """邮箱脱敏"""
        if '@' not in email:
            return email
        local, domain = email.split('@', 1)
        masked_local = local[:2] + '*' * (len(local) - 2)
        return f"{masked_local}@{domain}"
    
    @staticmethod
    def mask_phone(phone: str) -> str:
        """电话号码脱敏"""
        if len(phone) < 7:
            return phone
        return phone[:3] + '*' * (len(phone) - 6) + phone[-3:]
    
    @staticmethod
    def mask_id_card(id_card: str) -> str:
        """身份证脱敏"""
        if len(id_card) < 8:
            return id_card
        return id_card[:4] + '*' * (len(id_card) - 8) + id_card[-4:]

class SecureDataServer(McpServer):
    def __init__(self):
        super().__init__("secure-data-server")
        self.field_encryption = FieldEncryption(Fernet.generate_key())
        self.data_masking = DataMasking()
    
    async def read_resource(self, uri: str) -> str:
        """读取资源并进行数据脱敏"""
        raw_data = await self.load_raw_data(uri)
        
        # 根据资源类型进行脱敏
        if uri.startswith("user://"):
            return self.mask_user_data(raw_data)
        elif uri.startswith("sensitive://"):
            return self.decrypt_and_mask_data(raw_data)
        
        return raw_data
    
    def mask_user_data(self, data: str) -> str:
        """用户数据脱敏"""
        try:
            user_data = json.loads(data)
            if 'email' in user_data:
                user_data['email'] = self.data_masking.mask_email(user_data['email'])
            if 'phone' in user_data:
                user_data['phone'] = self.data_masking.mask_phone(user_data['phone'])
            if 'id_card' in user_data:
                user_data['id_card'] = self.data_masking.mask_id_card(user_data['id_card'])
            return json.dumps(user_data, ensure_ascii=False)
        except json.JSONDecodeError:
            return data

 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

class CustomProtocolExtension:
    """自定义协议扩展"""
    
    def __init__(self, namespace: str):
        self.namespace = namespace
        self.custom_methods = {}
    
    def register_method(self, method_name: str, handler):
        """注册自定义方法"""
        full_method_name = f"{self.namespace}/{method_name}"
        self.custom_methods[full_method_name] = handler
    
    async def handle_custom_request(self, method: str, params: dict):
        """处理自定义请求"""
        if method in self.custom_methods:
            return await self.custom_methods[method](params)
        else:
            raise McpError(
                code=ErrorCode.METHOD_NOT_FOUND,
                message=f"自定义方法未找到: {method}"
            )

class ExtendedServer(McpServer):
    def __init__(self):
        super().__init__("extended-server")
        self.extension = CustomProtocolExtension("mycompany")
        self.setup_custom_methods()
    
    def setup_custom_methods(self):
        """设置自定义方法"""
        self.extension.register_method("bulk_upload", self.handle_bulk_upload)
        self.extension.register_method("stream_data", self.handle_stream_data)
        self.extension.register_method("batch_process", self.handle_batch_process)
    
    async def handle_bulk_upload(self, params: dict):
        """处理批量上传"""
        files = params.get('files', [])
        results = []
        
        for file_info in files:
            result = await self.upload_single_file(file_info)
            results.append(result)
        
        return {
            "uploaded": len(results),
            "results": results
        }
    
    async def handle_stream_data(self, params: dict):
        """处理流数据"""
        stream_id = params.get('stream_id')
        chunk_size = params.get('chunk_size', 1024)
        
        async for chunk in self.get_data_stream(stream_id, chunk_size):
            yield {
                "stream_id": stream_id,
                "chunk": chunk,
                "timestamp": time.time()
            }

  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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295

import asyncio
from typing import Callable, List
from dataclasses import dataclass

@dataclass
class McpEvent:
    """MCP事件"""
    type: str
    data: dict
    timestamp: float
    source: str

class EventBus:
    """事件总线"""
    
    def __init__(self):
        self.subscribers: Dict[str, List[Callable]] = defaultdict(list)
        self.event_history: List[McpEvent] = []
    
    def subscribe(self, event_type: str, handler: Callable):
        """订阅事件"""
        self.subscribers[event_type].append(handler)
    
    def unsubscribe(self, event_type: str, handler: Callable):
        """取消订阅"""
        if handler in self.subscribers[event_type]:
            self.subscribers[event_type].remove(handler)
    
    async def publish(self, event: McpEvent):
        """发布事件"""
        self.event_history.append(event)
        
        # 通知所有订阅者
        handlers = self.subscribers.get(event.type, [])
        if handlers:
            await asyncio.gather(*[handler(event) for handler in handlers])
    
    def get_event_history(self, event_type: str = None, limit: int = 100) -> List[McpEvent]:
        """获取事件历史"""
        events = self.event_history
        if event_type:
            events = [e for e in events if e.type == event_type]
        return events[-limit:]

class EventDrivenServer(McpServer):
    def __init__(self):
        super().__init__("event-driven-server")
        self.event_bus = EventBus()
        self.setup_event_handlers()
    
    def setup_event_handlers(self):
        """设置事件处理器"""
        self.event_bus.subscribe("resource.created", self.on_resource_created)
        self.event_bus.subscribe("resource.updated", self.on_resource_updated)
        self.event_bus.subscribe("tool.executed", self.on_tool_executed)
        self.event_bus.subscribe("client.connected", self.on_client_connected)
    
    async def on_resource_created(self, event: McpEvent):
        """资源创建事件处理"""
        resource_uri = event.data.get('uri')
        await self.send_notification("notifications/resources/created", {
            "uri": resource_uri,
            "timestamp": event.timestamp
        })
    
    async def on_tool_executed(self, event: McpEvent):
        """工具执行事件处理"""
        tool_name = event.data.get('tool_name')
        execution_time = event.data.get('duration')
        
        # 记录性能指标
        await self.record_performance_metric(tool_name, execution_time)
        
        # 如果执行时间过长，发送警告
        if execution_time > 10:  # 10秒
            await self.send_notification("notifications/performance/warning", {
                "tool": tool_name,
                "duration": execution_time,
                "message": "工具执行时间过长"
            })

### 分布式MCP架构

#### 服务发现与负载均衡
```python
import consul
import hashlib
from typing import List, Dict, Optional

class ServiceDiscovery:
    """服务发现"""
    
    def __init__(self, consul_host: str = "localhost", consul_port: int = 8500):
        self.consul = consul.Consul(host=consul_host, port=consul_port)
        self.service_cache = {}
        self.cache_ttl = 30  # 缓存30秒
    
    async def register_service(self, service_name: str, host: str, port: int, tags: List[str] = None):
        """注册服务"""
        service_id = f"{service_name}-{host}-{port}"
        
        self.consul.agent.service.register(
            name=service_name,
            service_id=service_id,
            address=host,
            port=port,
            tags=tags or [],
            check=consul.Check.http(f"http://{host}:{port}/health", interval="10s")
        )
    
    async def discover_services(self, service_name: str) -> List[Dict]:
        """发现服务"""
        # 检查缓存
        cache_key = f"services:{service_name}"
        if cache_key in self.service_cache:
            cached_data = self.service_cache[cache_key]
            if time.time() - cached_data['timestamp'] < self.cache_ttl:
                return cached_data['services']
        
        # 从Consul获取服务列表
        _, services = self.consul.health.service(service_name, passing=True)
        
        service_list = []
        for service in services:
            service_info = service['Service']
            service_list.append({
                'id': service_info['ID'],
                'address': service_info['Address'],
                'port': service_info['Port'],
                'tags': service_info['Tags']
            })
        
        # 更新缓存
        self.service_cache[cache_key] = {
            'services': service_list,
            'timestamp': time.time()
        }
        
        return service_list

class LoadBalancer:
    """负载均衡器"""
    
    def __init__(self, strategy: str = "round_robin"):
        self.strategy = strategy
        self.round_robin_index = 0
        self.service_weights = {}
        self.service_health = {}
    
    def select_service(self, services: List[Dict], client_id: str = None) -> Optional[Dict]:
        """选择服务实例"""
        if not services:
            return None
        
        # 过滤健康的服务
        healthy_services = [s for s in services if self.service_health.get(s['id'], True)]
        
        if not healthy_services:
            return None
        
        if self.strategy == "round_robin":
            return self._round_robin_select(healthy_services)
        elif self.strategy == "weighted":
            return self._weighted_select(healthy_services)
        elif self.strategy == "consistent_hash":
            return self._consistent_hash_select(healthy_services, client_id)
        else:
            return healthy_services[0]
    
    def _round_robin_select(self, services: List[Dict]) -> Dict:
        """轮询选择"""
        service = services[self.round_robin_index % len(services)]
        self.round_robin_index += 1
        return service
    
    def _weighted_select(self, services: List[Dict]) -> Dict:
        """加权选择"""
        # 根据服务权重选择
        total_weight = sum(self.service_weights.get(s['id'], 1) for s in services)
        import random
        target = random.uniform(0, total_weight)
        
        current_weight = 0
        for service in services:
            current_weight += self.service_weights.get(service['id'], 1)
            if current_weight >= target:
                return service
        
        return services[0]
    
    def _consistent_hash_select(self, services: List[Dict], client_id: str) -> Dict:
        """一致性哈希选择"""
        if not client_id:
            return services[0]
        
        # 计算客户端ID的哈希值
        client_hash = int(hashlib.md5(client_id.encode()).hexdigest(), 16)
        
        # 为每个服务计算哈希值并找到最接近的
        service_hashes = []
        for service in services:
            service_key = f"{service['address']}:{service['port']}"
            service_hash = int(hashlib.md5(service_key.encode()).hexdigest(), 16)
            service_hashes.append((service_hash, service))
        
        # 排序并找到大于等于客户端哈希的第一个服务
        service_hashes.sort()
        for service_hash, service in service_hashes:
            if service_hash >= client_hash:
                return service
        
        # 如果没找到，返回第一个服务（环形结构）
        return service_hashes[0][1]

class DistributedMcpClient:
    """分布式MCP客户端"""
    
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.service_discovery = ServiceDiscovery()
        self.load_balancer = LoadBalancer(strategy="consistent_hash")
        self.circuit_breakers = {}
    
    async def call_service(self, method: str, params: dict, client_id: str = None):
        """调用分布式服务"""
        # 服务发现
        services = await self.service_discovery.discover_services(self.service_name)
        
        # 负载均衡选择服务
        selected_service = self.load_balancer.select_service(services, client_id)
        
        if not selected_service:
            raise McpError(
                code=ErrorCode.SERVICE_UNAVAILABLE,
                message=f"没有可用的 {self.service_name} 服务实例"
            )
        
        # 检查熔断器
        service_key = f"{selected_service['address']}:{selected_service['port']}"
        circuit_breaker = self.circuit_breakers.get(service_key)
        
        if circuit_breaker and circuit_breaker.is_open():
            raise McpError(
                code=ErrorCode.SERVICE_UNAVAILABLE,
                message=f"服务 {service_key} 熔断中"
            )
        
        try:
            # 调用远程服务
            result = await self.call_remote_service(selected_service, method, params)
            
            # 记录成功调用
            if circuit_breaker:
                circuit_breaker.record_success()
            
            return result
            
        except Exception as e:
            # 记录失败调用
            if circuit_breaker:
                circuit_breaker.record_failure()
            
            raise e

class CircuitBreaker:
    """熔断器"""
    
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    def is_open(self) -> bool:
        """检查熔断器是否开启"""
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
                return False
            return True
        return False
    
    def record_success(self):
        """记录成功调用"""
        self.failure_count = 0
        self.state = "CLOSED"
    
    def record_failure(self):
        """记录失败调用"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"