Spaces:
Running
Running
File size: 9,402 Bytes
3f0377e e2e01e0 3f0377e e2e01e0 3f0377e e2e01e0 3f0377e e2e01e0 3f0377e e2e01e0 3f0377e e2e01e0 3f0377e 921a78a 3f0377e 921a78a 3f0377e 921a78a 3f0377e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 |
"""
標準化 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"]
} |