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.
81 lines
2.9 KiB
81 lines
2.9 KiB
import json
|
|
from typing import Any
|
|
|
|
class OpenAPIParser:
|
|
def __init__(self, spec: dict):
|
|
self.spec = spec
|
|
self.base_url = ""
|
|
servers = spec.get("servers", [{}])
|
|
if servers and isinstance(servers, list):
|
|
self.base_url = servers[0].get("url", "")
|
|
|
|
def parse_tools(self) -> list[dict]:
|
|
tools = []
|
|
paths = self.spec.get("paths", {})
|
|
for path, methods in paths.items():
|
|
if not isinstance(methods, dict):
|
|
continue
|
|
for method, operation in methods.items():
|
|
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:
|
|
op_id = operation.get("operationId", "")
|
|
if not op_id:
|
|
op_id = f"{method}_{path.replace('/', '_').strip('_')}"
|
|
|
|
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": {
|
|
"type": "object",
|
|
"properties": properties,
|
|
"required": required,
|
|
},
|
|
"path": path,
|
|
"method": method.upper(),
|
|
}
|
|
|
|
def _parse_parameters(self, operation: dict) -> dict[str, Any]:
|
|
props = {}
|
|
for param in operation.get("parameters", []):
|
|
if not isinstance(param, dict):
|
|
continue
|
|
pname = param.get("name", "")
|
|
if not pname:
|
|
continue
|
|
schema = param.get("schema", {})
|
|
if not isinstance(schema, dict):
|
|
schema = {}
|
|
props[pname] = {
|
|
"type": schema.get("type", "string"),
|
|
"description": param.get("description", ""),
|
|
}
|
|
if "enum" in schema:
|
|
props[pname]["enum"] = schema["enum"]
|
|
|
|
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
|