awoooi/apps/api/src/plugins/mcp/interfaces.py

"""
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 src.utils.timezone import now_taipei

# =============================================================================
# Data Classes (DTO)
# =============================================================================


@dataclass
class MCPTool:
    """MCP 工具定義"""

    name: str
    description: str
    input_schema: dict[str, Any]
    server_name: str


@dataclass
class MCPToolResult:
    """
    工具執行結果

    符合 ActionResult 介面，可直接用於 HITL 審核流程
    """

    success: bool
    execution_id: str
    output: Any | None = None
    error: str | None = None
    duration: float = 0.0
    timestamp: datetime = field(default_factory=now_taipei)

    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