72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
import re
|
|
from typing import Any
|
|
|
|
from app.domain.schemas import ToolDefinition
|
|
|
|
|
|
def operation_name(method: str, path: str) -> str:
|
|
clean = path.strip("/").replace("{", "").replace("}", "")
|
|
parts = [part for part in re.split(r"[/_-]+", clean) if part and part != "id"]
|
|
resource = "_".join(parts)
|
|
verb = {"POST": "create", "PATCH": "update", "PUT": "replace", "GET": "get", "DELETE": "delete"}[method.upper()]
|
|
return f"{verb}_{resource}".rstrip("_")
|
|
|
|
|
|
def path_params(path: str) -> list[str]:
|
|
return re.findall(r"{([^}]+)}", path)
|
|
|
|
|
|
class OpenAPIToolGenerator:
|
|
"""Generates callable tool metadata from OpenAPI operations."""
|
|
|
|
def __init__(self, spec: dict[str, Any]) -> None:
|
|
self.spec = spec
|
|
|
|
def generate(self) -> list[ToolDefinition]:
|
|
tools: list[ToolDefinition] = []
|
|
for path, methods in self.spec.get("paths", {}).items():
|
|
for method, operation in methods.items():
|
|
upper = method.upper()
|
|
if upper not in {"POST", "PATCH", "PUT", "DELETE", "GET"}:
|
|
continue
|
|
request_schema = self._request_schema(operation)
|
|
params = path_params(path)
|
|
if not request_schema and upper in {"POST", "PATCH", "PUT"}:
|
|
continue
|
|
schema = request_schema or {"type": "object", "properties": {}}
|
|
if params:
|
|
schema = {
|
|
**schema,
|
|
"properties": {
|
|
**schema.get("properties", {}),
|
|
**{param: {"type": "string"} for param in params},
|
|
},
|
|
"required": list(dict.fromkeys([*schema.get("required", []), *params])),
|
|
}
|
|
tools.append(
|
|
ToolDefinition(
|
|
name=operation.get("operationId") or operation_name(upper, path),
|
|
description=operation.get("summary") or operation.get("description") or f"{upper} {path}",
|
|
method=upper,
|
|
path=path,
|
|
schema=schema,
|
|
required=schema.get("required", []),
|
|
path_params=params,
|
|
)
|
|
)
|
|
return tools
|
|
|
|
def _request_schema(self, operation: dict[str, Any]) -> dict[str, Any] | None:
|
|
body = operation.get("requestBody", {}).get("content", {}).get("application/json", {})
|
|
schema = body.get("schema")
|
|
if not schema:
|
|
return None
|
|
return self._resolve(schema)
|
|
|
|
def _resolve(self, schema: dict[str, Any]) -> dict[str, Any]:
|
|
ref = schema.get("$ref")
|
|
if not ref:
|
|
return schema
|
|
name = ref.split("/")[-1]
|
|
return self.spec.get("components", {}).get("schemas", {}).get(name, {})
|