refactor(agent): rename run_plugin_command to run_slash_command to avoid confusion with execute_command (shell)

2026-08-08 14:43:34 +08:00 · 2026-04-06 23:53:49 +08:00
parent 7806267e92
commit a8fb4a6d84
5 changed files with 16 additions and 16 deletions
--- a/app/agent/tools/impl/run_slash_command.py
+++ b/app/agent/tools/impl/run_slash_command.py
@@ -0,0 +1,114 @@
+"""运行斜杠命令工具（系统命令 + 插件命令）"""
+
+import json
+from typing import Optional, Type
+
+from pydantic import BaseModel, Field
+
+from app.agent.tools.base import MoviePilotTool
+from app.command import Command
+from app.core.event import eventmanager
+from app.log import logger
+from app.schemas.types import EventType, MessageChannel
+
+
+class RunSlashCommandInput(BaseModel):
+    """运行斜杠命令工具的输入参数模型"""
+
+    explanation: str = Field(
+        ...,
+        description="Clear explanation of why this tool is being used in the current context",
+    )
+    command: str = Field(
+        ...,
+        description="The slash command to execute, e.g. '/cookiecloud'. "
+        "Must start with '/'. Can include arguments after the command, e.g. '/command arg1 arg2'. "
+        "Use query_plugin_capabilities tool to discover available plugin commands, "
+        "or list_all_commands tool to discover all available commands (including system commands).",
+    )
+
+
+class RunSlashCommandTool(MoviePilotTool):
+    name: str = "run_slash_command"
+    description: str = (
+        "Execute a slash command (system or plugin) by sending a CommandExcute event. "
+        "This tool supports ALL registered slash commands, including: "
+        "1) System preset commands (e.g. /cookiecloud, /sites, /subscribes, /downloading, /transfer, /restart, etc.) "
+        "2) Plugin commands registered by installed plugins. "
+        "Use the query_plugin_capabilities tool to discover plugin commands, "
+        "or the list_all_commands tool to discover all available commands. "
+        "The command will be executed asynchronously. "
+        "Note: This tool triggers the command execution but the actual processing happens in the background."
+    )
+    args_schema: Type[BaseModel] = RunSlashCommandInput
+    require_admin: bool = True
+
+    def get_tool_message(self, **kwargs) -> Optional[str]:
+        """生成友好的提示消息"""
+        command = kwargs.get("command", "")
+        return f"正在执行命令: {command}"
+
+    async def run(self, command: str, **kwargs) -> str:
+        logger.info(f"执行工具: {self.name}, 参数: command={command}")
+
+        try:
+            # 确保命令以 / 开头
+            if not command.startswith("/"):
+                command = f"/{command}"
+
+            # 从全局 Command 单例中验证命令是否存在（包含系统预设命令 + 插件命令 + 其他命令）
+            cmd_name = command.split()[0]
+            command_obj = Command()
+            matched_command = command_obj.get(cmd_name)
+
+            if not matched_command:
+                # 列出所有可用命令帮助用户
+                all_commands = command_obj.get_commands()
+                available_cmds = [
+                    f"{cmd} - {info.get('description', '无描述')}"
+                    for cmd, info in all_commands.items()
+                ]
+                result = {
+                    "success": False,
+                    "message": f"命令 {cmd_name} 不存在",
+                }
+                if available_cmds:
+                    result["available_commands"] = available_cmds
+                return json.dumps(result, ensure_ascii=False, indent=2)
+
+            # 构建消息渠道，优先使用当前会话的渠道信息
+            channel = None
+            if self._channel:
+                try:
+                    channel = MessageChannel(self._channel)
+                except (ValueError, KeyError):
+                    channel = None
+
+            # 发送命令执行事件，与 message.py 中的方式一致
+            eventmanager.send_event(
+                EventType.CommandExcute,
+                {
+                    "cmd": command,
+                    "user": self._user_id,
+                    "channel": channel,
+                    "source": self._source,
+                },
+            )
+
+            result = {
+                "success": True,
+                "message": f"命令 {cmd_name} 已触发执行",
+                "command": command,
+                "command_desc": matched_command.get("description", ""),
+            }
+            # 如果是插件命令，附加插件ID
+            if matched_command.get("pid"):
+                result["plugin_id"] = matched_command["pid"]
+            return json.dumps(result, ensure_ascii=False, indent=2)
+
+        except Exception as e:
+            logger.error(f"执行命令失败: {e}", exc_info=True)
+            return json.dumps(
+                {"success": False, "message": f"执行命令时发生错误: {str(e)}"},
+                ensure_ascii=False,
+            )