features/mcp/tools/base_tool.py

"""
標準化 MCP 工具基類
定義統一的輸入輸出格式和錯誤處理
"""

import json
import logging
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
from datetime import datetime

try:
    import jsonschema
except ImportError:
    jsonschema = None

logger = logging.getLogger("mcp.tools.base")


class ToolError(Exception):
    """工具錯誤基類"""
    def __init__(self, code: str, message: str, details: Optional[Dict[str, Any]] = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(message)


class ValidationError(ToolError):
    """參數驗證錯誤"""
    def __init__(self, field: str, message: str):
        super().__init__(
            code="VALIDATION_ERROR",
            message=f"參數 '{field}' 驗證失敗: {message}",
            details={"field": field}
        )


class ExecutionError(ToolError):
    """執行錯誤"""
    def __init__(self, message: str, cause: Optional[Exception] = None):
        super().__init__(
            code="EXECUTION_ERROR",
            message=message,
            details={"cause": str(cause) if cause else None}
        )


class MCPTool(ABC):
    """標準化 MCP 工具基類"""

    # 工具基本信息
    NAME: str = ""
    DESCRIPTION: str = ""
    DESCRIPTION_SHORT: str = ""  # 簡短描述（用於 Intent Detection，10-20 tokens）
    CATEGORY: str = "general"
    TAGS: List[str] = []
    KEYWORDS: List[str] = []  # 用於快速意圖匹配的關鍵字
    USAGE_TIPS: List[str] = []
    IS_COMPLEX: bool = False  # 標記是否為複雜工具（需要兩階段參數填充）

    @classmethod
    def get_summary(cls) -> Dict[str, Any]:
        """獲取工具摘要（用於 Intent Detection，減少 token 消耗）"""
        # 自動生成簡短描述（截取前 50 字元或使用自訂）
        short_desc = cls.DESCRIPTION_SHORT or (
            cls.DESCRIPTION[:47] + "..." if len(cls.DESCRIPTION) > 50 else cls.DESCRIPTION
        )
        
        summary = {
            "name": cls.NAME,
            "description": short_desc,
            "category": cls.CATEGORY,
            "keywords": cls.KEYWORDS,
            "is_complex": cls.IS_COMPLEX
        }
        
        # 簡單工具：提供簡化參數列表（只有參數名，不含詳細 schema）
        if not cls.IS_COMPLEX:
            try:
                schema = cls.get_input_schema()
                summary["params"] = list(schema.get("properties", {}).keys())
            except:
                summary["params"] = []
        
        return summary
    
    @classmethod
    def get_full_definition(cls) -> Dict[str, Any]:
        """獲取完整工具定義（用於實際調用，包含完整 schema）"""
        return {
            "name": cls.NAME,
            "description": cls.DESCRIPTION,
            "metadata": {
                "category": cls.CATEGORY,
                "tags": cls.TAGS,
                "keywords": cls.KEYWORDS,
                "is_complex": cls.IS_COMPLEX,
                "usage_tips": cls.USAGE_TIPS
            },
            "inputSchema": cls.get_input_schema(),
            "outputSchema": cls.get_output_schema()
        }

    @classmethod
    def get_definition(cls) -> Dict[str, Any]:
        """獲取工具定義（向後兼容，使用完整定義）"""
        return cls.get_full_definition()

    @classmethod
    @abstractmethod
    def get_input_schema(cls) -> Dict[str, Any]:
        """獲取輸入參數模式"""
        pass

    @classmethod
    @abstractmethod
    def get_output_schema(cls) -> Dict[str, Any]:
        """獲取輸出結果模式"""
        pass

    @classmethod
    def validate_input(cls, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """驗證和標準化輸入參數"""
        try:
            # 修復：確保 arguments 是 dict 類型
            if not isinstance(arguments, dict):
                logger.error(f"工具 {cls.NAME} 收到非 dict 類型的參數: {type(arguments).__name__} = {arguments}")
                raise ValidationError("arguments", f"參數必須是 dict 類型，但收到: {type(arguments).__name__}")

            # 使用 JSON Schema 進行驗證
            import jsonschema

            schema = cls.get_input_schema()
            jsonschema.validate(arguments, schema)

            # 應用默認值
            validated_args = cls._apply_defaults(arguments, schema)

            return validated_args

        except jsonschema.ValidationError as e:
            raise ValidationError(e.absolute_path[0] if e.absolute_path else "unknown", str(e))
        except Exception as e:
            raise ValidationError("unknown", f"輸入驗證失敗: {str(e)}")

    @classmethod
    def _apply_defaults(cls, arguments: Dict[str, Any], schema: Dict[str, Any]) -> Dict[str, Any]:
        """應用默認值"""
        result = arguments.copy()
        properties = schema.get("properties", {})

        for prop_name, prop_schema in properties.items():
            if prop_name not in result and "default" in prop_schema:
                result[prop_name] = prop_schema["default"]

        return result

    @classmethod
    async def execute_safe(cls, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """安全執行工具，包含錯誤處理"""
        try:
            # 驗證輸入
            validated_args = cls.validate_input(arguments)

            # 執行工具
            start_time = datetime.now()
            result = await cls.execute(validated_args)
            end_time = datetime.now()

            # 驗證輸出
            validated_result = cls.validate_output(result)

            # 添加元數據
            validated_result["metadata"] = validated_result.get("metadata", {})
            validated_result["metadata"].update({
                "tool_name": cls.NAME,
                "execution_time": (end_time - start_time).total_seconds(),
                "timestamp": end_time.isoformat()
            })

            return validated_result

        except ToolError:
            raise  # 重新拋出工具錯誤
        except Exception as e:
            import traceback
            logger.error(f"工具 {cls.NAME} 執行失敗: {e}")
            logger.error(f"完整 traceback:\n{traceback.format_exc()}")
            raise ExecutionError(f"工具執行失敗: {str(e)}", e)

    @classmethod
    def validate_output(cls, result: Dict[str, Any]) -> Dict[str, Any]:
        """驗證輸出結果"""
        if jsonschema is None:
            logger.warning(f"jsonschema 未安裝，跳過輸出驗證")
            return result

        try:
            schema = cls.get_output_schema()
            jsonschema.validate(result, schema)

            return result

        except jsonschema.ValidationError as e:
            logger.warning(f"工具 {cls.NAME} 輸出格式不符合規範: {e}")
            # 不拋出錯誤，只記錄警告，允許繼續執行
            return result
        except Exception as e:
            logger.warning(f"工具 {cls.NAME} 輸出驗證失敗: {e}")
            return result

    @classmethod
    @abstractmethod
    async def execute(cls, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """執行工具邏輯"""
        pass

    @classmethod
    def create_success_response(cls, content: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """創建成功的回應"""
        response = {
            "success": True,
            "content": content,
            "metadata": {
                "tool_name": cls.NAME,
                "category": cls.CATEGORY
            }
        }

        if data:
            response.update(data)

        return response

    @classmethod
    def create_error_response(cls, error: str, code: str = "UNKNOWN_ERROR") -> Dict[str, Any]:
        """創建錯誤的回應"""
        return {
            "success": False,
            "error": error,
            "error_code": code,
            "metadata": {
                "tool_name": cls.NAME,
                "category": cls.CATEGORY
            }
        }


class StandardToolSchemas:
    """標準工具模式定義"""

    @staticmethod
    def create_input_schema(properties: Dict[str, Any], required: Optional[List[str]] = None) -> Dict[str, Any]:
        """創建標準輸入模式"""
        schema = {
            "type": "object",
            "properties": properties
        }

        if required:
            schema["required"] = required

        return schema

    @staticmethod
    def create_output_schema() -> Dict[str, Any]:
        """創建標準輸出模式"""
        return {
            "type": "object",
            "properties": {
                "success": {"type": "boolean"},
                "content": {"type": "string"},
                "error": {"type": ["string", "null"]},
                "error_code": {"type": ["string", "null"]},
                "metadata": {
                    "type": "object",
                    "properties": {
                        "tool_name": {"type": "string"},
                        "category": {"type": "string"},
                        "execution_time": {"type": "number"},
                        "timestamp": {"type": "string"}
                    }
                }
            },
            "required": ["success"]
        }