MCP Client 的失败回退机制

from typing import Dict, Any, Optional, List
import re

class ErrorDetector:
    """错误检测器"""
    
    def __init__(self):
        """初始化错误检测器"""
        # 可重试错误类型列表
        self.retryable_errors = [
            "ConnectionError",
            "TimeoutError",
            "HTTPError",
            "ServiceUnavailable",
            "GatewayTimeout",
        ]
        
        # 错误类型映射
        self.error_type_map = {
            r"connection\s+error": "ConnectionError",
            r"timeout": "TimeoutError",
            r"503\s+Service\s+Unavailable": "ServiceUnavailable",
            r"504\s+Gateway\s+Timeout": "GatewayTimeout",
            r"429\s+Too\s+Many\s+Requests": "TooManyRequests",
        }
    
    def detect_error_type(self, error: Exception) -> str:
        """
        检测错误类型
        
        Args:
            error: 异常对象
            
        Returns:
            错误类型字符串
        """
        # 首先尝试使用异常类名作为错误类型
        error_type = type(error).__name__
        
        # 然后检查异常消息，匹配更具体的错误类型
        error_msg = str(error).lower()
        for pattern, mapped_type in self.error_type_map.items():
            if re.search(pattern, error_msg):
                error_type = mapped_type
                break
        
        return error_type
    
    def is_retryable(self, error: Exception) -> bool:
        """
        判断错误是否可重试
        
        Args:
            error: 异常对象
            
        Returns:
            是否可重试
        """
        error_type = self.detect_error_type(error)
        return error_type in self.retryable_errors
    
    def add_retryable_error(self, error_type: str):
        """
        添加可重试错误类型
        
        Args:
            error_type: 错误类型字符串
        """
        if error_type not in self.retryable_errors:
            self.retryable_errors.append(error_type)
    
    def remove_retryable_error(self, error_type: str):
        """
        移除可重试错误类型
        
        Args:
            error_type: 错误类型字符串
        """
        if error_type in self.retryable_errors:
            self.retryable_errors.remove(error_type)
    
    def detect_error_context(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]:
        """
        检测错误上下文信息
        
        Args:
            error: 异常对象
            context: 请求上下文信息
            
        Returns:
            包含错误上下文的字典
        """
        error_type = self.detect_error_type(error)
        retryable = self.is_retryable(error)
        
        return {
            "error_type": error_type,
            "retryable": retryable,
            "error_message": str(error),
            "request_context": context,
            "timestamp": context.get("timestamp", 0),
            "request_id": context.get("request_id", ""),
        }

import asyncio
import time
from typing import Dict, Any, Optional, Callable, List
from dataclasses import dataclass, field

@dataclass
class RetryConfig:
    """重试配置"""
    max_retries: int = 3  # 最大重试次数
    initial_delay: float = 0.1  # 初始重试延迟，秒
    max_delay: float = 60.0  # 最大重试延迟，秒
    backoff_factor: float = 2.0  # 退避因子
    retryable_errors: List[str] = field(default_factory=list)  # 可重试错误类型
    retry_condition: Optional[Callable] = None  # 自定义重试条件函数

class RetryManager:
    """重试管理器"""
    
    def __init__(self, config: Optional[RetryConfig] = None):
        """
        初始化重试管理器
        
        Args:
            config: 重试配置
        """
        self.config = config or RetryConfig()
    
    async def retry(self, coro: Callable, context: Optional[Dict[str, Any]] = None) -> Any:
        """
        执行带重试的异步操作
        
        Args:
            coro: 异步函数
            context: 请求上下文
            
        Returns:
            异步操作结果
        """
        context = context or {}
        attempt = 0
        delay = self.config.initial_delay
        
        while attempt <= self.config.max_retries:
            attempt += 1
            context["attempt"] = attempt
            
            try:
                # 执行异步操作
                result = await coro()
                return result
            except Exception as e:
                # 检测错误类型和上下文
                error_context = {
                    "error": e,
                    "attempt": attempt,
                    "delay": delay,
                    **context
                }
                
                # 检查是否需要重试
                if not self._should_retry(error_context):
                    raise
                
                # 检查是否已达到最大重试次数
                if attempt >= self.config.max_retries:
                    raise
                
                # 等待重试延迟
                await asyncio.sleep(delay)
                
                # 计算下一次重试延迟，使用指数退避
                delay = min(delay * self.config.backoff_factor, self.config.max_delay)
        
        # 理论上不会到达这里，因为上面已经处理了所有情况
        raise Exception("重试次数超过最大值")
    
    def _should_retry(self, error_context: Dict[str, Any]) -> bool:
        """
        判断是否应该重试
        
        Args:
            error_context: 错误上下文
            
        Returns:
            是否应该重试
        """
        error = error_context["error"]
        
        # 1. 检查自定义重试条件
        if self.config.retry_condition:
            return self.config.retry_condition(error_context)
        
        # 2. 检查错误类型是否在可重试列表中
        error_type = type(error).__name__
        if self.config.retryable_errors:
            return error_type in self.config.retryable_errors
        
        # 3. 默认检查：只有连接错误、超时错误等可重试
        retryable_error_types = ["ConnectionError", "TimeoutError", "HTTPError", "ServiceUnavailable"]
        return error_type in retryable_error_types
    
    def update_config(self, config: RetryConfig):
        """
        更新重试配置
        
        Args:
            config: 新的重试配置
        """
        self.config = config
    
    def get_config(self) -> RetryConfig:
        """
        获取当前重试配置
        
        Returns:
            当前重试配置
        """
        return self.config

from typing import Dict, Any, Optional, List, Callable
from dataclasses import dataclass, field

@dataclass
class FallbackStrategy:
    """回退策略"""
    name: str  # 策略名称
    priority: int = 0  # 优先级，值越高优先级越高
    condition: Optional[Callable] = None  # 触发条件函数
    action: Callable  # 回退动作函数
    description: str = ""  # 策略描述

class FallbackStrategyManager:
    """回退策略管理器"""
    
    def __init__(self):
        """初始化回退策略管理器"""
        self.strategies: List[FallbackStrategy] = []
    
    def add_strategy(self, strategy: FallbackStrategy):
        """
        添加回退策略
        
        Args:
            strategy: 回退策略对象
        """
        self.strategies.append(strategy)
        # 按优先级排序，优先级高的在前
        self.strategies.sort(key=lambda x: -x.priority)
    
    def remove_strategy(self, strategy_name: str):
        """
        移除回退策略
        
        Args:
            strategy_name: 策略名称
        """
        self.strategies = [s for s in self.strategies if s.name != strategy_name]
    
    def get_strategy(self, strategy_name: str) -> Optional[FallbackStrategy]:
        """
        获取指定名称的回退策略
        
        Args:
            strategy_name: 策略名称
            
        Returns:
            回退策略对象，未找到则返回 None
        """
        for strategy in self.strategies:
            if strategy.name == strategy_name:
                return strategy
        return None
    
    def select_strategy(self, error_context: Dict[str, Any]) -> Optional[FallbackStrategy]:
        """
        根据错误上下文选择合适的回退策略
        
        Args:
            error_context: 错误上下文
            
        Returns:
            选中的回退策略，未找到则返回 None
        """
        for strategy in self.strategies:
            # 检查策略条件
            if strategy.condition is None or strategy.condition(error_context):
                return strategy
        return None
    
    async def execute_fallback(self, error_context: Dict[str, Any]) -> Any:
        """
        执行回退操作
        
        Args:
            error_context: 错误上下文
            
        Returns:
            回退操作结果
        """
        # 选择回退策略
        strategy = self.select_strategy(error_context)
        
        if not strategy:
            raise Exception("未找到合适的回退策略")
        
        # 执行回退动作
        return await strategy.action(error_context)
    
    def get_all_strategies(self) -> List[FallbackStrategy]:
        """
        获取所有回退策略
        
        Returns:
            所有回退策略列表
        """
        return self.strategies.copy()

import asyncio
import httpx
import time
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field

@dataclass
class ServiceInfo:
    """服务信息"""
    name: str  # 服务名称
    url: str  # 服务 URL
    health_check_url: Optional[str] = None  # 健康检查 URL
    is_healthy: bool = True  # 是否健康
    priority: int = 0  # 优先级，值越高优先级越高
    weight: float = 1.0  # 权重，用于负载均衡
    last_checked: float = 0.0  # 最后检查时间

class ServiceDiscovery:
    """服务发现与健康检查"""
    
    def __init__(self, check_interval: float = 30.0):
        """
        初始化服务发现与健康检查
        
        Args:
            check_interval: 健康检查间隔，秒
        """
        self.services: Dict[str, ServiceInfo] = {}
        self.check_interval = check_interval
        self.is_running = False
        self.health_check_task: Optional[asyncio.Task] = None
        self.client = httpx.AsyncClient()
    
    def register_service(self, service: ServiceInfo):
        """
        注册服务
        
        Args:
            service: 服务信息对象
        """
        self.services[service.name] = service
    
    def deregister_service(self, service_name: str):
        """
        注销服务
        
        Args:
            service_name: 服务名称
        """
        if service_name in self.services:
            del self.services[service_name]
    
    async def _check_service_health(self, service: ServiceInfo) -> bool:
        """
        检查单个服务的健康状态
        
        Args:
            service: 服务信息对象
            
        Returns:
            服务是否健康
        """
        try:
            # 使用服务的健康检查 URL 或默认 URL
            health_url = service.health_check_url or f"{service.url}/health"
            
            response = await self.client.get(health_url, timeout=5.0)
            response.raise_for_status()
            return True
        except Exception as e:
            print(f"服务 {service.name} 健康检查失败: {e}")
            return False
    
    async def _health_check_loop(self):
        """
        健康检查主循环
        """
        while self.is_running:
            # 检查所有服务的健康状态
            for service_name, service in self.services.items():
                is_healthy = await self._check_service_health(service)
                service.is_healthy = is_healthy
                service.last_checked = time.time()
            
            # 等待下一次检查
            await asyncio.sleep(self.check_interval)
    
    async def start(self):
        """
        启动健康检查
        """
        if self.is_running:
            return
        
        self.is_running = True
        
        # 启动健康检查任务
        self.health_check_task = asyncio.create_task(self._health_check_loop())
    
    async def stop(self):
        """
        停止健康检查
        """
        self.is_running = False
        
        if self.health_check_task:
            await self.health_check_task
            self.health_check_task = None
        
        # 关闭 HTTP 客户端
        await self.client.aclose()
    
    def get_healthy_services(self) -> List[ServiceInfo]:
        """
        获取所有健康的服务
        
        Returns:
            健康服务列表，按优先级排序
        """
        healthy_services = [service for service in self.services.values() if service.is_healthy]
        # 按优先级排序，优先级高的在前
        healthy_services.sort(key=lambda x: (-x.priority, x.name))
        return healthy_services
    
    def get_service(self, service_name: str) -> Optional[ServiceInfo]:
        """
        获取指定名称的服务
        
        Args:
            service_name: 服务名称
            
        Returns:
            服务信息对象，未找到则返回 None
        """
        return self.services.get(service_name)
    
    def update_service(self, service: ServiceInfo):
        """
        更新服务信息
        
        Args:
            service: 新的服务信息
        """
        if service.name in self.services:
            self.services[service.name] = service
    
    async def __aenter__(self):
        """进入上下文管理器"""
        await self.start()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """退出上下文管理器"""
        await self.stop()

# 示例：失败回退机制使用示例
import asyncio
import time
from error_detector import ErrorDetector
from retry_manager import RetryManager, RetryConfig
from fallback_strategy_manager import FallbackStrategyManager, FallbackStrategy
from service_discovery import ServiceDiscovery, ServiceInfo

async def example_error_detector():
    """错误检测器示例"""
    print("=== 错误检测器示例 ===")
    
    detector = ErrorDetector()
    
    # 测试不同类型的错误
    test_errors = [
        ConnectionError("Connection refused"),
        TimeoutError("Request timed out"),
        ValueError("Invalid parameter"),
        Exception("Generic error"),
    ]
    
    for error in test_errors:
        error_type = detector.detect_error_type(error)
        is_retryable = detector.is_retryable(error)
        print(f"错误: {error}, 类型: {error_type}, 可重试: {is_retryable}")
    
    # 添加自定义可重试错误类型
    detector.add_retryable_error("ValueError")
    error = ValueError("Invalid parameter")
    print(f"添加 ValueError 到可重试列表后，可重试: {detector.is_retryable(error)}")

async def example_retry_manager():
    """重试管理器示例"""
    print("\n=== 重试管理器示例 ===")
    
    # 创建重试配置
    config = RetryConfig(
        max_retries=3,
        initial_delay=0.1,
        max_delay=1.0,
        backoff_factor=2.0,
    )
    
    retry_manager = RetryManager(config)
    
    # 模拟一个可能失败的异步操作
    attempt_count = 0
    
    async def failing_operation():
        nonlocal attempt_count
        attempt_count += 1
        print(f"执行操作，尝试次数: {attempt_count}")
        
        # 前 2 次尝试失败，第 3 次成功
        if attempt_count < 3:
            raise ConnectionError(f"尝试 {attempt_count} 失败")
        return f"操作成功，尝试次数: {attempt_count}"
    
    start_time = time.time()
    result = await retry_manager.retry(failing_operation)
    end_time = time.time()
    
    print(f"重试结果: {result}")
    print(f"耗时: {end_time - start_time:.2f} 秒")

async def example_fallback_strategy():
    """回退策略示例"""
    print("\n=== 回退策略示例 ===")
    
    fallback_manager = FallbackStrategyManager()
    
    # 定义回退策略
    strategies = [
        FallbackStrategy(
            name="service_retry",
            priority=10,
            condition=lambda ctx: ctx["error_context"]["error_type"] == "ConnectionError",
            action=lambda ctx: f"执行服务重试回退，错误: {ctx['error_context']['error_message']}",
            description="服务连接错误时的重试策略"
        ),
        FallbackStrategy(
            name="model_fallback",
            priority=20,
            condition=lambda ctx: ctx["error_context"]["error_type"] == "TimeoutError",
            action=lambda ctx: f"执行模型回退，错误: {ctx['error_context']['error_message']}",
            description="模型超时错误时的回退策略"
        ),
        FallbackStrategy(
            name="default_fallback",
            priority=5,
            condition=lambda ctx: True,  # 无条件匹配，作为默认策略
            action=lambda ctx: f"执行默认回退，错误: {ctx['error_context']['error_message']}",
            description="默认回退策略"
        ),
    ]
    
    # 注册回退策略
    for strategy in strategies:
        fallback_manager.add_strategy(strategy)
    
    # 测试不同错误类型的回退策略选择
    test_error_contexts = [
        {
            "error_context": {
                "error_type": "ConnectionError",
                "error_message": "Connection refused"
            }
        },
        {
            "error_context": {
                "error_type": "TimeoutError",
                "error_message": "Request timed out"
            }
        },
        {
            "error_context": {
                "error_type": "ValueError",
                "error_message": "Invalid parameter"
            }
        },
    ]
    
    for i, error_ctx in enumerate(test_error_contexts):
        strategy = fallback_manager.select_strategy(error_ctx)
        result = await fallback_manager.execute_fallback(error_ctx)
        print(f"测试 {i+1}: 错误类型 = {error_ctx['error_context']['error_type']}, 选择策略 = {strategy.name}, 回退结果 = {result}")

async def example_service_discovery():
    """服务发现示例"""
    print("\n=== 服务发现示例 ===")
    
    async with ServiceDiscovery(check_interval=10.0) as service_discovery:
        # 注册服务
        services = [
            ServiceInfo(
                name="primary_mcp_server",
                url="http://localhost:8000/mcp",
                health_check_url="http://localhost:8000/health",
                priority=10,
                weight=1.0
            ),
            ServiceInfo(
                name="secondary_mcp_server",
                url="http://localhost:8001/mcp",
                health_check_url="http://localhost:8001/health",
                priority=5,
                weight=0.5
            ),
            ServiceInfo(
                name="cloud_mcp_server",
                url="https://api.example.com/mcp",
                health_check_url="https://api.example.com/health",
                priority=3,
                weight=0.3
            ),
        ]
        
        for service in services:
            service_discovery.register_service(service)
        
        # 等待一段时间，让健康检查运行一次
        await asyncio.sleep(2.0)
        
        # 获取健康服务
        healthy_services = service_discovery.get_healthy_services()
        print(f"健康服务列表: {[service.name for service in healthy_services]}")
        
        # 查看具体服务信息
        for service_name in service_discovery.services:
            service = service_discovery.get_service(service_name)
            print(f"服务 {service.name}: URL = {service.url}, 健康状态 = {service.is_healthy}, 优先级 = {service.priority}")

async def example_complete_fallback():
    """完整回退机制示例"""
    print("\n=== 完整回退机制示例 ===")
    
    # 1. 创建组件
    error_detector = ErrorDetector()
    
    retry_config = RetryConfig(
        max_retries=3,
        initial_delay=0.1,
        max_delay=1.0,
        backoff_factor=2.0,
    )
    retry_manager = RetryManager(retry_config)
    
    fallback_manager = FallbackStrategyManager()
    
    # 2. 注册回退策略
    async def service_retry_action(error_context):
        print(f"执行服务重试回退，错误: {error_context['error_context']['error_message']}")
        # 这里可以实现实际的服务重试逻辑
        return f"服务重试成功"
    
    async def model_fallback_action(error_context):
        print(f"执行模型回退，错误: {error_context['error_context']['error_message']}")
        # 这里可以实现实际的模型回退逻辑
        return f"模型回退成功"
    
    fallback_manager.add_strategy(FallbackStrategy(
        name="service_retry",
        priority=10,
        condition=lambda ctx: ctx["error_context"]["error_type"] == "ConnectionError",
        action=service_retry_action,
        description="服务连接错误时的重试策略"
    ))
    
    fallback_manager.add_strategy(FallbackStrategy(
        name="model_fallback",
        priority=20,
        condition=lambda ctx: ctx["error_context"]["error_type"] == "TimeoutError",
        action=model_fallback_action,
        description="模型超时错误时的回退策略"
    ))
    
    # 3. 模拟一个可能失败的操作
    async def main_operation():
        print("执行主操作")
        # 模拟不同类型的错误
        import random
        error_type = random.choice(["ConnectionError", "TimeoutError", "Success"])
        
        if error_type == "ConnectionError":
            raise ConnectionError("Connection refused")
        elif error_type == "TimeoutError":
            raise TimeoutError("Request timed out")
        else:
            return "主操作成功"
    
    # 4. 执行带回退的操作
    try:
        result = await retry_manager.retry(main_operation)
        print(f"操作结果: {result}")
    except Exception as e:
        # 检测错误
        error_context = error_detector.detect_error_context(e, {"request_id": "test-123"})
        
        # 执行回退
        try:
            fallback_result = await fallback_manager.execute_fallback({"error_context": error_context})
            print(f"回退结果: {fallback_result}")
        except Exception as fallback_e:
            print(f"回退失败: {fallback_e}")

async def main():
    """主函数"""
    # 依次运行所有示例
    await example_error_detector()
    await example_retry_manager()
    await example_fallback_strategy()
    await example_service_discovery()
    await example_complete_fallback()
    
    print("\n所有示例执行完成！")

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