每天10分钟轻松掌握MCP 40天学习计划的第6天

JSON-RPC通信协议在MCP中的应用机制（三）

3. 网络通信基础与优化

3.1 传输层协议选择对比

3.2 网络优化实现

import gzip
import json
import ssl
import websockets
from typing import Dict, Any, Optional, Callable
import logging

class OptimizedMCPTransport:
    """优化的MCP传输层"""
    
    def __init__(self, enable_compression: bool = True, 
                 max_message_size: int = 1024*1024,  # 1MB
                 heartbeat_interval: int = 30):
        self.enable_compression = enable_compression
        self.max_message_size = max_message_size
        self.heartbeat_interval = heartbeat_interval
        self.connection_pool: Dict[str, Any] = {}
        self.metrics = {
            "messages_sent": 0,
            "messages_received": 0,
            "bytes_sent": 0,
            "bytes_received": 0,
            "compression_ratio": 0.0,
            "connection_count": 0
        }
        
        # 配置日志
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    async def send_message(self, websocket, message: Dict[str, Any]) -> None:
        """发送优化的消息"""
        try:
            # 序列化消息
            json_data = json.dumps(message, ensure_ascii=False)
            original_size = len(json_data.encode('utf-8'))
            
            # 检查消息大小
            if original_size > self.max_message_size:
                self.logger.warning(f"消息大小 {original_size} 超过限制 {self.max_message_size}")
                raise ValueError("Message too large")
            
            # 压缩处理
            if self.enable_compression and original_size > 1024:  # 超过1KB才压缩
                compressed_data = gzip.compress(json_data.encode('utf-8'))
                compressed_size = len(compressed_data)
                
                if compressed_size < original_size * 0.9:  # 压缩率超过10%才使用
                    # 发送压缩数据 (实际实现中需要协商压缩协议)
                    await websocket.send(compressed_data)
                    
                    # 更新压缩统计
                    compression_ratio = (original_size - compressed_size) / original_size
                    self.metrics["compression_ratio"] = (
                        self.metrics["compression_ratio"] * self.metrics["messages_sent"] + compression_ratio
                    ) / (self.metrics["messages_sent"] + 1)
                    
                    self.logger.info(f"消息压缩: {original_size} -> {compressed_size} "
                                   f"(压缩率: {compression_ratio:.2%})")
                else:
                    await websocket.send(json_data)
            else:
                await websocket.send(json_data)
            
            # 更新统计
            self.metrics["messages_sent"] += 1
            self.metrics["bytes_sent"] += original_size
            
        except Exception as e:
            self.logger.error(f"发送消息失败: {e}")
            raise
    
    async def receive_message(self, websocket) -> Dict[str, Any]:
        """接收优化的消息"""
        try:
            raw_data = await websocket.recv()
            
            # 尝试解压缩 (简化处理，实际需要检查消息头)
            if isinstance(raw_data, bytes):
                try:
                    # 尝试解压缩
                    decompressed_data = gzip.decompress(raw_data)
                    json_data = decompressed_data.decode('utf-8')
                    self.logger.info(f"收到压缩消息: {len(raw_data)} -> {len(decompressed_data)}")
                except:
                    # 不是压缩数据，直接解码
                    json_data = raw_data.decode('utf-8')
            else:
                json_data = raw_data
            
            # 解析JSON
            message = json.loads(json_data)
            
            # 更新统计
            self.metrics["messages_received"] += 1
            self.metrics["bytes_received"] += len(json_data.encode('utf-8'))
            
            return message
            
        except Exception as e:
            self.logger.error(f"接收消息失败: {e}")
            raise
    
    def get_connection_info(self, websocket) -> Dict[str, Any]:
        """获取连接信息"""
        return {
            "local_address": websocket.local_address,
            "remote_address": websocket.remote_address,
            "state": websocket.state.name,
            "compression_enabled": self.enable_compression,
            "max_message_size": self.max_message_size
        }
    
    def get_performance_metrics(self) -> Dict[str, Any]:
        """获取性能指标"""
        total_messages = self.metrics["messages_sent"] + self.metrics["messages_received"]
        
        return {
            "total_messages": total_messages,
            "messages_sent": self.metrics["messages_sent"],
            "messages_received": self.metrics["messages_received"],
            "total_bytes": self.metrics["bytes_sent"] + self.metrics["bytes_received"],
            "average_message_size": (
                (self.metrics["bytes_sent"] + self.metrics["bytes_received"]) / total_messages
                if total_messages > 0 else 0
            ),
            "compression_ratio": self.metrics["compression_ratio"],
            "active_connections": self.metrics["connection_count"]
        }

# 使用示例
async def demo_optimized_transport():
    """演示优化传输层"""
    transport = OptimizedMCPTransport(enable_compression=True)
    
    # 模拟大消息发送
    large_message = {
        "jsonrpc": "2.0",
        "id": "large_001",
        "method": "tools/call",
        "params": {
            "name": "data_processor",
            "arguments": {
                "data": ["item_" + str(i) for i in range(1000)],  # 大量数据
                "options": {"format": "json", "compress": True}
            }
        }
    }
    
    print("=== 优化传输层演示 ===")
    print(f"原始消息大小: {len(json.dumps(large_message, ensure_ascii=False))} 字节")
    
    # 模拟发送 (实际需要WebSocket连接)
    # await transport.send_message(websocket, large_message)
    
    # 显示性能指标
    metrics = transport.get_performance_metrics()
    print(f"性能指标:")
    print(f"  - 平均消息大小: {metrics['average_message_size']:.0f} 字节")
    print(f"  - 压缩率: {metrics['compression_ratio']:.2%}")
    print(f"  - 活跃连接数: {metrics['active_connections']}")

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

4. 调试监控与故障排查

4.1 调试级别与监控指标

4.2 完整的监控调试系统

import time
import threading
from collections import deque, defaultdict
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional

class MCPDebugMonitor:
    """MCP调试监控系统"""
    
    def __init__(self, max_history: int = 1000, alert_threshold: float = 0.05):
        self.max_history = max_history
        self.alert_threshold = alert_threshold
        
        # 监控数据存储
        self.request_history = deque(maxlen=max_history)
        self.error_history = deque(maxlen=max_history) 
        self.performance_history = deque(maxlen=max_history)
        
        # 统计计数器
        self.counters = defaultdict(int)
        self.timers = defaultdict(list)
        
        # 告警系统
        self.alerts = []
        self.alert_handlers: List[Callable] = []
        
        # 线程锁
        self.lock = threading.Lock()
    
    def log_request(self, request: Dict[str, Any], start_time: float) -> str:
        """记录请求"""
        request_id = request.get("id", "unknown")
        method = request.get("method", "unknown")
        
        with self.lock:
            log_entry = {
                "id": request_id,
                "method": method,
                "timestamp": start_time,
                "params_size": len(str(request.get("params", {}))),
                "status": "started"
            }
            
            self.request_history.append(log_entry)
            self.counters[f"requests_{method}"] += 1
            self.counters["total_requests"] += 1
        
        return request_id
    
    def log_response(self, request_id: str, response: Dict[str, Any], 
                    end_time: float, start_time: float) -> None:
        """记录响应"""
        execution_time = end_time - start_time
        is_error = "error" in response
        
        with self.lock:
            # 更新请求记录
            for entry in reversed(self.request_history):
                if entry["id"] == request_id:
                    entry.update({
                        "status": "error" if is_error else "completed",
                        "execution_time": execution_time,
                        "response_size": len(str(response)),
                        "end_time": end_time
                    })
                    break
            
            # 记录性能数据
            perf_entry = {
                "timestamp": end_time,
                "request_id": request_id,
                "execution_time": execution_time,
                "is_error": is_error,
                "method": response.get("method", "unknown")
            }
            self.performance_history.append(perf_entry)
            
            # 更新计数器
            if is_error:
                self.counters["total_errors"] += 1
                error_code = response.get("error", {}).get("code", "unknown")
                self.counters[f"error_{error_code}"] += 1
                
                # 记录错误详情
                error_entry = {
                    "timestamp": end_time,
                    "request_id": request_id,
                    "error_code": error_code,
                    "error_message": response.get("error", {}).get("message", ""),
                    "execution_time": execution_time
                }
                self.error_history.append(error_entry)
            
            # 记录执行时间
            self.timers["execution_times"].append(execution_time)
            if len(self.timers["execution_times"]) > 100:  # 保持最近100个
                self.timers["execution_times"].pop(0)
            
            # 检查告警条件
            self._check_alerts()
    
    def _check_alerts(self) -> None:
        """检查告警条件"""
        current_time = time.time()
        
        # 检查错误率
        recent_requests = [
            entry for entry in self.performance_history
            if current_time - entry["timestamp"] < 300  # 最近5分钟
        ]
        
        if len(recent_requests) >= 10:  # 至少有10个请求才检查
            error_count = sum(1 for entry in recent_requests if entry["is_error"])
            error_rate = error_count / len(recent_requests)
            
            if error_rate > self.alert_threshold:
                alert = {
                    "type": "high_error_rate",
                    "message": f"错误率过高: {error_rate:.2%} (阈值: {self.alert_threshold:.2%})",
                    "timestamp": current_time,
                    "details": {
                        "error_count": error_count,
                        "total_requests": len(recent_requests),
                        "error_rate": error_rate
                    }
                }
                self._trigger_alert(alert)
        
        # 检查响应时间
        if self.timers["execution_times"]:
            avg_time = sum(self.timers["execution_times"]) / len(self.timers["execution_times"])
            if avg_time > 2.0:  # 平均响应时间超过2秒
                alert = {
                    "type": "slow_response",
                    "message": f"响应时间过慢: {avg_time:.2f}s",
                    "timestamp": current_time,
                    "details": {"average_time": avg_time}
                }
                self._trigger_alert(alert)
    
    def _trigger_alert(self, alert: Dict[str, Any]) -> None:
        """触发告警"""
        # 避免重复告警
        recent_alerts = [
            a for a in self.alerts 
            if a["type"] == alert["type"] and 
               time.time() - a["timestamp"] < 300  # 5分钟内
        ]
        
        if not recent_alerts:
            self.alerts.append(alert)
            print(f"🚨 告警: {alert['message']}")
            
            # 调用告警处理器
            for handler in self.alert_handlers:
                try:
                    handler(alert)
                except Exception as e:
                    print(f"告警处理器执行失败: {e}")
    
    def get_dashboard_data(self) -> Dict[str, Any]:
        """获取监控面板数据"""
        current_time = time.time()
        
        with self.lock:
            # 基础统计
            total_requests = self.counters["total_requests"]
            total_errors = self.counters["total_errors"]
            error_rate = (total_errors / total_requests * 100) if total_requests > 0 else 0
            
            # 最近1小时的性能数据
            recent_perf = [
                entry for entry in self.performance_history
                if current_time - entry["timestamp"] < 3600
            ]
            
            avg_response_time = (
                sum(entry["execution_time"] for entry in recent_perf) / len(recent_perf)
                if recent_perf else 0
            )
            
            # 热门方法统计
            method_stats = {}
            for entry in recent_perf:
                method = entry.get("method", "unknown")
                if method not in method_stats:
                    method_stats[method] = {"count": 0, "total_time": 0}
                method_stats[method]["count"] += 1
                method_stats[method]["total_time"] += entry["execution_time"]
            
            # 计算各方法平均时间
            for method, stats in method_stats.items():
                stats["avg_time"] = stats["total_time"] / stats["count"]
            
            # 最近错误
            recent_errors = list(self.error_history)[-5:]  # 最近5个错误
            
            return {
                "summary": {
                    "total_requests": total_requests,
                    "total_errors": total_errors,
                    "error_rate": error_rate,
                    "avg_response_time": avg_response_time,
                    "active_alerts": len([a for a in self.alerts if current_time - a["timestamp"] < 3600])
                },
                "method_stats": dict(sorted(
                    method_stats.items(), 
                    key=lambda x: x[1]["count"], 
                    reverse=True
                )),
                "recent_errors": recent_errors,
                "recent_alerts": [a for a in self.alerts if current_time - a["timestamp"] < 3600]
            }
    
    def add_alert_handler(self, handler: Callable[[Dict[str, Any]], None]) -> None:
        """添加告警处理器"""
        self.alert_handlers.append(handler)

# 告警处理示例
def email_alert_handler(alert: Dict[str, Any]) -> None:
    """邮件告警处理器 (示例)"""
    print(f"📧 发送邮件告警: {alert['message']}")

def webhook_alert_handler(alert: Dict[str, Any]) -> None:
    """Webhook告警处理器 (示例)"""
    print(f"🔗 发送Webhook告警: {alert['type']}")

# 使用示例
async def demo_monitoring_system():
    """演示监控系统"""
    monitor = MCPDebugMonitor()
    
    # 添加告警处理器
    monitor.add_alert_handler(email_alert_handler)
    monitor.add_alert_handler(webhook_alert_handler)
    
    print("=== MCP监控系统演示 ===")
    
    # 模拟一些请求和响应
    for i in range(20):
        # 模拟请求
        request = {
            "jsonrpc": "2.0",
            "id": f"req_{i}",
            "method": "tools/call" if i % 3 == 0 else "resources/read",
            "params": {"test": f"data_{i}"}
        }
        
        start_time = time.time()
        request_id = monitor.log_request(request, start_time)
        
        # 模拟处理时间
        await asyncio.sleep(0.1 + (i % 5) * 0.1)  # 0.1-0.6秒
        
        # 模拟响应 (偶尔出错)
        if i % 7 == 0:  # 大约14%的错误率
            response = {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {"code": -32603, "message": "Internal error"}
            }
        else:
            response = {
                "jsonrpc": "2.0", 
                "id": request_id,
                "result": {"status": "success", "data": f"result_{i}"}
            }
        
        end_time = time.time()
        monitor.log_response(request_id, response, end_time, start_time)
    
    # 显示监控面板数据
    dashboard = monitor.get_dashboard_data()
    print(f"\n=== 监控面板 ===")
    print(f"总请求数: {dashboard['summary']['total_requests']}")
    print(f"错误率: {dashboard['summary']['error_rate']:.1f}%")
    print(f"平均响应时间: {dashboard['summary']['avg_response_time']:.3f}s")
    print(f"活跃告警: {dashboard['summary']['active_alerts']}")
    
    print(f"\n方法统计:")
    for method, stats in dashboard['method_stats'].items():
        print(f"  {method}: {stats['count']}次, 平均{stats['avg_time']:.3f}s")
    
    if dashboard['recent_errors']:
        print(f"\n最近错误:")
        for error in dashboard['recent_errors']:
            print(f"  {error['error_code']}: {error['error_message']}")

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

🎨 第三部分：实战场景与性能优化

前面我们学了"理论基础"和"高级技巧"，现在该到了"实战演练"的时刻！就像学开车一样，理论考试过了，现在要上路实践。今天我们要看看在真实的MCP场景中，JSON-RPC是如何大显身手的！## 🎯 第三部分核心要点总结

1. MCP核心场景实现

我们深入学习了三个关键场景：

• 工具调用：最高频的操作，需要重点优化性能和错误处理
• 资源获取：通过智能缓存机制提升响应速度，缓存命中率可达75%+
• 提示模板：轻量级操作，但要确保模板管理的灵活性

2. 网络传输优化

• 协议选择：WebSocket适合实时场景，HTTP/2适合并发请求
• 数据压缩：对超过1KB的消息进行gzip压缩，平均节省30-50%带宽
• 连接管理：使用连接池复用，减少握手开销

3. 监控调试体系

• 性能指标：响应时间、错误率、吞吐量、资源使用率
• 智能告警：自动检测异常模式，避免雷群效应
• 故障排查：完整的请求链路追踪和错误分析

4. 实用性能数据

欢迎大家关注同名公众号《凡人的工具箱》：关注就送学习大礼包