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

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 typing import Callable, Optional 

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__(self, skill_name: str, skill_run: Callable[..., str], 

28 description: str = "", parameters: Optional[dict] = None): 

29 self._skill_name = skill_name 

30 self._run_fn = skill_run 

31 self._description = description 

32 self._parameters = parameters 

33 

34 @property 

35 def name(self) -> str: 

36 return self._skill_name 

37 

38 @property 

39 def description(self) -> str: 

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

41 

42 @property 

43 def parameters(self) -> dict: 

44 if self._parameters: 

45 return self._parameters 

46 # Default: accept arbitrary kwargs 

47 return { 

48 "type": "object", 

49 "properties": { 

50 "kwargs": { 

51 "type": "string", 

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

53 } 

54 }, 

55 } 

56 

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

58 try: 

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

60 if "kwargs" in input_data: 

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

62 else: 

63 parsed = input_data 

64 

65 result = self._run_fn(**parsed) 

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

67 except Exception as e: 

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

69 

70 

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

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

73 

74 Args: 

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

76 

77 Returns: 

78 SkillTool 实例列表。 

79 """ 

80 if skills_dir is None: 

81 # Auto-locate 

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

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

84 

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

86 return [] 

87 

88 tools: list[SkillTool] = [] 

89 

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

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

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

93 continue 

94 

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

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

97 continue 

98 

99 try: 

100 # Import the skill module 

101 spec = importlib.util.spec_from_file_location( 

102 f"agentos_marketplace_skill_{entry}", skill_py 

103 ) 

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

105 continue 

106 mod = importlib.util.module_from_spec(spec) 

107 sys.modules[spec.name] = mod 

108 spec.loader.exec_module(mod) 

109 

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

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

112 continue 

113 

114 # Get docstring as description 

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

116 

117 # Try to get parameter schema from function signature 

118 params = _infer_parameters(run_fn) 

119 

120 tool = SkillTool( 

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

122 skill_run=run_fn, 

123 description=desc, 

124 parameters=params, 

125 ) 

126 tools.append(tool) 

127 except Exception: 

128 # Skip skills that fail to load 

129 continue 

130 

131 return tools 

132 

133 

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

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

136 try: 

137 sig = inspect.signature(fn) 

138 except (ValueError, TypeError): 

139 return { 

140 "type": "object", 

141 "properties": { 

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

143 }, 

144 } 

145 

146 properties = {} 

147 required = [] 

148 

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

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

151 continue 

152 param_type = "string" 

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

154 anno = param.annotation 

155 if anno is str: 

156 param_type = "string" 

157 elif anno is int: 

158 param_type = "integer" 

159 elif anno is float: 

160 param_type = "number" 

161 elif anno is bool: 

162 param_type = "boolean" 

163 elif anno is list: 

164 param_type = "array" 

165 

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

167 

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

169 required.append(name) 

170 

171 return { 

172 "type": "object", 

173 "properties": properties, 

174 "required": required, 

175 }