74 lines
1.9 KiB
Python
74 lines
1.9 KiB
Python
|
"""
|
|||
|
|
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
|