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

1""" 

2文件操作工具集。 

3""" 

4 

5from __future__ import annotations 

6 

7import os 

8 

9import aiofiles 

10 

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

12 

13 

14class ReadFileTool(BaseTool): 

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, 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 name = "write_file" 

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

53 permission_level = PermissionLevel.MODERATE 

54 

55 @property 

56 def parameters(self) -> dict: 

57 return { 

58 "type": "object", 

59 "properties": { 

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

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

62 }, 

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

64 } 

65 

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

67 file_path = arguments["file_path"] 

68 content = arguments["content"] 

69 try: 

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

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

72 await f.write(content) 

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

74 except Exception as e: 

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

76 

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

78 return True 

79 

80 

81class ListDirectoryTool(BaseTool): 

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

83 

84 name = "list_directory" 

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

86 permission_level = PermissionLevel.SAFE 

87 

88 @property 

89 def parameters(self) -> dict: 

90 return { 

91 "type": "object", 

92 "properties": { 

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

94 }, 

95 "required": ["path"], 

96 } 

97 

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

99 path = arguments["path"] 

100 try: 

101 entries = os.listdir(path) 

102 lines = [] 

103 for entry in sorted(entries): 

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

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

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

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

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

109 except FileNotFoundError: 

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

111 except Exception as e: 

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