Coverage for agentos/tools/generator.py: 23%

1""" 

2OpenAPI工具自动生成器 — 从OpenAPI/Swagger spec自动生成Agent工具包装器。 

3v0.50: 新增模块。将REST API端点自动转换为Agent可调用的ToolCall格式。 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9import re 

10from dataclasses import dataclass, field 

11from pathlib import Path 

12from typing import Any 

13 

14import httpx 

15import yaml 

16 

17 

18@dataclass 

19class GeneratedTool: 

20 """单个生成的工具描述。""" 

21 

22 name: str 

23 description: str 

24 operation_id: str = "" 

25 method: str = "GET" 

26 path: str = "" 

27 parameters_schema: dict = field(default_factory=dict) 

28 auth_header: str = "" 

29 base_url: str = "" 

30 

31 def to_openai_function(self) -> dict: 

32 """转换为OpenAI function calling格式。""" 

33 func = { 

34 "type": "function", 

35 "function": { 

36 "name": self.name, 

37 "description": self.description, 

38 }, 

39 } 

40 if self.parameters_schema: 

41 func["function"]["parameters"] = self.parameters_schema 

42 return func 

43 

44 def to_tool_dict(self) -> dict: 

45 """转换为通用工具描述字典。""" 

46 return { 

47 "name": self.name, 

48 "description": self.description, 

49 "operation_id": self.operation_id, 

50 "method": self.method, 

51 "path_template": self.path, 

52 "parameters": self.parameters_schema, 

53 "base_url": self.base_url, 

54 "auth_header": self.auth_header, 

55 } 

56 

57 

58class OpenAPIToolGenerator: 

59 """ 

60 从OpenAPI 3.x / Swagger 2.0 spec生成Agent工具。 

61 

62 用法: 

63 gen = OpenAPIToolGenerator("https://api.example.com/openapi.json") 

64 tools = await gen.generate() 

65 # tools是GeneratedTool列表，可直接注入Agent context 

66 """ 

67 

68 PARAM_TYPE_MAP = { 

69 "string": {"type": "string"}, 

70 "integer": {"type": "integer"}, 

71 "number": {"type": "number"}, 

72 "boolean": {"type": "boolean"}, 

73 "array": {"type": "array", "items": {"type": "string"}}, 

74 "object": {"type": "object"}, 

75 } 

76 

77 def __init__( 

78 self, 

79 spec_url: str = "", 

80 spec_path: str = "", 

81 api_base: str = "", 

82 auth_header: str = "Authorization", 

83 auth_value: str = "", 

84 ): 

85 self.spec_url = spec_url 

86 self.spec_path = spec_path 

87 self.api_base = api_base 

88 self.auth_header = auth_header 

89 self.auth_value = auth_value 

90 self._http = httpx.AsyncClient(timeout=30) 

91 

92 async def load_spec(self) -> dict: 

93 """加载OpenAPI spec（URL或本地文件）。""" 

94 if self.spec_url: 

95 resp = await self._http.get(self.spec_url) 

96 resp.raise_for_status() 

97 if self.spec_url.endswith((".yaml", ".yml")): 

98 return yaml.safe_load(resp.text) 

99 return resp.json() 

100 

101 if self.spec_path: 

102 path = Path(self.spec_path) 

103 text = path.read_text(encoding="utf-8") 

104 if path.suffix in (".yaml", ".yml"): 

105 return yaml.safe_load(text) 

106 return json.loads(text) 

107 

108 raise ValueError("spec_url or spec_path required") 

109 

110 async def generate(self, filter_tag: str = "", max_tools: int = 100) -> list[GeneratedTool]: 

111 """解析spec并生成工具列表。""" 

112 spec = await self.load_spec() 

113 tools: list[GeneratedTool] = [] 

114 base_url = self.api_base or self._extract_base_url(spec) 

115 paths = spec.get("paths", {}) 

116 

117 for path_url, methods in paths.items(): 

118 if not isinstance(methods, dict): 

119 continue 

120 for method, operation in methods.items(): 

121 if method.upper() not in ("GET", "POST", "PUT", "DELETE", "PATCH"): 

122 continue 

123 if not isinstance(operation, dict): 

124 continue 

125 

126 tags = operation.get("tags", []) 

127 if filter_tag and filter_tag not in tags: 

128 continue 

129 

130 tool = self._build_tool(path_url, method, operation, base_url) 

131 tools.append(tool) 

132 if len(tools) >= max_tools: 

133 return tools 

134 

135 return tools 

136 

137 def _build_tool( 

138 self, path_url: str, method: str, operation: dict, base_url: str 

139 ) -> GeneratedTool: 

140 """从单个endpoint构建GeneratedTool。""" 

141 operation_id = operation.get("operationId", self._generate_operation_id(method, path_url)) 

142 summary = operation.get("summary", "") 

143 description = operation.get("description", summary or f"{method.upper()} {path_url}") 

144 tool_name = self._sanitize_name(operation_id) 

145 

146 schema = self._build_parameters_schema(operation) 

147 return GeneratedTool( 

148 name=tool_name, 

149 description=description, 

150 operation_id=operation_id, 

151 method=method.upper(), 

152 path=path_url, 

153 parameters_schema=schema, 

154 base_url=base_url, 

155 auth_header=self.auth_header, 

156 ) 

157 

158 def _extract_base_url(self, spec: dict) -> str: 

159 """提取API base URL。""" 

160 servers = spec.get("servers", []) 

161 if servers: 

162 return servers[0].get("url", "") 

163 host = spec.get("host", "") 

164 base_path = spec.get("basePath", "") 

165 schemes = spec.get("schemes", ["https"]) 

166 if host: 

167 return f"{schemes[0]}://{host}{base_path}" 

168 return "" 

169 

170 def _build_parameters_schema(self, operation: dict) -> dict: 

171 """构建parameters JSON Schema。""" 

172 properties: dict[str, Any] = {} 

173 required: list[str] = [] 

174 

175 # 路径/查询/header参数 

176 for param in operation.get("parameters", []): 

177 name = param["name"] 

178 schema = param.get("schema", {}) 

179 param_type = schema.get("type") or param.get("type", "string") 

180 properties[name] = self.PARAM_TYPE_MAP.get(param_type, {"type": "string"}) 

181 description = param.get("description", "") 

182 if description: 

183 properties[name]["description"] = description 

184 if param.get("required"): 

185 required.append(name) 

186 

187 # requestBody (POST/PUT/PATCH) 

188 request_body = operation.get("requestBody", {}) 

189 content = request_body.get("content", {}) 

190 json_content = content.get("application/json", {}) 

191 json_schema = json_content.get("schema", {}) 

192 if json_schema.get("properties"): 

193 for prop_name, prop_schema in json_schema["properties"].items(): 

194 properties[prop_name] = prop_schema 

195 if json_schema.get("required"): 

196 required.extend(json_schema["required"]) 

197 

198 if not properties: 

199 return {} 

200 

201 schema = {"type": "object", "properties": properties} 

202 if required: 

203 schema["required"] = required 

204 return schema 

205 

206 @staticmethod 

207 def _sanitize_name(operation_id: str) -> str: 

208 """清理operationId为合法的函数名。""" 

209 name = re.sub(r"[^a-zA-Z0-9_]", "_", operation_id) 

210 name = re.sub(r"_{2,}", "_", name) 

211 name = name.strip("_").lower() 

212 if not name[0].isalpha() and name[0] != "_": 

213 name = "tool_" + name 

214 return name[:64] 

215 

216 @staticmethod 

217 def _generate_operation_id(method: str, path: str) -> str: 

218 """无operationId时从method+path生成。""" 

219 clean = re.sub(r"[{}]", "", path).replace("/", "_").strip("_") 

220 clean = re.sub(r"[^a-zA-Z0-9_]", "_", clean) 

221 return f"{method.lower()}_{clean}" 

222 

223 async def invoke(self, tool: GeneratedTool, params: dict) -> dict: 

224 """执行生成的工具调用。""" 

225 url = tool.base_url.rstrip("/") + tool.path 

226 # 替换路径参数 

227 for key, val in params.items(): 

228 placeholder = "{" + key + "}" 

229 if placeholder in url: 

230 url = url.replace(placeholder, str(val)) 

231 params = {k: v for k, v in params.items() if k != key} 

232 

233 headers = {} 

234 if tool.auth_header: 

235 headers[tool.auth_header] = self.auth_value 

236 

237 if tool.method == "GET": 

238 resp = await self._http.get(url, params=params, headers=headers) 

239 else: 

240 headers.setdefault("Content-Type", "application/json") 

241 resp = await self._http.request(tool.method, url, json=params, headers=headers) 

242 

243 resp.raise_for_status() 

244 return resp.json() 

245 

246 async def close(self): 

247 await self._http.aclose()