Coverage for agentos/tools/skill_tool.py: 0%

1""" 

2SkillTool — 将 marketplace skill 包装为 BaseTool，通过 Bridge 注册到 ToolAgent。 

3 

4每个 skill 的 run(**kwargs) 函数变成 Agent 可直接调用的工具。 

5""" 

6 

7from __future__ import annotations 

8 

9import importlib 

10import inspect 

11import json 

12import os 

13import sys 

14from collections.abc import Callable 

15 

16from agentos.tools.base import BaseTool, ToolResult 

17 

18 

19class SkillTool(BaseTool): 

20 """将一个 marketplace skill 的 run() 函数包装为 BaseTool。 

21 

22 skill 的 run(**kwargs) 接收任意关键字参数并返回字符串。 

23 """ 

24 

25 permission_level = "safe" # type: ignore 

26 

27 def __init__( 

28 self, 

29 skill_name: str, 

30 skill_run: Callable[..., str], 

31 description: str = "", 

32 parameters: dict | None = None, 

33 ): 

34 self._skill_name = skill_name 

35 self._run_fn = skill_run 

36 self._description = description 

37 self._parameters = parameters 

38 

39 @property 

40 def name(self) -> str: 

41 return self._skill_name 

42 

43 @property 

44 def description(self) -> str: 

45 return self._description or f"Execute the '{self._skill_name}' skill." 

46 

47 @property 

48 def parameters(self) -> dict: 

49 if self._parameters: 

50 return self._parameters 

51 # Default: accept arbitrary kwargs 

52 return { 

53 "type": "object", 

54 "properties": { 

55 "kwargs": { 

56 "type": "string", 

57 "description": "JSON string of keyword arguments to pass to the skill", 

58 } 

59 }, 

60 } 

61 

62 async def execute(self, input_data: dict) -> ToolResult: 

63 try: 

64 # If input has 'kwargs', parse as JSON 

65 if "kwargs" in input_data: 

66 parsed = json.loads(input_data["kwargs"]) 

67 else: 

68 parsed = input_data 

69 

70 result = self._run_fn(**parsed) 

71 return ToolResult(call_id=self.name, output=str(result)) 

72 except Exception as e: 

73 return ToolResult(call_id=self.name, error=str(e)) 

74 

75 

76def discover_skills(skills_dir: str = None) -> list[SkillTool]: 

77 """自动发现 marketplace/skills 下所有 skill 并包装为 SkillTool。 

78 

79 Args: 

80 skills_dir: skills 目录路径。默认自动定位。 

81 

82 Returns: 

83 SkillTool 实例列表。 

84 """ 

85 if skills_dir is None: 

86 # Auto-locate 

87 agentos_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 

88 skills_dir = os.path.join(agentos_dir, "marketplace", "skills") 

89 

90 if not os.path.isdir(skills_dir): 

91 return [] 

92 

93 tools: list[SkillTool] = [] 

94 

95 for entry in sorted(os.listdir(skills_dir)): 

96 skill_path = os.path.join(skills_dir, entry) 

97 if not os.path.isdir(skill_path): 

98 continue 

99 

100 skill_py = os.path.join(skill_path, f"{entry}.py") 

101 if not os.path.isfile(skill_py): 

102 continue 

103 

104 try: 

105 # Import the skill module 

106 spec = importlib.util.spec_from_file_location( 

107 f"agentos_marketplace_skill_{entry}", skill_py 

108 ) 

109 if spec is None or spec.loader is None: 

110 continue 

111 mod = importlib.util.module_from_spec(spec) 

112 sys.modules[spec.name] = mod 

113 spec.loader.exec_module(mod) 

114 

115 run_fn = getattr(mod, "run", None) 

116 if run_fn is None or not callable(run_fn): 

117 continue 

118 

119 # Get docstring as description 

120 desc = (mod.__doc__ or f"Execute the '{entry}' skill.").strip().split("\n")[0] 

121 

122 # Try to get parameter schema from function signature 

123 params = _infer_parameters(run_fn) 

124 

125 tool = SkillTool( 

126 skill_name=f"skill_{entry.replace('-', '_')}", 

127 skill_run=run_fn, 

128 description=desc, 

129 parameters=params, 

130 ) 

131 tools.append(tool) 

132 except Exception: 

133 # Skip skills that fail to load 

134 continue 

135 

136 return tools 

137 

138 

139def _infer_parameters(fn: Callable) -> dict: 

140 """从函数签名推断 JSON Schema 参数定义。""" 

141 try: 

142 sig = inspect.signature(fn) 

143 except (ValueError, TypeError): 

144 return { 

145 "type": "object", 

146 "properties": {"kwargs": {"type": "string", "description": "JSON string of arguments"}}, 

147 } 

148 

149 properties = {} 

150 required = [] 

151 

152 for name, param in sig.parameters.items(): 

153 if name in ("self", "cls"): 

154 continue 

155 param_type = "string" 

156 if param.annotation is not inspect.Parameter.empty: 

157 anno = param.annotation 

158 if anno is str: 

159 param_type = "string" 

160 elif anno is int: 

161 param_type = "integer" 

162 elif anno is float: 

163 param_type = "number" 

164 elif anno is bool: 

165 param_type = "boolean" 

166 elif anno is list: 

167 param_type = "array" 

168 

169 properties[name] = {"type": param_type, "description": f"Parameter: {name}"} 

170 

171 if param.default is inspect.Parameter.empty: 

172 required.append(name) 

173 

174 return { 

175 "type": "object", 

176 "properties": properties, 

177 "required": required, 

178 }