Coverage for agentos/tools/base.py: 89%

1""" 

2工具基类 — 所有工具的抽象父类。 

3""" 

4 

5from __future__ import annotations 

6 

7from abc import ABC, abstractmethod 

8from dataclasses import dataclass, field 

9from enum import Enum 

10from typing import Any 

11 

12 

13class PermissionLevel(str, Enum): 

14 """工具权限等级。""" 

15 

16 SAFE = "safe" 

17 MODERATE = "moderate" 

18 SENSITIVE = "sensitive" 

19 

20 

21@dataclass 

22class ToolCall: 

23 """工具调用请求。""" 

24 

25 id: str 

26 name: str 

27 arguments: dict[str, Any] 

28 

29 

30@dataclass 

31class ToolResult: 

32 """工具调用返回结果。""" 

33 

34 call_id: str 

35 output: str | None = None 

36 error: str | None = None 

37 exit_code: int | None = None 

38 

39 @classmethod 

40 def ok(cls, call_id: str, output: str) -> "ToolResult": 

41 return cls(call_id=call_id, output=output) 

42 

43 @classmethod 

44 def fail(cls, call_id: str, error: str) -> "ToolResult": 

45 return cls(call_id=call_id, error=error) 

46 

47 

48class BaseTool(ABC): 

49 """工具基类 — 所有工具必须实现的接口。""" 

50 

51 name: str = "" 

52 description: str = "" 

53 permission_level: PermissionLevel = PermissionLevel.MODERATE 

54 concurrent_safe: bool = True # 是否可与其他工具并行执行 

55 

56 @property 

57 @abstractmethod 

58 def parameters(self) -> dict: 

59 """返回OpenAI function calling格式的参数schema。""" 

60 

61 @abstractmethod 

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

63 """执行工具。""" 

64 

65 def to_openai_schema(self) -> dict: 

66 return { 

67 "type": "function", 

68 "function": { 

69 "name": self.name, 

70 "description": self.description, 

71 "parameters": self.parameters, 

72 }, 

73 } 

74 

75 def to_anthropic_schema(self) -> dict: 

76 return { 

77 "name": self.name, 

78 "description": self.description, 

79 "input_schema": self.parameters, 

80 } 

81 

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

83 """是否为写操作（用于读写冲突检测）。默认False。""" 

84 return False 

85 

86 def is_read_operation(self, arguments: dict) -> bool: 

87 """是否为读操作。默认True。""" 

88 return True 

89 

90 def extract_target_path(self, arguments: dict) -> str | None: 

91 """提取操作的目标路径（用于冲突检测）。默认None。""" 

92 return arguments.get("file_path") or arguments.get("path")