You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
132 lines
5.1 KiB
132 lines
5.1 KiB
"""OpenAPI 规范解析器。
|
|
|
|
提供从 OpenAPI/Swagger 规范文档中自动解析 API 端点并转换为自定义工具定义的功能。
|
|
"""
|
|
import json
|
|
from typing import Any
|
|
|
|
|
|
class OpenAPIParser:
|
|
"""OpenAPI 规范解析器类。
|
|
|
|
解析 OpenAPI 3.0 规范文档，提取其中的 API 端点信息并转换为自定义工具定义。
|
|
|
|
Attributes:
|
|
spec: OpenAPI 规范文档的字典表示。
|
|
base_url: API 服务的基础 URL。
|
|
"""
|
|
|
|
def __init__(self, spec: dict):
|
|
"""初始化 OpenAPI 解析器。
|
|
|
|
Args:
|
|
spec: OpenAPI 规范文档的字典表示，包含 servers、paths 等字段。
|
|
"""
|
|
self.spec = spec # OpenAPI 规范文档内容
|
|
self.base_url = "" # API 基础 URL
|
|
servers = spec.get("servers", [{}])
|
|
if servers and isinstance(servers, list):
|
|
self.base_url = servers[0].get("url", "") # 获取第一个服务器 URL 作为基础地址
|
|
|
|
def parse_tools(self) -> list[dict]:
|
|
"""解析 OpenAPI 规范中的所有 API 端点。
|
|
|
|
遍历 paths 中的所有 HTTP 方法，将每个端点转换为工具定义。
|
|
|
|
Returns:
|
|
list[dict]: 工具定义列表，每个工具包含 name、description、parameters、path、method 等信息。
|
|
"""
|
|
tools = []
|
|
paths = self.spec.get("paths", {}) # 获取所有 API 路径
|
|
for path, methods in paths.items():
|
|
if not isinstance(methods, dict):
|
|
continue
|
|
for method, operation in methods.items():
|
|
# 只处理标准的 HTTP 方法
|
|
if method in ("get", "post", "put", "delete", "patch") and isinstance(operation, dict):
|
|
tool = self._parse_endpoint(path, method, operation)
|
|
if tool:
|
|
tools.append(tool)
|
|
return tools
|
|
|
|
def _parse_endpoint(self, path: str, method: str, operation: dict) -> dict | None:
|
|
"""解析单个 API 端点的详细信息。
|
|
|
|
Args:
|
|
path: API 路径，如 "/users/{id}"。
|
|
method: HTTP 方法，如 "get"、"post" 等。
|
|
operation: 端点的操作定义，包含 operationId、summary、parameters 等。
|
|
|
|
Returns:
|
|
dict | None: 工具定义字典，包含名称、描述、参数等信息；如果解析失败返回 None。
|
|
"""
|
|
# 生成工具名称：优先使用 operationId，否则从路径生成
|
|
op_id = operation.get("operationId", "")
|
|
if not op_id:
|
|
op_id = f"{method}_{path.replace('/', '_').strip('_')}"
|
|
|
|
# 生成工具描述：优先使用 summary，其次 description，最后使用方法和路径
|
|
description = operation.get("summary") or operation.get("description") or f"{method.upper()} {path}"
|
|
properties = self._parse_parameters(operation) # 解析参数
|
|
required = []
|
|
for param in operation.get("parameters", []):
|
|
if isinstance(param, dict) and param.get("required"):
|
|
required.append(param["name"]) # 收集必填参数名
|
|
|
|
return {
|
|
"name": op_id, # 工具名称
|
|
"description": description, # 工具描述
|
|
"parameters": { # 参数 Schema
|
|
"type": "object",
|
|
"properties": properties,
|
|
"required": required,
|
|
},
|
|
"path": path, # API 路径
|
|
"method": method.upper(), # HTTP 方法（大写）
|
|
}
|
|
|
|
def _parse_parameters(self, operation: dict) -> dict[str, Any]:
|
|
"""解析 API 端点的参数定义。
|
|
|
|
包括查询参数、路径参数、请求头参数和请求体参数。
|
|
|
|
Args:
|
|
operation: 端点的操作定义。
|
|
|
|
Returns:
|
|
dict[str, Any]: 参数属性字典，键为参数名，值为参数类型和描述。
|
|
"""
|
|
props = {}
|
|
# 解析 query/path/header 参数
|
|
for param in operation.get("parameters", []):
|
|
if not isinstance(param, dict):
|
|
continue
|
|
pname = param.get("name", "")
|
|
if not pname:
|
|
continue
|
|
schema = param.get("schema", {}) # 参数的 Schema 定义
|
|
if not isinstance(schema, dict):
|
|
schema = {}
|
|
props[pname] = {
|
|
"type": schema.get("type", "string"), # 参数类型，默认为 string
|
|
"description": param.get("description", ""), # 参数描述
|
|
}
|
|
if "enum" in schema: # 如果有限定值列表
|
|
props[pname]["enum"] = schema["enum"]
|
|
|
|
# 解析请求体（requestBody）中的 JSON Schema 属性
|
|
body = (
|
|
operation.get("requestBody", {})
|
|
.get("content", {})
|
|
.get("application/json", {})
|
|
.get("schema", {})
|
|
)
|
|
if isinstance(body, dict):
|
|
for name, prop in body.get("properties", {}).items():
|
|
if isinstance(prop, dict):
|
|
props[name] = {
|
|
"type": prop.get("type", "string"),
|
|
"description": prop.get("description", ""),
|
|
}
|
|
|
|
return props
|