常见 MCP 设计反模式总结

# 反模式：硬编码 Tool 定义
@app.post("/api/v1/tools/execute")
async def execute_tool(tool_call: ToolCall):
    if tool_call.tool_name == "file_reader":
        # 硬编码处理 file_reader 工具
        file_path = tool_call.params.get("file_path")
        if not file_path:
            return {"error": "Missing file_path parameter"}
        # 执行文件读取操作
        return {"result": read_file(file_path)}
    elif tool_call.tool_name == "database_query":
        # 硬编码处理 database_query 工具
        query = tool_call.params.get("query")
        if not query:
            return {"error": "Missing query parameter"}
        # 执行数据库查询操作
        return {"result": execute_query(query)}
    # 更多硬编码工具...
    else:
        return {"error": "Unknown tool"}

# 最佳实践：动态 Tool 注册机制
class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register_tool(self, tool: ToolDefinition):
        self.tools[tool.name] = tool
    
    def get_tool(self, name: str) -> Optional[ToolDefinition]:
        return self.tools.get(name)

# 工具注册示例
registry = ToolRegistry()

# 从配置文件加载工具定义
def load_tools_from_config(config_path: str):
    with open(config_path, "r") as f:
        tools_config = json.load(f)
    for tool_config in tools_config["tools"]:
        tool = ToolDefinition(**tool_config)
        registry.register_tool(tool)

# API 端点
def execute_tool(tool_call: ToolCall):
    tool = registry.get_tool(tool_call.tool_name)
    if not tool:
        return {"error": "Unknown tool"}
    # 验证参数
    if not validate_params(tool_call.params, tool.schema):
        return {"error": "Invalid parameters"}
    # 执行工具
    result = tool.execute(tool_call.params)
    return {"result": result}

// 反模式：忽略能力协商，直接调用工具
async function callMcpTool(toolName, params) {
    // 直接发送工具调用请求，不进行能力协商
    const response = await fetch('http://localhost:8000/api/v1/tools/execute', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json'
        },
        body: JSON.stringify({
            tool_name: toolName,
            params: params
        })
    });
    return await response.json();
}

// 最佳实践：实现完整的能力协商
class McpClient {
    constructor(serverUrl) {
        this.serverUrl = serverUrl;
        this.supportedCapabilities = null;
    }
    
    // 获取服务器能力
    async getServerCapabilities() {
        if (this.supportedCapabilities) {
            return this.supportedCapabilities;
        }
        
        const response = await fetch(`${this.serverUrl}/api/v1/capabilities`);
        this.supportedCapabilities = await response.json();
        return this.supportedCapabilities;
    }
    
    // 检查工具是否可用
    async isToolAvailable(toolName) {
        const capabilities = await this.getServerCapabilities();
        return capabilities.tools.includes(toolName);
    }
    
    // 调用工具
    async callTool(toolName, params) {
        // 检查工具是否可用
        if (!(await this.isToolAvailable(toolName))) {
            throw new Error(`Tool ${toolName} is not available`);
        }
        
        // 发送工具调用请求
        const response = await fetch(`${this.serverUrl}/api/v1/tools/execute`, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json'
            },
            body: JSON.stringify({
                tool_name: toolName,
                params: params
            })
        });
        
        return await response.json();
    }
}

# 反模式：过度信任模型输出
async def process_model_response(response):
    # 直接解析模型输出，不进行验证
    tool_calls = parse_tool_calls(response)
    for tool_call in tool_calls:
        # 直接调用工具，不检查权限或参数
        result = await call_mcp_tool(tool_call)
        # 处理结果...

# 最佳实践：多层验证机制
async def process_model_response(response, context):
    # 1. 解析模型输出
    tool_calls = parse_tool_calls(response)
    
    for tool_call in tool_calls:
        # 2. 检查模型是否有权限调用该工具
        if not check_tool_permission(context.user_id, tool_call.tool_name):
            raise PermissionError(f"User {context.user_id} has no permission to call {tool_call.tool_name}")
        
        # 3. 验证参数格式和范围
        if not validate_tool_params(tool_call.tool_name, tool_call.params):
            raise ValueError(f"Invalid parameters for tool {tool_call.tool_name}")
        
        # 4. 检查参数是否包含恶意内容
        if contains_malicious_content(tool_call.params):
            raise SecurityError(f"Malicious content detected in parameters")
        
        # 5. 速率限制检查
        if not check_rate_limit(context.user_id, tool_call.tool_name):
            raise RateLimitError(f"Rate limit exceeded for tool {tool_call.tool_name}")
        
        # 6. 调用工具
        result = await call_mcp_tool(tool_call)
        # 处理结果...

# 反模式：资源访问控制缺失
@app.post("/api/v1/tools/file_reader")
async def file_reader(params: dict):
    file_path = params.get("file_path")
    # 直接打开文件，不检查路径是否合法
    with open(file_path, "r") as f:
        content = f.read()
    return {"content": content}

# 最佳实践：严格的资源访问控制
@app.post("/api/v1/tools/file_reader")
async def file_reader(params: dict):
    file_path = params.get("file_path")
    
    # 1. 验证路径格式
    if not file_path or not isinstance(file_path, str):
        return {"error": "Invalid file path"}
    
    # 2. 防止路径遍历攻击
    base_dir = Path("/safe/files")
    full_path = base_dir / file_path
    # 确保路径在允许的目录内
    if not str(full_path).startswith(str(base_dir)):
        return {"error": "Access denied"}
    
    # 3. 检查文件是否存在
    if not full_path.exists():
        return {"error": "File not found"}
    
    # 4. 检查文件类型和大小
    if full_path.is_dir():
        return {"error": "Not a file"}
    if full_path.stat().st_size > 10 * 1024 * 1024:  # 限制文件大小为 10MB
        return {"error": "File too large"}
    
    # 5. 读取文件
    with open(full_path, "r") as f:
        content = f.read()
    
    return {"content": content}

// 反模式：状态管理混乱
class McpServer {
    constructor() {
        this.connections = []; // 存储所有连接
        this.activeCalls = {}; // 存储活跃的调用
    }
    
    // 处理新连接
    handleConnection(connection) {
        this.connections.push(connection);
        
        connection.on('message', (message) => {
            const data = JSON.parse(message);
            if (data.type === 'tool_call') {
                // 存储活跃调用，但没有清理机制
                this.activeCalls[data.call_id] = {
                    connection: connection,
                    startTime: Date.now()
                };
                // 处理工具调用...
            }
        });
        
        connection.on('close', () => {
            // 清理连接，但没有清理相关的活跃调用
            const index = this.connections.indexOf(connection);
            if (index > -1) {
                this.connections.splice(index, 1);
            }
        });
    }
}

// 最佳实践：清晰的状态管理
class McpServer {
    constructor() {
        this.connections = new Map(); // 使用 Map 存储连接
        this.activeCalls = new Map(); // 使用 Map 存储活跃调用
        this.callTimeout = 30000; // 调用超时时间（30秒）
        
        // 定期清理超时调用
        setInterval(() => this.cleanupTimeoutCalls(), 60000);
    }
    
    // 处理新连接
    handleConnection(connection) {
        const connectionId = generateUniqueId();
        this.connections.set(connectionId, {
            connection: connection,
            connectedAt: Date.now()
        });
        
        connection.on('message', (message) => {
            const data = JSON.parse(message);
            if (data.type === 'tool_call') {
                // 存储活跃调用
                this.activeCalls.set(data.call_id, {
                    connectionId: connectionId,
                    startTime: Date.now()
                });
                // 处理工具调用...
            }
        });
        
        connection.on('close', () => {
            // 清理连接和相关的活跃调用
            this.connections.delete(connectionId);
            // 清理该连接的所有活跃调用
            for (const [callId, callInfo] of this.activeCalls.entries()) {
                if (callInfo.connectionId === connectionId) {
                    this.activeCalls.delete(callId);
                }
            }
        });
    }
    
    // 清理超时调用
    cleanupTimeoutCalls() {
        const now = Date.now();
        for (const [callId, callInfo] of this.activeCalls.entries()) {
            if (now - callInfo.startTime > this.callTimeout) {
                // 通知客户端调用超时
                const connectionInfo = this.connections.get(callInfo.connectionId);
                if (connectionInfo) {
                    connectionInfo.connection.send(JSON.stringify({
                        type: 'call_timeout',
                        call_id: callId
                    }));
                }
                // 删除超时调用
                this.activeCalls.delete(callId);
            }
        }
    }
}

# 反模式：忽略错误处理
def execute_database_query(params):
    query = params.get("query")
    # 直接执行查询，不处理异常
    result = db.execute(query)
    return result

@app.post("/api/v1/tools/database_query")
async def database_query(params: dict):
    result = execute_database_query(params)
    return {"result": result}

# 最佳实践：完善的错误处理
def execute_database_query(params):
    query = params.get("query")
    try:
        # 验证查询类型和复杂度
        if not is_valid_query(query):
            raise ValueError("Invalid query")
        
        # 检查查询复杂度
        if is_complex_query(query):
            raise ValueError("Query too complex")
        
        # 执行查询
        result = db.execute(query)
        return result
    except ValueError as e:
        # 客户端错误
        raise ClientError(str(e))
    except DatabaseError as e:
        # 数据库错误
        logger.error(f"Database error: {e}")
        raise ServerError("Database operation failed")
    except Exception as e:
        # 其他未预期的错误
        logger.error(f"Unexpected error: {e}")
        raise ServerError("Internal server error")

@app.post("/api/v1/tools/database_query")
async def database_query(params: dict):
    try:
        result = execute_database_query(params)
        return {"result": result}
    except ClientError as e:
        return {"error": str(e)}, 400
    except ServerError as e:
        return {"error": str(e)}, 500
    except Exception as e:
        return {"error": "Internal server error"}, 500

# 反模式：缺乏可观测性
@app.post("/api/v1/tools/execute")
async def execute_tool(tool_call: ToolCall):
    # 直接执行工具，没有日志或指标
    result = await execute_tool_internal(tool_call)
    return {"result": result}

# 最佳实践：完善的可观测性
from prometheus_client import Counter, Histogram
from opentelemetry import trace

# 初始化指标
tool_call_counter = Counter('mcp_tool_calls_total', 'Total number of tool calls', ['tool_name', 'status'])
tool_call_duration = Histogram('mcp_tool_call_duration_seconds', 'Duration of tool calls in seconds', ['tool_name'])

# 初始化追踪
 tracer = trace.get_tracer(__name__)

@app.post("/api/v1/tools/execute")
async def execute_tool(tool_call: ToolCall):
    # 生成追踪上下文
    with tracer.start_as_current_span("execute_tool") as span:
        span.set_attribute("tool.name", tool_call.tool_name)
        span.set_attribute("params", str(tool_call.params))
        
        # 记录日志
        logger.info(f"Received tool call: {tool_call.tool_name}, params: {tool_call.params}")
        
        # 计时开始
        start_time = time.time()
        
        try:
            # 执行工具
            result = await execute_tool_internal(tool_call)
            
            # 计算持续时间
            duration = time.time() - start_time
            
            # 记录成功指标
            tool_call_counter.labels(tool_call.tool_name, "success").inc()
            tool_call_duration.labels(tool_call.tool_name).observe(duration)
            
            # 记录成功日志
            logger.info(f"Tool call succeeded: {tool_call.tool_name}, duration: {duration:.2f}s")
            
            # 记录追踪结果
            span.set_attribute("result", str(result))
            span.set_status(trace.StatusCode.OK)
            
            return {"result": result}
        except Exception as e:
            # 计算持续时间
            duration = time.time() - start_time
            
            # 记录失败指标
            tool_call_counter.labels(tool_call.tool_name, "failure").inc()
            tool_call_duration.labels(tool_call.tool_name).observe(duration)
            
            # 记录失败日志
            logger.error(f"Tool call failed: {tool_call.tool_name}, error: {str(e)}, duration: {duration:.2f}s")
            
            # 记录追踪错误
            span.record_exception(e)
            span.set_status(trace.StatusCode.ERROR)
            
            raise

# 反模式：不遵循单一职责原则
class McpServer:
    def __init__(self):
        self.tools = {}
        self.connections = []
        self.database = DatabaseConnection()
        self.cache = Cache()
    
    # 处理 HTTP 请求
    async def handle_http_request(self, request):
        # 解析请求
        # 验证权限
        # 执行工具
        # 返回响应
        pass
    
    # 处理 WebSocket 连接
    def handle_websocket_connection(self, connection):
        # 处理连接
        # 处理消息
        # 处理断开
        pass
    
    # 管理工具
    def register_tool(self, tool):
        self.tools[tool.name] = tool
    
    # 访问数据库
    async def query_database(self, query):
        return await self.database.execute(query)
    
    # 管理缓存
    def get_from_cache(self, key):
        return self.cache.get(key)
    
    def set_to_cache(self, key, value):
        self.cache.set(key, value)

# 最佳实践：遵循单一职责原则

# 1. Tool 注册表
class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register_tool(self, tool):
        self.tools[tool.name] = tool
    
    def get_tool(self, name):
        return self.tools.get(name)

# 2. HTTP 处理
class HttpHandler:
    def __init__(self, tool_registry, permission_manager):
        self.tool_registry = tool_registry
        self.permission_manager = permission_manager
    
    async def handle_request(self, request):
        # 处理 HTTP 请求
        pass

# 3. WebSocket 处理
class WebSocketHandler:
    def __init__(self, tool_registry, permission_manager):
        self.tool_registry = tool_registry
        self.permission_manager = permission_manager
    
    def handle_connection(self, connection):
        # 处理 WebSocket 连接
        pass

# 4. 权限管理
class PermissionManager:
    def check_permission(self, user_id, action):
        # 检查权限
        pass

# 5. 主服务器
class McpServer:
    def __init__(self):
        self.tool_registry = ToolRegistry()
        self.permission_manager = PermissionManager()
        self.http_handler = HttpHandler(self.tool_registry, self.permission_manager)
        self.websocket_handler = WebSocketHandler(self.tool_registry, self.permission_manager)
    
    # 启动服务器
    def start(self):
        # 启动 HTTP 和 WebSocket 服务器
        pass

# 最佳实践：遵循安全最佳实践
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from pydantic import BaseModel, Field
from typing import Optional

# 配置
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

# OAuth2 密码承载令牌
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 工具调用模型
class ToolCall(BaseModel):
    tool_name: str = Field(..., min_length=1, max_length=100)
    params: dict = Field(..., description="Tool parameters")

# 验证令牌
def verify_token(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    return user_id

# 验证参数
def validate_params(tool_name: str, params: dict):
    # 具体的参数验证逻辑
    # 1. 检查参数是否符合 JSON Schema
    # 2. 检查参数值是否在允许范围内
    # 3. 检查参数是否包含恶意内容
    # 4. 检查参数长度和大小
    pass

# API 端点
@app.post("/api/v1/tools/execute")
async def execute_tool(tool_call: ToolCall, user_id: str = Depends(verify_token)):
    # 1. 验证工具是否存在
    tool = tool_registry.get_tool(tool_call.tool_name)
    if not tool:
        raise HTTPException(status_code=404, detail="Tool not found")
    
    # 2. 验证用户是否有权限调用该工具
    if not permission_manager.check_permission(user_id, f"tool:{tool_call.tool_name}:execute"):
        raise HTTPException(status_code=403, detail="Access denied")
    
    # 3. 验证参数
    if not validate_params(tool_call.tool_name, tool_call.params):
        raise HTTPException(status_code=400, detail="Invalid parameters")
    
    # 4. 执行工具
    result = await tool.execute(tool_call.params)
    
    # 5. 记录安全日志
    security_logger.info(f"User {user_id} executed tool {tool_call.tool_name}, params: {tool_call.params}")
    
    # 6. 返回结果
    return {"result": result}