""" MCP Tool Provider Interfaces - ADR-015 模組化架構 ================================================= 定義 MCP Tool Provider 的抽象介面,確保: 1. Interface 先行 (Contract-First) 2. 模組間透過 Public API 溝通 3. 可測試性 (易於 Mock) @see docs/adr/ADR-015-mcp-modular-architecture.md """ from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import Any from uuid import uuid4 from src.utils.timezone import now_taipei # ============================================================================= # Data Classes (DTO) # ============================================================================= @dataclass class MCPTool: """MCP 工具定義""" name: str description: str input_schema: dict[str, Any] # 2026-05-06 Codex: 部分舊 provider 的 list_tools() 尚未傳 server_name。 # 先給 DTO 預設值,registry 會以 provider.name 補正,避免啟動登記直接 crash。 server_name: str = "" @dataclass class MCPToolResult: """ 工具執行結果 符合 ActionResult 介面,可直接用於 HITL 審核流程 """ success: bool execution_id: str = "" output: Any | None = None # 2026-05-06 Codex: 舊 provider 曾使用 data=... 作為成功輸出欄位。 # 保留 alias,避免 provider 成功路徑因 dataclass 參數不相容而 crash。 data: Any | None = None error: str | None = None duration: float = 0.0 timestamp: datetime = field(default_factory=now_taipei) def __post_init__(self) -> None: if not self.execution_id: self.execution_id = f"mcp-{uuid4()}" if self.output is None and self.data is not None: self.output = self.data def to_dict(self) -> dict: return { "success": self.success, "executionId": self.execution_id, "output": self.output, "error": self.error, "duration": self.duration, "timestamp": self.timestamp.isoformat(), } # ============================================================================= # Abstract Base Classes # ============================================================================= class MCPToolProvider(ABC): """ MCP Tool Provider 抽象介面 所有 MCP Tool 實作必須繼承此類別,確保: - 統一的工具列表格式 - 統一的執行介面 - 統一的結果格式 Usage: class K8sProvider(MCPToolProvider): @property def name(self) -> str: return "kubernetes" async def list_tools(self) -> list[MCPTool]: return [MCPTool(name="kubectl_get", ...)] async def execute(self, tool_name: str, parameters: dict) -> MCPToolResult: if tool_name == "kubectl_get": return await self._kubectl_get(parameters) """ @property @abstractmethod def name(self) -> str: """ Provider 名稱 必須唯一,用於 ProviderRegistry 註冊 例如: 'kubernetes', 'signoz', 'database' """ pass @property def enabled(self) -> bool: """ 是否啟用 可覆寫此方法根據環境變數決定是否啟用 """ return True @abstractmethod async def list_tools(self) -> list[MCPTool]: """ 列出可用工具 Returns: list[MCPTool]: 工具定義列表 """ pass @abstractmethod async def execute( self, tool_name: str, parameters: dict[str, Any], ) -> MCPToolResult: """ 執行工具 Args: tool_name: 工具名稱 (必須在 list_tools() 中定義) parameters: 工具參數 (已經過 Rehydration 還原) Returns: MCPToolResult: 執行結果 Raises: ValueError: 未知的工具名稱 """ pass async def health_check(self) -> bool: """ 健康檢查 可覆寫此方法檢查 Provider 依賴的外部服務是否可用 """ return True class RehydrationProvider(ABC): """ Rehydration Provider 抽象介面 負責將 Privacy Shield 脫敏標籤還原為真實值 """ @abstractmethod def unredact( self, data: Any, mapping: dict[str, str], ) -> Any: """ 還原脫敏資料 Args: data: 可能包含脫敏標籤的資料 mapping: 原始值 → 標籤 的映射表 Returns: 還原後的資料 """ pass