feat: 统一错误处理 - ToolResult 类

- 新增 xcli_core/tool_result.py: ToolResult + ErrorCode
- unified_tool_manager: execute/mcp_call 返回 ToolResult
- cmd_executor: 所有返回改为 ToolResult
- file_manager: 所有返回改为 ToolResult
- 兼容旧接口: execute_dict() 返回 dict, handler 可返回 str 或 ToolResult
This commit is contained in:
shuiyu1123 committed 2026-06-19 00:26:49 +08:00
1 parent 161ab9d3a8
commit bac5ea01a9
4 files changed
+235 -83

No files matched your search

+18 -12
View File
@@ -5,6 +5,11 @@ import os
import re import re
import subprocess import subprocess
import platform import platform
import sys
# 添加项目根目录到 path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from xcli_core.tool_result import ToolResult, ErrorCode
class Liugin: class Liugin:
@@ -68,18 +73,18 @@ class Liugin:
':(){ :|:& };:', 'chmod -R 777 /', 'mv / ', 'wget http', ':(){ :|:& };:', 'chmod -R 777 /', 'mv / ', 'wget http',
} }
def handle(self, args: str) -> str: def handle(self, args: str):
try: try:
parts = args.strip().split(maxsplit=1) parts = args.strip().split(maxsplit=1)
if not parts: if not parts:
return "错误:请提供操作。可用: run <命令>" return ToolResult.missing("操作。可用: run <命令>")
operation = parts[0].lower() operation = parts[0].lower()
if operation != "run": if operation != "run":
return f"错误:不支持的操作 '{operation}'。可用: run" return ToolResult.fail(f"不支持的操作 '{operation}'。可用: run", ErrorCode.UNSUPPORTED_OP)
if len(parts) < 2: if len(parts) < 2:
return "错误:请提供要执行的命令" return ToolResult.missing("要执行的命令")
rest = parts[1] rest = parts[1]
@@ -91,21 +96,20 @@ class Liugin:
rest = re.sub(r'--timeout\s+\d+', '', rest).strip() rest = re.sub(r'--timeout\s+\d+', '', rest).strip()
if not rest: if not rest:
return "错误:请提供要执行的命令" return ToolResult.missing("要执行的命令")
# 安全检查 # 安全检查
for blocked in self.BLOCKED: for blocked in self.BLOCKED:
if blocked in rest: if blocked in rest:
return f" 安全拦截: 检测到危险命令模式 '{blocked}'" return ToolResult.fail(f"安全拦截: 检测到危险命令模式 '{blocked}'", ErrorCode.PERMISSION_DENIED)
return self._execute(rest, timeout) return self._execute(rest, timeout)
except Exception as e: except Exception as e:
return f"命令执行错误: {str(e)}" return ToolResult.fail(str(e), ErrorCode.INTERNAL_ERROR, tool_name="cmd_executor")
def _execute(self, command: str, timeout: int) -> str: def _execute(self, command: str, timeout: int):
try: try:
# Windows 下用 shell=True,Unix 下直接执行
use_shell = platform.system() == 'Windows' use_shell = platform.system() == 'Windows'
result = subprocess.run( result = subprocess.run(
@@ -136,9 +140,11 @@ class Liugin:
if len(output) > 5000: if len(output) > 5000:
output = output[:5000] + f"\n... (截断,超时: {timeout}s)" output = output[:5000] + f"\n... (截断,超时: {timeout}s)"
return output.strip() if rc != 0:
return ToolResult.fail(output.strip(), ErrorCode.EXEC_FAILED, tool_name="cmd_executor")
return ToolResult.ok(output.strip(), tool_name="cmd_executor")
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
return f" 命令超时 ({timeout}s): {command}" return ToolResult.timeout(timeout, tool_name="cmd_executor")
except Exception as e: except Exception as e:
return f"执行失败: {str(e)}" return ToolResult.fail(str(e), ErrorCode.EXEC_FAILED, tool_name="cmd_executor")
+55 -50
View File
@@ -5,10 +5,15 @@
import os import os
import json import json
import shutil import shutil
import sys
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import List from typing import List
# 添加项目根目录到 path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from xcli_core.tool_result import ToolResult, ErrorCode
class Liugin: class Liugin:
"""文件管理器 - 文件和目录的增删改查""" """文件管理器 - 文件和目录的增删改查"""
@@ -70,11 +75,11 @@ class Liugin:
path = arguments.get("path", "") path = arguments.get("path", "")
return f"{op} {path}".strip() return f"{op} {path}".strip()
def handle(self, args: str) -> str: def handle(self, args: str):
try: try:
parts = self._parse_args(args) parts = self._parse_args(args)
if not parts: if not parts:
return "错误:请提供操作类型" return ToolResult.missing("操作类型")
operation = parts[0].lower() operation = parts[0].lower()
handlers = { handlers = {
@@ -92,7 +97,7 @@ class Liugin:
handler = handlers.get(operation) handler = handlers.get(operation)
if not handler: if not handler:
return f"错误:不支持的操作 '{operation}'" return ToolResult.fail(f"不支持的操作 '{operation}'", ErrorCode.UNSUPPORTED_OP)
return handler(parts[1:]) return handler(parts[1:])
except Exception as e: except Exception as e:
@@ -115,7 +120,7 @@ class Liugin:
parts.append(current) parts.append(current)
return parts return parts
def _op_list(self, args: List[str]) -> str: def _op_list(self, args: List[str]):
path, pattern = ".", None path, pattern = ".", None
i = 0 i = 0
while i < len(args): while i < len(args):
@@ -127,7 +132,7 @@ class Liugin:
path = os.path.abspath(path) path = os.path.abspath(path)
if not os.path.isdir(path): if not os.path.isdir(path):
return f"错误:不是目录: {path}" return ToolResult.not_found(f"目录: {path}")
items = [] items = []
try: try:
@@ -153,13 +158,13 @@ class Liugin:
result.append(f" ... 还有 {len(items) - 100} 项") result.append(f" ... 还有 {len(items) - 100} 项")
result.append("─" * 50) result.append("─" * 50)
result.append(f"共 {dirs} 个文件夹, {files} 个文件") result.append(f"共 {dirs} 个文件夹, {files} 个文件")
return '\n'.join(result) return ToolResult.ok('\n'.join(result), tool_name="file_manager")
except Exception as e: except Exception as e:
return f"列出目录失败: {e}" return ToolResult.fail(str(e), ErrorCode.EXEC_FAILED, tool_name="file_manager")
def _op_read(self, args: List[str]) -> str: def _op_read(self, args: List[str]):
if not args: if not args:
return "错误:请提供文件路径" return ToolResult.missing("文件路径")
file_path = os.path.abspath(args[0]) file_path = os.path.abspath(args[0])
max_lines = None max_lines = None
i = 1 i = 1
@@ -168,16 +173,16 @@ class Liugin:
try: try:
max_lines = int(args[i + 1]) max_lines = int(args[i + 1])
except ValueError: except ValueError:
return "错误:行数必须是整数" return ToolResult.fail("行数必须是整数", ErrorCode.INVALID_ARGS)
i += 1 i += 1
i += 1 i += 1
if not os.path.isfile(file_path): if not os.path.isfile(file_path):
return f"错误:文件不存在: {file_path}" return ToolResult.not_found(f"文件: {file_path}")
size = os.path.getsize(file_path) size = os.path.getsize(file_path)
if size > 10 * 1024 * 1024: if size > 10 * 1024 * 1024:
return f"文件过大 ({size/1024/1024:.1f}MB),请用 -n 限制行数" return ToolResult.fail(f"文件过大 ({size/1024/1024:.1f}MB),请用 -n 限制行数", ErrorCode.INVALID_ARGS)
try: try:
with open(file_path, 'r', encoding='utf-8') as f: with open(file_path, 'r', encoding='utf-8') as f:
@@ -195,13 +200,13 @@ class Liugin:
content = '\n'.join(f"{i+1:4d} | {l.rstrip()}" for i, l in enumerate(all_lines)) content = '\n'.join(f"{i+1:4d} | {l.rstrip()}" for i, l in enumerate(all_lines))
header = f" {file_path} ({size}B, {len(all_lines)} 行)" header = f" {file_path} ({size}B, {len(all_lines)} 行)"
return f"{header}\n{'─' * 50}\n{content}" return ToolResult.ok(f"{header}\n{'─' * 50}\n{content}", tool_name="file_manager")
except UnicodeDecodeError: except UnicodeDecodeError:
return "错误:无法解码(可能是二进制文件)" return ToolResult.fail("无法解码(可能是二进制文件)", ErrorCode.DECODE_ERROR)
def _op_write(self, args: List[str]) -> str: def _op_write(self, args: List[str]):
if len(args) < 2: if len(args) < 2:
return "错误:格式: write <文件> -c <内容>" return ToolResult.fail("格式: write <文件> -c <内容>", ErrorCode.MISSING_ARGS)
fp = os.path.abspath(args[0]) fp = os.path.abspath(args[0])
content = None content = None
i = 1 i = 1
@@ -210,15 +215,15 @@ class Liugin:
content = args[i + 1]; i += 1 content = args[i + 1]; i += 1
i += 1 i += 1
if content is None: if content is None:
return "错误:请用 -c 提供内容" return ToolResult.missing("内容 (-c 参数)")
os.makedirs(os.path.dirname(fp) or ".", exist_ok=True) os.makedirs(os.path.dirname(fp) or ".", exist_ok=True)
with open(fp, 'w', encoding='utf-8') as f: with open(fp, 'w', encoding='utf-8') as f:
f.write(content) f.write(content)
return f" 已写入: {fp} ({len(content)} 字符)" return ToolResult.ok(f"已写入: {fp} ({len(content)} 字符)", tool_name="file_manager")
def _op_append(self, args: List[str]) -> str: def _op_append(self, args: List[str]):
if len(args) < 2: if len(args) < 2:
return "错误:格式: append <文件> -c <内容>" return ToolResult.fail("格式: append <文件> -c <内容>", ErrorCode.MISSING_ARGS)
fp = os.path.abspath(args[0]) fp = os.path.abspath(args[0])
content = None content = None
i = 1 i = 1
@@ -227,59 +232,59 @@ class Liugin:
content = args[i + 1]; i += 1 content = args[i + 1]; i += 1
i += 1 i += 1
if content is None: if content is None:
return "错误:请用 -c 提供内容" return ToolResult.missing("内容 (-c 参数)")
if not os.path.isfile(fp): if not os.path.isfile(fp):
return f"错误:文件不存在: {fp}" return ToolResult.not_found(f"文件: {fp}")
with open(fp, 'a', encoding='utf-8') as f: with open(fp, 'a', encoding='utf-8') as f:
f.write(content) f.write(content)
return f" 已追加: {fp}" return ToolResult.ok(f"已追加: {fp}", tool_name="file_manager")
def _op_copy(self, args: List[str]) -> str: def _op_copy(self, args: List[str]):
if len(args) < 2: if len(args) < 2:
return "错误:请提供源和目标路径" return ToolResult.missing("源和目标路径")
src, dst = os.path.abspath(args[0]), os.path.abspath(args[1]) src, dst = os.path.abspath(args[0]), os.path.abspath(args[1])
if not os.path.exists(src): if not os.path.exists(src):
return f"错误:源不存在: {src}" return ToolResult.not_found(f"源: {src}")
os.makedirs(os.path.dirname(dst) or ".", exist_ok=True) os.makedirs(os.path.dirname(dst) or ".", exist_ok=True)
if os.path.isfile(src): if os.path.isfile(src):
shutil.copy2(src, dst) shutil.copy2(src, dst)
else: else:
shutil.copytree(src, dst, dirs_exist_ok=True) shutil.copytree(src, dst, dirs_exist_ok=True)
return f" 已复制:\n {src}\n → {dst}" return ToolResult.ok(f"已复制:\n {src}\n → {dst}", tool_name="file_manager")
def _op_move(self, args: List[str]) -> str: def _op_move(self, args: List[str]):
if len(args) < 2: if len(args) < 2:
return "错误:请提供源和目标路径" return ToolResult.missing("源和目标路径")
src, dst = os.path.abspath(args[0]), os.path.abspath(args[1]) src, dst = os.path.abspath(args[0]), os.path.abspath(args[1])
if not os.path.exists(src): if not os.path.exists(src):
return f"错误:源不存在: {src}" return ToolResult.not_found(f"源: {src}")
os.makedirs(os.path.dirname(dst) or ".", exist_ok=True) os.makedirs(os.path.dirname(dst) or ".", exist_ok=True)
shutil.move(src, dst) shutil.move(src, dst)
return f" 已移动:\n {src}\n → {dst}" return ToolResult.ok(f"已移动:\n {src}\n → {dst}", tool_name="file_manager")
def _op_delete(self, args: List[str]) -> str: def _op_delete(self, args: List[str]):
if not args: if not args:
return "错误:请提供路径" return ToolResult.missing("路径")
path = os.path.abspath(args[0]) path = os.path.abspath(args[0])
recursive = "-r" in args recursive = "-r" in args
if not os.path.exists(path): if not os.path.exists(path):
return f"错误:不存在: {path}" return ToolResult.not_found(f"路径: {path}")
if os.path.isfile(path): if os.path.isfile(path):
os.remove(path) os.remove(path)
return f" 已删除文件: {path}" return ToolResult.ok(f"已删除文件: {path}", tool_name="file_manager")
elif os.path.isdir(path): elif os.path.isdir(path):
if recursive: if recursive:
shutil.rmtree(path) shutil.rmtree(path)
return f" 已删除目录: {path}" return ToolResult.ok(f"已删除目录: {path}", tool_name="file_manager")
elif not os.listdir(path): elif not os.listdir(path):
os.rmdir(path) os.rmdir(path)
return f" 已删除空目录: {path}" return ToolResult.ok(f"已删除空目录: {path}", tool_name="file_manager")
else: else:
return "错误:目录不为空,用 -r 递归删除" return ToolResult.fail("目录不为空,用 -r 递归删除", ErrorCode.INVALID_ARGS)
def _op_search(self, args: List[str]) -> str: def _op_search(self, args: List[str]):
if not args: if not args:
return "错误:请提供搜索目录" return ToolResult.missing("搜索目录")
directory = os.path.abspath(args[0]) directory = os.path.abspath(args[0])
pattern, content_search = None, None pattern, content_search = None, None
i = 1 i = 1
@@ -291,7 +296,7 @@ class Liugin:
i += 1 i += 1
if not os.path.isdir(directory): if not os.path.isdir(directory):
return f"错误:目录不存在: {directory}" return ToolResult.not_found(f"目录: {directory}")
results = [] results = []
for root, dirs, files in os.walk(directory): for root, dirs, files in os.walk(directory):
@@ -317,14 +322,14 @@ class Liugin:
if len(results) >= 50: if len(results) >= 50:
break break
return f"搜索结果 ({len(results)}):\n" + '\n'.join(results) if results else "未找到匹配" return ToolResult.ok(f"搜索结果 ({len(results)}):\n" + '\n'.join(results), tool_name="file_manager") if results else ToolResult.ok("未找到匹配", tool_name="file_manager")
def _op_info(self, args: List[str]) -> str: def _op_info(self, args: List[str]):
if not args: if not args:
return "错误:请提供路径" return ToolResult.missing("路径")
path = os.path.abspath(args[0]) path = os.path.abspath(args[0])
if not os.path.exists(path): if not os.path.exists(path):
return f"错误:不存在: {path}" return ToolResult.not_found(f"路径: {path}")
stat = os.stat(path) stat = os.stat(path)
result = [ result = [
@@ -337,13 +342,13 @@ class Liugin:
if os.path.isfile(path): if os.path.isfile(path):
ext = os.path.splitext(path)[1] ext = os.path.splitext(path)[1]
result.append(f"扩展名: {ext or '无'}") result.append(f"扩展名: {ext or '无'}")
return '\n'.join(result) return ToolResult.ok('\n'.join(result), tool_name="file_manager")
def _op_mkdir(self, args: List[str]) -> str: def _op_mkdir(self, args: List[str]):
if not args: if not args:
return "错误:请提供目录路径" return ToolResult.missing("目录路径")
path = os.path.abspath(args[0]) path = os.path.abspath(args[0])
if os.path.exists(path): if os.path.exists(path):
return f"错误:已存在: {path}" return ToolResult.fail(f"已存在: {path}", ErrorCode.INVALID_ARGS)
os.makedirs(path, exist_ok=True) os.makedirs(path, exist_ok=True)
return f" 已创建: {path}" return ToolResult.ok(f"已创建: {path}", tool_name="file_manager")
+43 -21
View File
@@ -9,6 +9,7 @@ import logging
import threading import threading
from typing import Dict, List, Any, Optional, Union, Callable from typing import Dict, List, Any, Optional, Union, Callable
from colorama import Fore, Style from colorama import Fore, Style
from xcli_core.tool_result import ToolResult, ErrorCode
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -192,7 +193,8 @@ class UnifiedToolManager:
"""为 Skill 创建兼容 Plugin 的 handler""" """为 Skill 创建兼容 Plugin 的 handler"""
from skills.base import MarkdownSkill, Skill from skills.base import MarkdownSkill, Skill
def handler(args: str) -> str: def handler(args: str):
"""Skill handler - 返回 ToolResult"""
kwargs = {} kwargs = {}
args = args.strip() args = args.strip()
@@ -212,16 +214,16 @@ class UnifiedToolManager:
kwargs = {'operation': args} kwargs = {'operation': args}
if isinstance(skill, MarkdownSkill): if isinstance(skill, MarkdownSkill):
return f"[Skill: {skill.name}]\n{skill.instructions}" return ToolResult.ok(f"[Skill: {skill.name}]\n{skill.instructions}")
if hasattr(skill, 'run'): if hasattr(skill, 'run'):
result = skill.run(**kwargs) result = skill.run(**kwargs)
if result.success: if result.success:
return result.message or str(result.data) return ToolResult.ok(result.message or str(result.data))
else: else:
return f"错误: {result.error}" return ToolResult.fail(str(result.error), ErrorCode.EXEC_FAILED)
return f"未知的技能类型: {type(skill)}" return ToolResult.fail(f"未知的技能类型: {type(skill)}", ErrorCode.NOT_IMPLEMENTED)
return handler return handler
@@ -312,12 +314,13 @@ class UnifiedToolManager:
"""获取工具使用的协议""" """获取工具使用的协议"""
return self._protocol_map.get(name) return self._protocol_map.get(name)
def execute(self, tool_name: str, args: str = "", **kwargs) -> Dict[str, Any]: def execute(self, tool_name: str, args: str = "", **kwargs) -> ToolResult:
"""执行工具调用""" """执行工具调用,返回 ToolResult"""
tool = self._tools.get(tool_name) with self._lock:
tool = self._tools.get(tool_name)
if not tool: if not tool:
return {"result": f"未找到工具: {tool_name}"} return ToolResult.fail(f"未找到工具: {tool_name}", ErrorCode.NOT_FOUND, tool_name=tool_name)
try: try:
handler = tool['handler'] handler = tool['handler']
@@ -327,7 +330,10 @@ class UnifiedToolManager:
if protocol == 'skill' and kwargs: if protocol == 'skill' and kwargs:
skill = tool['instance'] skill = tool['instance']
result = skill.run(**kwargs) result = skill.run(**kwargs)
return {"result": result.message or str(result.data) if result.success else f"错误: {result.error}"} if result.success:
return ToolResult.ok(result.message or str(result.data), tool_name=tool_name)
else:
return ToolResult.fail(str(result.error), ErrorCode.EXEC_FAILED, tool_name=tool_name)
elif protocol == 'liugin' and kwargs and not args: elif protocol == 'liugin' and kwargs and not args:
if instance and hasattr(instance, 'convert_mcp_args'): if instance and hasattr(instance, 'convert_mcp_args'):
@@ -336,10 +342,19 @@ class UnifiedToolManager:
args = self._convert_mcp_args_to_plugin_args(tool_name, kwargs, instance) args = self._convert_mcp_args_to_plugin_args(tool_name, kwargs, instance)
result = handler(args) result = handler(args)
return {"result": result if result is not None else "无结果"} # 兼容:handler 可能返回 ToolResult 或字符串
if isinstance(result, ToolResult):
result.tool_name = tool_name
return result
return ToolResult.ok(result if result is not None else "执行完成", tool_name=tool_name)
except Exception as e: except Exception as e:
return {"result": f"工具执行错误: {e}"} logger.exception(f"工具 {tool_name} 执行异常")
return ToolResult.fail(str(e), ErrorCode.INTERNAL_ERROR, tool_name=tool_name)
def execute_dict(self, tool_name: str, args: str = "", **kwargs) -> Dict[str, Any]:
"""执行工具调用,返回 dict(兼容旧接口)"""
return self.execute(tool_name, args, **kwargs).to_dict()
# ==================== MCP 协议支持 ==================== # ==================== MCP 协议支持 ====================
@@ -355,7 +370,7 @@ class UnifiedToolManager:
tool = self._tools.get(tool_name) tool = self._tools.get(tool_name)
if not tool: if not tool:
return self._mcp_error(f"未找到工具: {tool_name}") return ToolResult.fail(f"未找到工具: {tool_name}", ErrorCode.NOT_FOUND, tool_name=tool_name).to_mcp()
try: try:
protocol = tool['protocol'] protocol = tool['protocol']
@@ -364,25 +379,30 @@ class UnifiedToolManager:
if protocol == 'liugin': if protocol == 'liugin':
args = self._convert_mcp_args_to_plugin_args(tool_name, arguments, instance) args = self._convert_mcp_args_to_plugin_args(tool_name, arguments, instance)
result = tool['handler'](args) result = tool['handler'](args)
return self._mcp_success(result if result else "执行完成") # 兼容:handler 可能返回 ToolResult 或字符串
if isinstance(result, ToolResult):
result.tool_name = tool_name
return result.to_mcp()
return ToolResult.ok(result or "执行完成", tool_name=tool_name).to_mcp()
elif protocol == 'skill': elif protocol == 'skill':
from skills.base import MarkdownSkill from skills.base import MarkdownSkill
if isinstance(instance, MarkdownSkill): if isinstance(instance, MarkdownSkill):
return self._mcp_success(f"[Skill: {instance.name}]\n{instance.instructions}") return ToolResult.ok(f"[Skill: {instance.name}]\n{instance.instructions}", tool_name=tool_name).to_mcp()
if hasattr(instance, 'run'): if hasattr(instance, 'run'):
result = instance.run(**arguments) result = instance.run(**arguments)
if result.success: if result.success:
return self._mcp_success(result.message or str(result.data)) return ToolResult.ok(result.message or str(result.data), tool_name=tool_name).to_mcp()
else: else:
return self._mcp_error(f"执行失败: {result.error}") return ToolResult.fail(str(result.error), ErrorCode.EXEC_FAILED, tool_name=tool_name).to_mcp()
return self._mcp_error(f"未知协议类型: {protocol}") return ToolResult.fail(f"未知协议类型: {protocol}", ErrorCode.INTERNAL_ERROR, tool_name=tool_name).to_mcp()
except Exception as e: except Exception as e:
return self._mcp_error(f"工具执行错误: {str(e)}") logger.exception(f"MCP 调用 {tool_name} 异常")
return ToolResult.fail(str(e), ErrorCode.INTERNAL_ERROR, tool_name=tool_name).to_mcp()
def _convert_mcp_args_to_plugin_args(self, tool_name: str, arguments: Dict[str, Any], def _convert_mcp_args_to_plugin_args(self, tool_name: str, arguments: Dict[str, Any],
instance: Any) -> str: instance: Any) -> str:
@@ -423,10 +443,12 @@ class UnifiedToolManager:
return s return s
def _mcp_success(self, text: str) -> Dict[str, Any]: def _mcp_success(self, text: str) -> Dict[str, Any]:
return {"content": [{"type": "text", "text": text}], "isError": False} """MCP 成功响应(兼容旧接口,推荐用 ToolResult.ok().to_mcp())"""
return ToolResult.ok(text).to_mcp()
def _mcp_error(self, error_msg: str) -> Dict[str, Any]: def _mcp_error(self, error_msg: str) -> Dict[str, Any]:
return {"content": [{"type": "text", "text": error_msg}], "isError": True} """MCP 错误响应(兼容旧接口,推荐用 ToolResult.fail().to_mcp())"""
return ToolResult.fail(error_msg).to_mcp()
# ==================== 兼容性 API ==================== # ==================== 兼容性 API ====================
+119
View File
@@ -0,0 +1,119 @@
"""
统一工具返回格式 - ToolResult
所有插件和工具管理器使用统一的返回格式,
替代原来混乱的字符串/Dict 返回。
"""
from dataclasses import dataclass, field
from typing import Any, Optional
# 错误码常量
class ErrorCode:
"""工具错误码"""
SUCCESS = 0
# 参数错误 (1xx)
MISSING_ARGS = 100
INVALID_ARGS = 101
UNSUPPORTED_OP = 102
# 资源错误 (2xx)
NOT_FOUND = 200
FILE_NOT_FOUND = 201
PERMISSION_DENIED = 202
# 执行错误 (3xx)
EXEC_FAILED = 300
TIMEOUT = 301
DECODE_ERROR = 302
# 系统错误 (4xx)
INTERNAL_ERROR = 400
NOT_IMPLEMENTED = 401
_NAMES = {
0: "成功",
100: "缺少参数",
101: "参数无效",
102: "不支持的操作",
200: "未找到",
201: "文件不存在",
202: "权限不足",
300: "执行失败",
301: "超时",
302: "解码错误",
400: "内部错误",
401: "未实现",
}
@classmethod
def name(cls, code: int) -> str:
return cls._NAMES.get(code, f"未知({code})")
@dataclass
class ToolResult:
"""统一工具返回格式"""
success: bool
data: Any = None
error_code: int = ErrorCode.SUCCESS
error_message: str = ""
# 可选元数据
tool_name: str = ""
execution_time: float = 0.0
metadata: dict = field(default_factory=dict)
def __str__(self) -> str:
if self.success:
return str(self.data) if self.data is not None else "执行完成"
return f"[{ErrorCode.name(self.error_code)}] {self.error_message}"
# ── 便捷构造方法 ──
@classmethod
def ok(cls, data: Any = None, **kwargs) -> "ToolResult":
"""成功"""
return cls(success=True, data=data, **kwargs)
@classmethod
def fail(cls, message: str, code: int = ErrorCode.EXEC_FAILED, **kwargs) -> "ToolResult":
"""失败"""
return cls(success=False, error_code=code, error_message=message, **kwargs)
@classmethod
def missing(cls, what: str = "参数", **kwargs) -> "ToolResult":
"""缺少参数"""
return cls.fail(f"缺少{what}", ErrorCode.MISSING_ARGS, **kwargs)
@classmethod
def not_found(cls, what: str = "资源", **kwargs) -> "ToolResult":
"""未找到"""
return cls.fail(f"{what}不存在", ErrorCode.NOT_FOUND, **kwargs)
@classmethod
def timeout(cls, seconds: int = 0, **kwargs) -> "ToolResult":
"""超时"""
msg = f"执行超时"
if seconds:
msg += f" ({seconds}秒)"
return cls.fail(msg, ErrorCode.TIMEOUT, **kwargs)
# ── 格式转换 ──
def to_dict(self) -> dict:
"""转为普通 dict(兼容旧接口)"""
if self.success:
return {"result": self.data if self.data is not None else "执行完成"}
return {"result": str(self)}
def to_mcp(self) -> dict:
"""转为 MCP 协议格式"""
text = str(self)
return {
"content": [{"type": "text", "text": text}],
"isError": not self.success,
}
def to_string(self) -> str:
"""转为字符串(兼容旧插件 handle() 返回值)"""
return str(self)