""" MCP适配器基类 所有外部接口适配器都继承自这个基类 """ from abc import ABC, abstractmethod from typing import Dict, Any, Optional, List from dataclasses import dataclass @dataclass class AdapterResult: """适配器执行结果""" success: bool data: Any = None error: Optional[str] = None source: str = "mcp" class BaseAdapter(ABC): """ MCP适配器基类 职责: 1. 定义统一的接口规范 2. 处理缓存逻辑 3. 错误处理和降级 """ name: str = "" # 适配器名称 description: str = "" # 适配器描述 def __init__(self, mcp_client=None, repository=None): self.mcp_client = mcp_client self.repository = repository self._use_cache = repository is not None @abstractmethod async def execute(self, action: str, **kwargs) -> AdapterResult: """ 执行操作(统一入口) Args: action: 操作类型 **kwargs: 操作参数 Returns: AdapterResult: 执行结果 """ pass async def _get_from_cache(self, key: str, **kwargs) -> Optional[Any]: """从缓存获取数据(子类实现)""" return None async def _save_to_cache(self, key: str, data: Any, **kwargs): """保存数据到缓存(子类实现)""" pass def _fallback(self, action: str, **kwargs) -> AdapterResult: """ 降级方案(模拟数据) 当MCP不可用时,返回模拟数据保持系统可用 """ return AdapterResult( success=True, data=self._get_mock_data(action, **kwargs), source="mock" ) def _get_mock_data(self, action: str, **kwargs) -> Any: """获取模拟数据(子类实现)""" return None