pythonfrom typing import Dict, Typefrom .base import Toolclass ToolRegistry:"""工具注册表，单例模式，管理所有工具的注册与调用"""_instance = Nonedef __new__(cls, *args, **kwargs):单例模式：确保整个系统只有一个ToolRegistry实例if cls._instance is None:cls._instance = super().new(cls)cls._instance.tools: Dict[str, Tool] = {} # 存储工具：key=工具name，value=工具实例return cls._instancedef register_tool(self, tool: Tool) -> None:"""注册工具：将工具实例添加到注册表中"""if tool.name in self.tools:raise ValueError(f"工具 {tool.name} 已注册，不可重复注册")self.tools[tool.name] = tooldef get_tool(self, tool_name: str) -> Tool:"""获取工具：根据工具name，从注册表中获取工具实例"""if tool_name not in self.tools:raise ValueError(f"工具 {tool_name} 未注册")return self.tools[tool_name]async def execute_tool(self, tool_name: str, kwargs) -> str:"""执行工具：根据工具name获取工具，调用execute方法执行"""tool = self.get_tool(tool_name)return await tool.execute( kwargs)

pythonfrom .base import Toolfrom typing import Dict, Anyimport osclass ReadFileTool(Tool):@propertydef name(self) -> str:return "read_file" 工具唯一标识，与OpenClaw保持一致@propertydef description(self) -> str:return "读取指定路径的文件内容，支持绝对路径和相对路径；如果文件不存在，返回错误信息。" # 让LLM理解工具功能@propertydef parameters(self) -> Dict[str, Any]:# JSON Schema格式，定义输入参数，让LLM知道需要传入什么参数return {"type": "object","properties": {"file_path": {"type": "string","description": "文件路径，支持绝对路径和相对路径"}},"required": ["file_path"] # 必须传入的参数}async def execute(self, **kwargs) -> str:# 核心执行逻辑：读取文件内容file_path = kwargs.get("file_path")if not os.path.exists(file_path):return f"错误：文件 {file_path} 不存在"try:with open(file_path, "r", encoding="utf-8") as f:content = f.read()return f"文件 {file_path} 内容如下：\n{content}"except Exception as e:return f"读取文件失败：{str(e)}"

pythonfrom .base import Toolfrom typing import Dict, Anyimport asyncioclass ExecTool(Tool):@propertydef name(self) -> str:return "exec"@propertydef description(self) -> str:return "执行Shell命令，返回命令执行结果；支持超时控制（默认30秒），避免命令阻塞。"@propertydef parameters(self) -> Dict[str, Any]:return {"type": "object","properties": {"command": {"type": "string","description": "要执行的Shell命令（如ls、pwd）"},"timeout": {"type": "integer","description": "超时时间（秒），默认30秒","default": 30}},"required": ["command"]}async def execute(self, **kwargs) -> str:command = kwargs.get("command")timeout = kwargs.get("timeout", 30)try:异步执行Shell命令，避免阻塞process = await asyncio.create_subprocess_shell(command,stdout=asyncio.subprocess.PIPE,stderr=asyncio.subprocess.PIPE)# 超时控制stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)if process.returncode == 0:return f"命令执行成功：\n{stdout.decode('utf-8')}"else:return f"命令执行失败：\n{stderr.decode('utf-8')}"except asyncio.TimeoutError:return f"命令执行超时（超过{timeout}秒）"except Exception as e:return f"命令执行异常：{str(e)}"