Coverage for merco/tools/bash_tools.py: 66%

1"""终端命令执行工具""" 

2 

3import asyncio 

4import subprocess 

5from .base import BaseTool 

6 

7 

8class BashTool(BaseTool): 

9 """执行 bash 命令""" 

10 

11 name = "bash" 

12 description = "在终端执行 shell 命令" 

13 toolset = "bash" 

14 parameters = { 

15 "type": "object", 

16 "properties": { 

17 "command": {"type": "string", "description": "要执行的命令"}, 

18 "timeout": {"type": "integer", "description": "超时时间（秒）"}, 

19 "workdir": {"type": "string", "description": "工作目录"}, 

20 }, 

21 "required": ["command"], 

22 } 

23 

24 def __init__(self): 

25 self._active_processes: set[asyncio.subprocess.Process] = set() 

26 

27 async def execute(self, command: str, timeout: int = 60, workdir: str = None) -> dict: 

28 try: 

29 process = await asyncio.create_subprocess_shell( 

30 command, 

31 stdout=subprocess.PIPE, 

32 stderr=subprocess.PIPE, 

33 cwd=workdir, 

34 ) 

35 self._active_processes.add(process) 

36 

37 try: 

38 stdout, stderr = await asyncio.wait_for( 

39 process.communicate(), timeout=timeout 

40 ) 

41 return { 

42 "stdout": stdout.decode("utf-8", errors="replace") if stdout else "", 

43 "stderr": stderr.decode("utf-8", errors="replace") if stderr else "", 

44 "returncode": process.returncode, 

45 } 

46 except asyncio.TimeoutError: 

47 process.kill() 

48 return {"error": f"Command timed out after {timeout}s"} 

49 finally: 

50 self._active_processes.discard(process) 

51 

52 except Exception as e: 

53 return {"error": str(e)} 

54 

55 def kill_all(self): 

56 """终止所有活跃的子进程。""" 

57 for proc in self._active_processes: 

58 try: 

59 proc.kill() 

60 except ProcessLookupError: 

61 pass 

62 self._active_processes.clear() 

63 

64 

65from .registry import tool_registry # noqa: E402 — 模块末尾自注册 

66tool_registry.register(BashTool())