""" 沙盒MCP插件主模块 提供安全的文件操作和命令执行能力 """ import logging from pathlib import Path from app.config.settings import settings logger = logging.getLogger(__name__) class SandboxMCPPlugin: """沙盒MCP插件""" def __init__(self): self.name = "sandbox" self.version = "1.0.0" self.enabled = False self.sandbox_root = None self.security = None self.file_ops = None self.command_exec = None def on_enable(self): """插件启用时调用""" self.enabled = True logger.info(f"{self.name} on_enable 开始执行") try: # 确保沙盒目录存在 settings.ensure_sandbox_directory() self.sandbox_root = settings.SANDBOX_ROOT logger.info(f"{self.name} 沙盒目录: {self.sandbox_root}") # 初始化安全控制器和操作器(使用绝对导入) from plugins.sandbox.sandbox.security import SecurityController from plugins.sandbox.sandbox.file_ops import FileOperations from plugins.sandbox.sandbox.command_exec import CommandExecutor self.security = SecurityController() logger.info(f"{self.name} SecurityController 创建成功") self.file_ops = FileOperations(self.security) logger.info(f"{self.name} FileOperations 创建成功: {self.file_ops is not None}") self.command_exec = CommandExecutor(self.security) logger.info(f"{self.name} CommandExecutor 创建成功") logger.info(f"{self.name} v{self.version} 已启用,沙盒目录: {self.sandbox_root}") except Exception as e: logger.error(f"{self.name} on_enable 失败: {e}") import traceback traceback.print_exc() def on_disable(self): """插件禁用时调用""" self.enabled = False logger.info(f"{self.name} 已禁用") def get_status(self) -> dict: """获取插件状态""" file_count = 0 total_size = 0 if self.sandbox_root and self.sandbox_root.exists(): for file_path in self.sandbox_root.rglob("*"): if file_path.is_file(): file_count += 1 try: total_size += file_path.stat().st_size except Exception: pass tools = [ "sandbox-read", "sandbox-write", "sandbox-list", "sandbox-exec", ] return { "name": self.name, "version": self.version, "enabled": self.enabled, "sandbox_root": str(self.sandbox_root) if self.sandbox_root else "", "file_count": file_count, "total_size": total_size, "tools": tools, } # 创建插件实例 plugin = SandboxMCPPlugin()