refactor(api): ADR-015 MCP 模組化架構重構

## 重構內容

符合 leWOOOgo 積木化原則:
- 新增 interfaces.py: MCPToolProvider ABC 定義
- 新增 registry.py: Provider 註冊中心 (DI 模式)
- 新增 providers/: K8s, SignOz, Database 具體實作
- 重構 mcp_bridge.py: 透過 ProviderRegistry 委派執行

## 修復 Code Review 問題

- 🔴 移除 _execute_stdio logging 敏感 parameters
- 🔴 修復 conversational-view.tsx i18n 硬編碼

## 新增檔案

- apps/api/src/plugins/mcp/interfaces.py
- apps/api/src/plugins/mcp/registry.py
- apps/api/src/plugins/mcp/providers/__init__.py
- apps/api/src/plugins/mcp/providers/k8s_provider.py
- apps/api/src/plugins/mcp/providers/signoz_provider.py
- apps/api/src/plugins/mcp/providers/database_provider.py
- docs/adr/ADR-015-mcp-modular-architecture.md
- .dependency-cruiser.cjs (Phase 14.2 準備)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

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

@@ -0,0 +1,173 @@
+"""
+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

apps/api/src/plugins/mcp/mcp_bridge.py

@@ -561,9 +561,25 @@ class MCPBridge:
         """
         HTTP 方式執行工具
 
-        Phase 13.2: 整合真實 K8s Executor (不再是 Mock)
+        ADR-015 重構: 透過 ProviderRegistry 委派執行
+        不再直接 import services，符合 leWOOOgo 積木化原則
         """
         # =============================================
+        # ADR-015: 使用 Provider Registry (DI 模式)
+        # =============================================
+        from src.plugins.mcp.registry import get_provider
+
+        provider = get_provider(server.name)
+        if provider:
+            result = await provider.execute(tool_name, parameters)
+            if result.success:
+                return result.output
+            else:
+                return {"error": result.error}
+
+        # =============================================
+        # Fallback: 舊邏輯 (逐步遷移後刪除)
+        # =============================================
         # Kubernetes: 使用真實 ActionExecutor
         # =============================================
         if server.name == "kubernetes":
@@ -859,7 +875,8 @@ class MCPBridge:
     ) -> Any:
         """STDIO 方式執行工具 (Mock 實作)"""
         # Phase 3: Mock 執行
-        logger.info(f"[MOCK] STDIO call to {server.endpoint}: {tool_name}({parameters})")
+        # ⛔ 禁止 logging parameters！(ADR-015 Code Review 修復)
+        logger.info(f"[MOCK] STDIO call to {server.endpoint}: {tool_name}")
 
         mock_responses = {
             "read_file": f"[Mock] Contents of {parameters.get('path')}",

apps/api/src/plugins/mcp/providers/__init__.py

@@ -0,0 +1,34 @@
+"""
+MCP Tool Providers - ADR-015 模組化架構
+======================================
+
+每個 Provider 負責一個領域的 MCP 工具：
+- K8sProvider: Kubernetes 操作 (kubectl)
+- SignOzProvider: 監控指標查詢
+- DatabaseProvider: 資料庫查詢 (Approval/Incident)
+
+@see docs/adr/ADR-015-mcp-modular-architecture.md
+"""
+
+from src.plugins.mcp.providers.database_provider import DatabaseProvider
+from src.plugins.mcp.providers.k8s_provider import K8sProvider
+from src.plugins.mcp.providers.signoz_provider import SignOzProvider
+
+__all__ = [
+    "K8sProvider",
+    "SignOzProvider",
+    "DatabaseProvider",
+]
+
+
+def register_all_providers() -> None:
+    """
+    註冊所有 Provider 到全域 Registry
+
+    應在 FastAPI lifespan 中呼叫
+    """
+    from src.plugins.mcp.registry import register_provider
+
+    register_provider(K8sProvider())
+    register_provider(SignOzProvider())
+    register_provider(DatabaseProvider())

apps/api/src/plugins/mcp/providers/database_provider.py

@@ -0,0 +1,261 @@
+"""
+Database MCP Tool Provider - ADR-015 模組化架構
+==============================================
+
+提供資料庫查詢工具：
+- list_approvals: 列出 Approval 請求
+- get_approval: 取得單一 Approval 詳情
+- list_incidents: 列出活躍事件
+- list_timeline: 列出時間線事件
+
+透過 DI 注入 Services，不直接 import。
+
+@see docs/adr/ADR-015-mcp-modular-architecture.md
+"""
+
+import uuid
+from typing import Any
+from uuid import UUID
+
+import structlog
+
+from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
+
+logger = structlog.get_logger(__name__)
+
+
+class DatabaseProvider(MCPToolProvider):
+    """
+    Database MCP Tool Provider
+
+    封裝所有 Approval/Incident 資料庫查詢操作。
+    """
+
+    def __init__(self) -> None:
+        self._approval_svc = None
+        self._incident_svc = None
+        self._timeline_svc = None
+
+    @property
+    def name(self) -> str:
+        return "database"
+
+    def _get_approval_service(self):
+        """Lazy load approval service"""
+        if self._approval_svc is None:
+            from src.services.approval_db import get_approval_service
+            self._approval_svc = get_approval_service()
+        return self._approval_svc
+
+    def _get_incident_service(self):
+        """Lazy load incident service"""
+        if self._incident_svc is None:
+            from src.services.incident_service import get_incident_service
+            self._incident_svc = get_incident_service()
+        return self._incident_svc
+
+    def _get_timeline_service(self):
+        """Lazy load timeline service"""
+        if self._timeline_svc is None:
+            from src.services.approval_db import get_timeline_service
+            self._timeline_svc = get_timeline_service()
+        return self._timeline_svc
+
+    async def list_tools(self) -> list[MCPTool]:
+        return [
+            MCPTool(
+                name="list_approvals",
+                description="List approval requests with optional status filter",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "status": {"type": "string", "description": "Filter by status: pending, approved, rejected, expired"},
+                        "limit": {"type": "integer", "description": "Max results (default: 20)"},
+                    },
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="get_approval",
+                description="Get detailed information about a specific approval",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "approval_id": {"type": "string", "description": "Approval UUID"},
+                    },
+                    "required": ["approval_id"],
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="list_incidents",
+                description="List incidents with optional status filter",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "status": {"type": "string", "description": "Filter by status: active, resolved, escalated"},
+                        "limit": {"type": "integer", "description": "Max results (default: 20)"},
+                    },
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="list_timeline",
+                description="List timeline events for audit trail",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "limit": {"type": "integer", "description": "Max results (default: 50)"},
+                    },
+                },
+                server_name=self.name,
+            ),
+        ]
+
+    async def execute(
+        self,
+        tool_name: str,
+        parameters: dict[str, Any],
+    ) -> MCPToolResult:
+        execution_id = str(uuid.uuid4())[:8]
+
+        try:
+            if tool_name == "list_approvals":
+                output = await self._list_approvals(parameters)
+            elif tool_name == "get_approval":
+                output = await self._get_approval(parameters)
+            elif tool_name == "list_incidents":
+                output = await self._list_incidents(parameters)
+            elif tool_name == "list_timeline":
+                output = await self._list_timeline(parameters)
+            else:
+                return MCPToolResult(
+                    success=False,
+                    execution_id=execution_id,
+                    error=f"Unknown tool: {tool_name}",
+                )
+
+            return MCPToolResult(
+                success=True,
+                execution_id=execution_id,
+                output=output,
+            )
+
+        except Exception as e:
+            logger.exception("database_provider_error", tool=tool_name, error=str(e))
+            return MCPToolResult(
+                success=False,
+                execution_id=execution_id,
+                error=str(e),
+            )
+
+    async def _list_approvals(self, parameters: dict) -> dict:
+        from src.models.approval import ApprovalStatus
+
+        approval_svc = self._get_approval_service()
+        status_str = parameters.get("status")
+        limit = parameters.get("limit", 20)
+
+        status_filter = None
+        if status_str:
+            try:
+                status_filter = ApprovalStatus(status_str.lower())
+            except ValueError:
+                return {"error": f"Invalid status: {status_str}. Valid: pending, approved, rejected, expired"}
+
+        approvals = await approval_svc.get_all_approvals(
+            status=status_filter,
+            limit=limit,
+        )
+
+        return {
+            "count": len(approvals),
+            "approvals": [
+                {
+                    "id": str(a.id),
+                    "action": a.action[:80] if a.action else "",
+                    "status": a.status.value if hasattr(a.status, 'value') else str(a.status),
+                    "risk_level": a.risk_level.value if hasattr(a.risk_level, 'value') else str(a.risk_level),
+                    "signatures": f"{a.current_signatures}/{a.required_signatures}",
+                    "created_at": a.created_at.isoformat() if a.created_at else None,
+                }
+                for a in approvals
+            ],
+        }
+
+    async def _get_approval(self, parameters: dict) -> dict:
+        approval_id = parameters.get("approval_id")
+        if not approval_id:
+            return {"error": "Missing 'approval_id' parameter"}
+
+        approval_svc = self._get_approval_service()
+        try:
+            approval = await approval_svc.get_approval_by_id(UUID(approval_id))
+        except ValueError:
+            return {"error": f"Invalid UUID format: {approval_id}"}
+
+        if not approval:
+            return {"error": f"Approval not found: {approval_id}"}
+
+        return {
+            "id": str(approval.id),
+            "action": approval.action,
+            "description": approval.description,
+            "status": approval.status.value if hasattr(approval.status, 'value') else str(approval.status),
+            "risk_level": approval.risk_level.value if hasattr(approval.risk_level, 'value') else str(approval.risk_level),
+            "required_signatures": approval.required_signatures,
+            "current_signatures": approval.current_signatures,
+            "signatures": [
+                {"signer": s.signer_name, "timestamp": s.timestamp.isoformat()}
+                for s in (approval.signatures or [])
+            ],
+            "created_at": approval.created_at.isoformat() if approval.created_at else None,
+            "resolved_at": approval.resolved_at.isoformat() if approval.resolved_at else None,
+        }
+
+    async def _list_incidents(self, parameters: dict) -> dict:
+        incident_svc = self._get_incident_service()
+        status_filter = parameters.get("status")
+        limit = parameters.get("limit", 20)
+
+        incidents = await incident_svc.get_active_incidents()
+
+        # Status filter
+        if status_filter:
+            incidents = [i for i in incidents if i.status.value == status_filter.lower()]
+
+        incidents = incidents[:limit]
+
+        return {
+            "count": len(incidents),
+            "incidents": [
+                {
+                    "id": i.incident_id,
+                    "severity": i.severity.value if hasattr(i.severity, 'value') else str(i.severity),
+                    "status": i.status.value if hasattr(i.status, 'value') else str(i.status),
+                    "affected_services": i.affected_services,
+                    "created_at": i.created_at.isoformat() if i.created_at else None,
+                }
+                for i in incidents
+            ],
+        }
+
+    async def _list_timeline(self, parameters: dict) -> dict:
+        timeline_svc = self._get_timeline_service()
+        limit = parameters.get("limit", 50)
+
+        events = await timeline_svc.get_events(limit=limit)
+
+        return {
+            "count": len(events),
+            "events": events,
+        }
+
+    async def health_check(self) -> bool:
+        """Check if database services are accessible"""
+        try:
+            # Try to get approval service
+            self._get_approval_service()
+            return True
+        except Exception:
+            return False

apps/api/src/plugins/mcp/providers/k8s_provider.py

@@ -0,0 +1,231 @@
+"""
+Kubernetes MCP Tool Provider - ADR-015 模組化架構
+================================================
+
+提供 Kubernetes 操作工具：
+- kubectl_get: 查詢資源
+- kubectl_delete: 刪除 Pod
+- kubectl_scale: 調整副本數
+- kubectl_restart: 重啟 Deployment
+
+透過 DI 注入 ActionExecutor，不直接 import services。
+
+@see docs/adr/ADR-015-mcp-modular-architecture.md
+"""
+
+import uuid
+from typing import Any
+
+import structlog
+
+from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
+
+logger = structlog.get_logger(__name__)
+
+
+class K8sProvider(MCPToolProvider):
+    """
+    Kubernetes MCP Tool Provider
+
+    封裝所有 kubectl 操作，透過 ActionExecutor 執行。
+    """
+
+    def __init__(self) -> None:
+        # Lazy import to avoid circular dependency
+        self._executor = None
+
+    @property
+    def name(self) -> str:
+        return "kubernetes"
+
+    def _get_executor(self):
+        """Lazy load executor to avoid import at module load time"""
+        if self._executor is None:
+            from src.services.executor import get_executor
+            self._executor = get_executor()
+        return self._executor
+
+    async def list_tools(self) -> list[MCPTool]:
+        return [
+            MCPTool(
+                name="kubectl_get",
+                description="Query Kubernetes resources (pods, deployments, services)",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "resource": {"type": "string", "description": "Resource type: pods, deployments, services"},
+                        "namespace": {"type": "string", "description": "Namespace (default: awoooi-prod)"},
+                        "name": {"type": "string", "description": "Resource name (optional)"},
+                    },
+                    "required": ["resource"],
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="kubectl_delete",
+                description="Delete a Pod (with dry-run validation)",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "resource": {"type": "string", "description": "Resource type: pod"},
+                        "name": {"type": "string", "description": "Pod name"},
+                        "namespace": {"type": "string", "description": "Namespace"},
+                    },
+                    "required": ["name"],
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="kubectl_scale",
+                description="Scale a Deployment to N replicas",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "deployment": {"type": "string", "description": "Deployment name"},
+                        "replicas": {"type": "integer", "description": "Target replica count"},
+                        "namespace": {"type": "string", "description": "Namespace"},
+                    },
+                    "required": ["deployment", "replicas"],
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="kubectl_restart",
+                description="Restart a Deployment (rollout restart)",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "deployment": {"type": "string", "description": "Deployment name"},
+                        "namespace": {"type": "string", "description": "Namespace"},
+                    },
+                    "required": ["deployment"],
+                },
+                server_name=self.name,
+            ),
+        ]
+
+    async def execute(
+        self,
+        tool_name: str,
+        parameters: dict[str, Any],
+    ) -> MCPToolResult:
+        execution_id = str(uuid.uuid4())[:8]
+        executor = self._get_executor()
+
+        try:
+            if tool_name == "kubectl_get":
+                output = await self._kubectl_get(executor, parameters)
+            elif tool_name == "kubectl_delete":
+                output = await self._kubectl_delete(executor, parameters)
+            elif tool_name == "kubectl_scale":
+                output = await self._kubectl_scale(executor, parameters)
+            elif tool_name == "kubectl_restart":
+                output = await self._kubectl_restart(executor, parameters)
+            else:
+                return MCPToolResult(
+                    success=False,
+                    execution_id=execution_id,
+                    error=f"Unknown tool: {tool_name}",
+                )
+
+            return MCPToolResult(
+                success=True,
+                execution_id=execution_id,
+                output=output,
+            )
+
+        except Exception as e:
+            logger.exception("k8s_provider_error", tool=tool_name, error=str(e))
+            return MCPToolResult(
+                success=False,
+                execution_id=execution_id,
+                error=str(e),
+            )
+
+    async def _kubectl_get(self, executor, parameters: dict) -> dict:
+        namespace = parameters.get("namespace", "awoooi-prod")
+        resource = parameters.get("resource", "pods")
+        name = parameters.get("name", "")
+
+        cmd = f"kubectl get {resource} {name} -n {namespace} -o json".strip()
+        result = await executor.execute_kubectl_command(cmd)
+
+        if result.success and result.k8s_response:
+            return result.k8s_response.get("stdout", "")
+        return {"error": result.error}
+
+    async def _kubectl_delete(self, executor, parameters: dict) -> dict:
+        namespace = parameters.get("namespace", "awoooi-prod")
+        resource = parameters.get("resource", "pod")
+        name = parameters.get("name", "")
+
+        if not name:
+            return {"error": "Missing 'name' parameter"}
+
+        # Dry-run validation
+        if resource == "pod":
+            dry_run = await executor.validate_pod_exists(name, namespace)
+        else:
+            dry_run = await executor.validate_deployment_exists(name, namespace)
+
+        if not dry_run.passed:
+            return {"error": dry_run.message, "dry_run": False}
+
+        # Execute deletion
+        if resource == "pod":
+            result = await executor.delete_pod(name, namespace)
+        else:
+            return {"error": "Direct deployment deletion not supported, use restart"}
+
+        return {
+            "success": result.success,
+            "message": result.message,
+            "duration_ms": result.duration_ms,
+        }
+
+    async def _kubectl_scale(self, executor, parameters: dict) -> dict:
+        namespace = parameters.get("namespace", "awoooi-prod")
+        deployment = parameters.get("deployment", "")
+        replicas = parameters.get("replicas", 1)
+
+        if not deployment:
+            return {"error": "Missing 'deployment' parameter"}
+
+        cmd = f"kubectl scale deployment/{deployment} --replicas={replicas} -n {namespace}"
+        result = await executor.execute_kubectl_command(cmd)
+
+        return {
+            "success": result.success,
+            "scaled": result.success,
+            "replicas": replicas,
+            "message": result.message,
+        }
+
+    async def _kubectl_restart(self, executor, parameters: dict) -> dict:
+        namespace = parameters.get("namespace", "awoooi-prod")
+        deployment = parameters.get("deployment", "")
+
+        if not deployment:
+            return {"error": "Missing 'deployment' parameter"}
+
+        dry_run = await executor.validate_deployment_exists(deployment, namespace)
+        if not dry_run.passed:
+            return {"error": dry_run.message, "dry_run": False}
+
+        result = await executor.restart_deployment(deployment, namespace)
+
+        return {
+            "success": result.success,
+            "restarted": result.success,
+            "message": result.message,
+            "duration_ms": result.duration_ms,
+        }
+
+    async def health_check(self) -> bool:
+        """Check if kubectl is accessible"""
+        try:
+            executor = self._get_executor()
+            result = await executor.execute_kubectl_command("kubectl version --client")
+            return result.success
+        except Exception:
+            return False

apps/api/src/plugins/mcp/providers/signoz_provider.py

@@ -0,0 +1,194 @@
+"""
+SignOz MCP Tool Provider - ADR-015 模組化架構
+=============================================
+
+提供 SignOz 監控查詢工具：
+- gold_metrics: 取得 Gold Metrics (RPS, Error Rate, P99)
+- trace_url: 生成 Trace 查詢 URL
+- system_metrics: 取得系統指標 (CPU/Disk)
+
+透過 DI 注入 SignOzClient，不直接 import services。
+
+@see docs/adr/ADR-015-mcp-modular-architecture.md
+"""
+
+import uuid
+from typing import Any
+
+import structlog
+
+from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
+
+logger = structlog.get_logger(__name__)
+
+
+class SignOzProvider(MCPToolProvider):
+    """
+    SignOz MCP Tool Provider
+
+    封裝所有監控指標查詢操作。
+    """
+
+    def __init__(self) -> None:
+        self._client = None
+
+    @property
+    def name(self) -> str:
+        return "signoz"
+
+    def _get_client(self):
+        """Lazy load SignOz client"""
+        if self._client is None:
+            from src.services.signoz_client import get_signoz_client
+            self._client = get_signoz_client()
+        return self._client
+
+    async def list_tools(self) -> list[MCPTool]:
+        return [
+            MCPTool(
+                name="gold_metrics",
+                description="Get service Gold Metrics: RPS, Error Rate, P50/P95/P99 Latency",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "service_name": {"type": "string", "description": "Service name (e.g., awoooi-api)"},
+                        "namespace": {"type": "string", "description": "Namespace (default: awoooi-prod)"},
+                        "time_window_minutes": {"type": "integer", "description": "Time window in minutes (default: 10)"},
+                    },
+                    "required": ["service_name"],
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="trace_url",
+                description="Generate SignOz trace URL for a service",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "service_name": {"type": "string", "description": "Service name"},
+                        "window_minutes": {"type": "integer", "description": "Time window (default: 5)"},
+                    },
+                    "required": ["service_name"],
+                },
+                server_name=self.name,
+            ),
+            MCPTool(
+                name="system_metrics",
+                description="Get system metrics (CPU, Disk) for a host",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "host": {"type": "string", "description": "Host IP (default: 192.168.0.188)"},
+                        "time_window_minutes": {"type": "integer", "description": "Time window (default: 5)"},
+                    },
+                },
+                server_name=self.name,
+            ),
+        ]
+
+    async def execute(
+        self,
+        tool_name: str,
+        parameters: dict[str, Any],
+    ) -> MCPToolResult:
+        execution_id = str(uuid.uuid4())[:8]
+        client = self._get_client()
+
+        try:
+            if tool_name == "gold_metrics":
+                output = await self._gold_metrics(client, parameters)
+            elif tool_name == "trace_url":
+                output = self._trace_url(client, parameters)
+            elif tool_name == "system_metrics":
+                output = await self._system_metrics(client, parameters)
+            else:
+                return MCPToolResult(
+                    success=False,
+                    execution_id=execution_id,
+                    error=f"Unknown tool: {tool_name}",
+                )
+
+            return MCPToolResult(
+                success=True,
+                execution_id=execution_id,
+                output=output,
+            )
+
+        except Exception as e:
+            logger.exception("signoz_provider_error", tool=tool_name, error=str(e))
+            return MCPToolResult(
+                success=False,
+                execution_id=execution_id,
+                error=str(e),
+            )
+
+    async def _gold_metrics(self, client, parameters: dict) -> dict:
+        service_name = parameters.get("service_name", "")
+        if not service_name:
+            return {"error": "Missing 'service_name' parameter"}
+
+        namespace = parameters.get("namespace", "awoooi-prod")
+        time_window = parameters.get("time_window_minutes", 10)
+
+        metrics = await client.get_gold_metrics(
+            service_name=service_name,
+            namespace=namespace,
+            time_window_minutes=time_window,
+        )
+
+        return {
+            "service_name": metrics.service_name,
+            "namespace": metrics.namespace,
+            "rps": round(metrics.rps, 2),
+            "rps_trend": metrics.rps_trend,
+            "error_rate": round(metrics.error_rate, 2),
+            "error_count": metrics.error_count,
+            "total_requests": metrics.total_requests,
+            "p50_latency_ms": round(metrics.p50_latency_ms, 1),
+            "p95_latency_ms": round(metrics.p95_latency_ms, 1),
+            "p99_latency_ms": round(metrics.p99_latency_ms, 1),
+            "latency_trend": metrics.latency_trend,
+            "summary": metrics.to_summary(),
+        }
+
+    def _trace_url(self, client, parameters: dict) -> dict:
+        service_name = parameters.get("service_name", "")
+        if not service_name:
+            return {"error": "Missing 'service_name' parameter"}
+
+        window_minutes = parameters.get("window_minutes", 5)
+        url = client.generate_trace_url(
+            service_name=service_name,
+            window_minutes=window_minutes,
+        )
+
+        return {
+            "service_name": service_name,
+            "trace_url": url,
+            "window_minutes": window_minutes,
+        }
+
+    async def _system_metrics(self, client, parameters: dict) -> dict:
+        host = parameters.get("host", "192.168.0.188")
+        time_window = parameters.get("time_window_minutes", 5)
+
+        metrics = await client.get_system_metrics(
+            _host=host,
+            time_window_minutes=time_window,
+        )
+
+        return {
+            "host": host,
+            "cpu": metrics.get("cpu", {}),
+            "disk": metrics.get("disk", {}),
+            "time_range": metrics.get("time_range", {}),
+        }
+
+    async def health_check(self) -> bool:
+        """Check if SignOz is accessible"""
+        try:
+            client = self._get_client()
+            # Try to get metrics for a test service
+            return True  # SignOz client initialization is the check
+        except Exception:
+            return False

apps/api/src/plugins/mcp/registry.py

@@ -0,0 +1,170 @@
+"""
+MCP Provider Registry - ADR-015 模組化架構
+==========================================
+
+Provider 註冊中心，實現依賴注入 (DI) 模式：
+1. 統一管理所有 MCP Tool Providers
+2. 支援動態註冊/反註冊
+3. 支援健康檢查
+
+@see docs/adr/ADR-015-mcp-modular-architecture.md
+"""
+
+import structlog
+
+from src.plugins.mcp.interfaces import MCPToolProvider
+
+logger = structlog.get_logger(__name__)
+
+
+class ProviderRegistry:
+    """
+    MCP Tool Provider 註冊中心
+
+    使用方式:
+        # 註冊
+        registry = ProviderRegistry()
+        registry.register(K8sProvider())
+        registry.register(SignOzProvider())
+
+        # 取得
+        k8s = registry.get("kubernetes")
+        await k8s.execute("kubectl_get", {...})
+
+        # 列出所有
+        for provider in registry.all():
+            print(provider.name)
+    """
+
+    def __init__(self) -> None:
+        self._providers: dict[str, MCPToolProvider] = {}
+
+    def register(self, provider: MCPToolProvider) -> None:
+        """
+        註冊 Provider
+
+        Args:
+            provider: MCPToolProvider 實例
+
+        Raises:
+            ValueError: 如果 Provider 名稱已存在
+        """
+        if provider.name in self._providers:
+            raise ValueError(f"Provider '{provider.name}' already registered")
+
+        self._providers[provider.name] = provider
+        logger.info(
+            "provider_registered",
+            name=provider.name,
+            enabled=provider.enabled,
+        )
+
+    def unregister(self, name: str) -> bool:
+        """
+        反註冊 Provider
+
+        Args:
+            name: Provider 名稱
+
+        Returns:
+            bool: 是否成功反註冊
+        """
+        if name in self._providers:
+            del self._providers[name]
+            logger.info("provider_unregistered", name=name)
+            return True
+        return False
+
+    def get(self, name: str) -> MCPToolProvider | None:
+        """
+        取得 Provider
+
+        Args:
+            name: Provider 名稱
+
+        Returns:
+            MCPToolProvider | None: Provider 實例，若不存在則返回 None
+        """
+        provider = self._providers.get(name)
+        if provider and not provider.enabled:
+            logger.warning("provider_disabled", name=name)
+            return None
+        return provider
+
+    def all(self) -> list[MCPToolProvider]:
+        """
+        取得所有已啟用的 Providers
+
+        Returns:
+            list[MCPToolProvider]: 已啟用的 Provider 列表
+        """
+        return [p for p in self._providers.values() if p.enabled]
+
+    def names(self) -> list[str]:
+        """
+        取得所有已啟用的 Provider 名稱
+
+        Returns:
+            list[str]: Provider 名稱列表
+        """
+        return [p.name for p in self.all()]
+
+    async def health_check_all(self) -> dict[str, bool]:
+        """
+        檢查所有 Provider 健康狀態
+
+        Returns:
+            dict[str, bool]: {provider_name: is_healthy}
+        """
+        results = {}
+        for provider in self.all():
+            try:
+                results[provider.name] = await provider.health_check()
+            except Exception as e:
+                logger.warning(
+                    "provider_health_check_failed",
+                    name=provider.name,
+                    error=str(e),
+                )
+                results[provider.name] = False
+        return results
+
+    def __contains__(self, name: str) -> bool:
+        return name in self._providers
+
+    def __len__(self) -> int:
+        return len(self._providers)
+
+
+# =============================================================================
+# Global Registry Singleton
+# =============================================================================
+
+_registry: ProviderRegistry | None = None
+
+
+def get_provider_registry() -> ProviderRegistry:
+    """
+    取得全域 Provider Registry
+
+    Returns:
+        ProviderRegistry: 單例實例
+    """
+    global _registry
+    if _registry is None:
+        _registry = ProviderRegistry()
+    return _registry
+
+
+def register_provider(provider: MCPToolProvider) -> None:
+    """
+    便捷函數: 註冊 Provider 到全域 Registry
+    """
+    get_provider_registry().register(provider)
+
+
+def get_provider(name: str) -> MCPToolProvider | None:
+    """
+    便捷函數: 從全域 Registry 取得 Provider
+    """
+    return get_provider_registry().get(name)