Coverage for agentos/tools/file_tools.py: 41%

1""" 

2文件操作工具集。 

3""" 

4 

5from __future__ import annotations 

6 

7import os 

8import aiofiles 

9 

10from agentos.tools.base import BaseTool, PermissionLevel, ToolResult 

11 

12 

13class ReadFileTool(BaseTool): 

14 

15 """文件读取工具。""" 

16 

17 name = "read_file" 

18 description = "读取文件内容，返回全部文本。" 

19 permission_level = PermissionLevel.SAFE 

20 

21 @property 

22 def parameters(self) -> dict: 

23 return { 

24 "type": "object", 

25 "properties": { 

26 "file_path": { 

27 "type": "string", 

28 "description": "要读取的文件绝对路径", 

29 }, 

30 }, 

31 "required": ["file_path"], 

32 } 

33 

34 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

35 file_path = arguments["file_path"] 

36 try: 

37 async with aiofiles.open(file_path, "r", encoding="utf-8") as f: 

38 content = await f.read() 

39 return ToolResult.ok("", output=content) 

40 except FileNotFoundError: 

41 return ToolResult.fail("", error=f"File not found: {file_path}") 

42 except PermissionError: 

43 return ToolResult.fail("", error=f"Permission denied: {file_path}") 

44 except Exception as e: 

45 return ToolResult.fail("", error=str(e)) 

46 

47 

48class WriteFileTool(BaseTool): 

49 

50 """文件写入工具。""" 

51 

52 name = "write_file" 

53 description = "写入文本内容到文件。如果文件已存在则覆盖。" 

54 permission_level = PermissionLevel.MODERATE 

55 

56 @property 

57 def parameters(self) -> dict: 

58 return { 

59 "type": "object", 

60 "properties": { 

61 "file_path": {"type": "string", "description": "要写入的文件路径"}, 

62 "content": {"type": "string", "description": "要写入的文本内容"}, 

63 }, 

64 "required": ["file_path", "content"], 

65 } 

66 

67 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

68 file_path = arguments["file_path"] 

69 content = arguments["content"] 

70 try: 

71 os.makedirs(os.path.dirname(file_path), exist_ok=True) 

72 async with aiofiles.open(file_path, "w", encoding="utf-8") as f: 

73 await f.write(content) 

74 return ToolResult.ok("", output=f"Written {len(content)} bytes to {file_path}") 

75 except Exception as e: 

76 return ToolResult.fail("", error=str(e)) 

77 

78 def is_write_operation(self, arguments: dict) -> bool: 

79 return True 

80 

81 

82class ListDirectoryTool(BaseTool): 

83 

84 """目录列表工具。""" 

85 

86 name = "list_directory" 

87 description = "列出目录下的所有文件和子目录。" 

88 permission_level = PermissionLevel.SAFE 

89 

90 @property 

91 def parameters(self) -> dict: 

92 return { 

93 "type": "object", 

94 "properties": { 

95 "path": {"type": "string", "description": "要列出的目录路径"}, 

96 }, 

97 "required": ["path"], 

98 } 

99 

100 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

101 path = arguments["path"] 

102 try: 

103 entries = os.listdir(path) 

104 lines = [] 

105 for entry in sorted(entries): 

106 full_path = os.path.join(path, entry) 

107 tag = "[DIR]" if os.path.isdir(full_path) else "[FILE]" 

108 size = os.path.getsize(full_path) if os.path.isfile(full_path) else 0 

109 lines.append(f"{tag} {entry} ({size} bytes)") 

110 return ToolResult.ok("", output="\n".join(lines)) 

111 except FileNotFoundError: 

112 return ToolResult.fail("", error=f"Directory not found: {path}") 

113 except Exception as e: 

114 return ToolResult.fail("", error=str(e))