From bac5ea01a97e8d57e1f71ddb5805246360faebd7 Mon Sep 17 00:00:00 2001 From: shuiyu1123 Date: Fri, 19 Jun 2026 00:26:49 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=BB=9F=E4=B8=80=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E5=A4=84=E7=90=86=20-=20ToolResult=20=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 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 --- plugins/cmd_executor.py | 30 ++++++---- plugins/file_manager.py | 105 ++++++++++++++++++---------------- unified_tool_manager.py | 64 ++++++++++++++------- xcli_core/tool_result.py | 119 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 235 insertions(+), 83 deletions(-) create mode 100644 xcli_core/tool_result.py diff --git a/plugins/cmd_executor.py b/plugins/cmd_executor.py index 394f485..07a6ff5 100644 --- a/plugins/cmd_executor.py +++ b/plugins/cmd_executor.py @@ -5,6 +5,11 @@ import os import re import subprocess 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: @@ -68,18 +73,18 @@ class Liugin: ':(){ :|:& };:', 'chmod -R 777 /', 'mv / ', 'wget http', } - def handle(self, args: str) -> str: + def handle(self, args: str): try: parts = args.strip().split(maxsplit=1) if not parts: - return "错误：请提供操作。可用: run <命令>" + return ToolResult.missing("操作。可用: run <命令>") operation = parts[0].lower() if operation != "run": - return f"错误：不支持的操作 '{operation}'。可用: run" + return ToolResult.fail(f"不支持的操作 '{operation}'。可用: run", ErrorCode.UNSUPPORTED_OP) if len(parts) < 2: - return "错误：请提供要执行的命令" + return ToolResult.missing("要执行的命令") rest = parts[1] @@ -91,21 +96,20 @@ class Liugin: rest = re.sub(r'--timeout\s+\d+', '', rest).strip() if not rest: - return "错误：请提供要执行的命令" + return ToolResult.missing("要执行的命令") # 安全检查 for blocked in self.BLOCKED: if blocked in rest: - return f" 安全拦截: 检测到危险命令模式 '{blocked}'" + return ToolResult.fail(f"安全拦截: 检测到危险命令模式 '{blocked}'", ErrorCode.PERMISSION_DENIED) return self._execute(rest, timeout) 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: - # Windows 下用 shell=True，Unix 下直接执行 use_shell = platform.system() == 'Windows' result = subprocess.run( @@ -136,9 +140,11 @@ class Liugin: if len(output) > 5000: 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: - return f" 命令超时 ({timeout}s): {command}" + return ToolResult.timeout(timeout, tool_name="cmd_executor") except Exception as e: - return f"执行失败: {str(e)}" + return ToolResult.fail(str(e), ErrorCode.EXEC_FAILED, tool_name="cmd_executor") diff --git a/plugins/file_manager.py b/plugins/file_manager.py index 6fd1c23..e45e1b1 100644 --- a/plugins/file_manager.py +++ b/plugins/file_manager.py @@ -5,10 +5,15 @@ import os import json import shutil +import sys from datetime import datetime from pathlib import Path 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: """文件管理器 - 文件和目录的增删改查""" @@ -70,11 +75,11 @@ class Liugin: path = arguments.get("path", "") return f"{op} {path}".strip() - def handle(self, args: str) -> str: + def handle(self, args: str): try: parts = self._parse_args(args) if not parts: - return "错误：请提供操作类型" + return ToolResult.missing("操作类型") operation = parts[0].lower() handlers = { @@ -92,7 +97,7 @@ class Liugin: handler = handlers.get(operation) if not handler: - return f"错误：不支持的操作 '{operation}'" + return ToolResult.fail(f"不支持的操作 '{operation}'", ErrorCode.UNSUPPORTED_OP) return handler(parts[1:]) except Exception as e: @@ -115,7 +120,7 @@ class Liugin: parts.append(current) return parts - def _op_list(self, args: List[str]) -> str: + def _op_list(self, args: List[str]): path, pattern = ".", None i = 0 while i < len(args): @@ -127,7 +132,7 @@ class Liugin: path = os.path.abspath(path) if not os.path.isdir(path): - return f"错误：不是目录: {path}" + return ToolResult.not_found(f"目录: {path}") items = [] try: @@ -153,13 +158,13 @@ class Liugin: result.append(f" ... 还有 {len(items) - 100} 项") result.append("─" * 50) result.append(f"共 {dirs} 个文件夹, {files} 个文件") - return '\n'.join(result) + return ToolResult.ok('\n'.join(result), tool_name="file_manager") 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: - return "错误：请提供文件路径" + return ToolResult.missing("文件路径") file_path = os.path.abspath(args[0]) max_lines = None i = 1 @@ -168,16 +173,16 @@ class Liugin: try: max_lines = int(args[i + 1]) except ValueError: - return "错误：行数必须是整数" + return ToolResult.fail("行数必须是整数", ErrorCode.INVALID_ARGS) i += 1 i += 1 if not os.path.isfile(file_path): - return f"错误：文件不存在: {file_path}" + return ToolResult.not_found(f"文件: {file_path}") size = os.path.getsize(file_path) 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: 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)) 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: - 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: - return "错误：格式: write <文件> -c <内容>" + return ToolResult.fail("格式: write <文件> -c <内容>", ErrorCode.MISSING_ARGS) fp = os.path.abspath(args[0]) content = None i = 1 @@ -210,15 +215,15 @@ class Liugin: content = args[i + 1]; i += 1 i += 1 if content is None: - return "错误：请用 -c 提供内容" + return ToolResult.missing("内容 (-c 参数)") os.makedirs(os.path.dirname(fp) or ".", exist_ok=True) with open(fp, 'w', encoding='utf-8') as f: 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: - return "错误：格式: append <文件> -c <内容>" + return ToolResult.fail("格式: append <文件> -c <内容>", ErrorCode.MISSING_ARGS) fp = os.path.abspath(args[0]) content = None i = 1 @@ -227,59 +232,59 @@ class Liugin: content = args[i + 1]; i += 1 i += 1 if content is None: - return "错误：请用 -c 提供内容" + return ToolResult.missing("内容 (-c 参数)") if not os.path.isfile(fp): - return f"错误：文件不存在: {fp}" + return ToolResult.not_found(f"文件: {fp}") with open(fp, 'a', encoding='utf-8') as f: 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: - return "错误：请提供源和目标路径" + return ToolResult.missing("源和目标路径") src, dst = os.path.abspath(args[0]), os.path.abspath(args[1]) 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) if os.path.isfile(src): shutil.copy2(src, dst) else: 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: - return "错误：请提供源和目标路径" + return ToolResult.missing("源和目标路径") src, dst = os.path.abspath(args[0]), os.path.abspath(args[1]) 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) 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: - return "错误：请提供路径" + return ToolResult.missing("路径") path = os.path.abspath(args[0]) recursive = "-r" in args if not os.path.exists(path): - return f"错误：不存在: {path}" + return ToolResult.not_found(f"路径: {path}") if os.path.isfile(path): os.remove(path) - return f" 已删除文件: {path}" + return ToolResult.ok(f"已删除文件: {path}", tool_name="file_manager") elif os.path.isdir(path): if recursive: shutil.rmtree(path) - return f" 已删除目录: {path}" + return ToolResult.ok(f"已删除目录: {path}", tool_name="file_manager") elif not os.listdir(path): os.rmdir(path) - return f" 已删除空目录: {path}" + return ToolResult.ok(f"已删除空目录: {path}", tool_name="file_manager") 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: - return "错误：请提供搜索目录" + return ToolResult.missing("搜索目录") directory = os.path.abspath(args[0]) pattern, content_search = None, None i = 1 @@ -291,7 +296,7 @@ class Liugin: i += 1 if not os.path.isdir(directory): - return f"错误：目录不存在: {directory}" + return ToolResult.not_found(f"目录: {directory}") results = [] for root, dirs, files in os.walk(directory): @@ -317,14 +322,14 @@ class Liugin: if len(results) >= 50: 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: - return "错误：请提供路径" + return ToolResult.missing("路径") path = os.path.abspath(args[0]) if not os.path.exists(path): - return f"错误：不存在: {path}" + return ToolResult.not_found(f"路径: {path}") stat = os.stat(path) result = [ @@ -337,13 +342,13 @@ class Liugin: if os.path.isfile(path): ext = os.path.splitext(path)[1] 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: - return "错误：请提供目录路径" + return ToolResult.missing("目录路径") path = os.path.abspath(args[0]) if os.path.exists(path): - return f"错误：已存在: {path}" + return ToolResult.fail(f"已存在: {path}", ErrorCode.INVALID_ARGS) os.makedirs(path, exist_ok=True) - return f" 已创建: {path}" + return ToolResult.ok(f"已创建: {path}", tool_name="file_manager") diff --git a/unified_tool_manager.py b/unified_tool_manager.py index 415a618..289d7af 100644 --- a/unified_tool_manager.py +++ b/unified_tool_manager.py @@ -9,6 +9,7 @@ import logging import threading from typing import Dict, List, Any, Optional, Union, Callable from colorama import Fore, Style +from xcli_core.tool_result import ToolResult, ErrorCode logger = logging.getLogger(__name__) @@ -192,7 +193,8 @@ class UnifiedToolManager: """为 Skill 创建兼容 Plugin 的 handler""" from skills.base import MarkdownSkill, Skill - def handler(args: str) -> str: + def handler(args: str): + """Skill handler - 返回 ToolResult""" kwargs = {} args = args.strip() @@ -212,16 +214,16 @@ class UnifiedToolManager: kwargs = {'operation': args} 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'): result = skill.run(**kwargs) if result.success: - return result.message or str(result.data) + return ToolResult.ok(result.message or str(result.data)) 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 @@ -312,12 +314,13 @@ class UnifiedToolManager: """获取工具使用的协议""" return self._protocol_map.get(name) - def execute(self, tool_name: str, args: str = "", **kwargs) -> Dict[str, Any]: - """执行工具调用""" - tool = self._tools.get(tool_name) + def execute(self, tool_name: str, args: str = "", **kwargs) -> ToolResult: + """执行工具调用，返回 ToolResult""" + with self._lock: + tool = self._tools.get(tool_name) if not tool: - return {"result": f"未找到工具: {tool_name}"} + return ToolResult.fail(f"未找到工具: {tool_name}", ErrorCode.NOT_FOUND, tool_name=tool_name) try: handler = tool['handler'] @@ -327,7 +330,10 @@ class UnifiedToolManager: if protocol == 'skill' and kwargs: skill = tool['instance'] 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: 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) 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: - 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 协议支持 ==================== @@ -355,7 +370,7 @@ class UnifiedToolManager: tool = self._tools.get(tool_name) 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: protocol = tool['protocol'] @@ -364,25 +379,30 @@ class UnifiedToolManager: if protocol == 'liugin': args = self._convert_mcp_args_to_plugin_args(tool_name, arguments, instance) 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': from skills.base import 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'): result = instance.run(**arguments) 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: - 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: - 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], instance: Any) -> str: @@ -423,10 +443,12 @@ class UnifiedToolManager: return s 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]: - return {"content": [{"type": "text", "text": error_msg}], "isError": True} + """MCP 错误响应（兼容旧接口，推荐用 ToolResult.fail().to_mcp()）""" + return ToolResult.fail(error_msg).to_mcp() # ==================== 兼容性 API ==================== diff --git a/xcli_core/tool_result.py b/xcli_core/tool_result.py new file mode 100644 index 0000000..0d507a3 --- /dev/null +++ b/xcli_core/tool_result.py @@ -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)