_tools.py

"""OpenAPI 3.1 → MCP tool catalog."""

from __future__ import annotations

from typing import Any


def openapi_to_tools(
    spec: dict[str, Any],
    *,
    include_only: set[str] | None = None,
    exclude: set[str] | None = None,
) -> list[dict[str, Any]]:
    """Convert an OpenAPI 3.1 spec into a flat list of MCP tool definitions.

    One tool per (path, method). Tool name is ``operationId`` when present,
    otherwise ``{method}_{path}`` with non-identifier characters squashed.
    ``include_only`` / ``exclude`` filter by tool name.
    """
    tools: list[dict[str, Any]] = []
    paths = spec.get("paths", {})
    if not isinstance(paths, dict):
        return tools
    for path, ops in paths.items():
        if not isinstance(ops, dict):
            continue
        for method, op in ops.items():
            if method.lower() not in {"get", "post", "put", "patch", "delete", "head", "options"}:
                continue
            if not isinstance(op, dict):
                continue
            name = _tool_name(op.get("operationId"), method, path)
            if include_only is not None and name not in include_only:
                continue
            if exclude is not None and name in exclude:
                continue
            description = op.get("summary") or op.get("description") or f"{method.upper()} {path}"
            tools.append(
                {
                    "name": name,
                    "description": description,
                    "inputSchema": _input_schema(op),
                    "_meta": {
                        "method": method.upper(),
                        "path": path,
                        "operation_id": op.get("operationId"),
                    },
                }
            )
    return tools


def _tool_name(operation_id: Any, method: str, path: str) -> str:
    if isinstance(operation_id, str) and operation_id:
        return operation_id
    sanitized = path.replace("/", "_").replace("{", "").replace("}", "").strip("_")
    if not sanitized:
        sanitized = "root"
    return f"{method.lower()}_{sanitized}"


def _input_schema(op: dict[str, Any]) -> dict[str, Any]:
    """Synthesise a JSON Schema describing the tool input.

    Combines path parameters, query parameters, header parameters, and the
    request body (if any) into a single object schema. Path / query / header
    parameter names get prefixed namespaces to avoid collisions.
    """
    properties: dict[str, Any] = {}
    required: list[str] = []

    for param in op.get("parameters", []):
        if not isinstance(param, dict):
            continue
        loc = param.get("in")
        name = param.get("name")
        # "cookie" is deliberately excluded: exposing session cookies as tool
        # inputs trains agents to handle credentials as ordinary data.
        if not isinstance(name, str) or loc not in {"path", "query", "header"}:
            continue
        schema = param.get("schema") or {"type": "string"}
        key = f"{loc}.{name}"
        properties[key] = dict(schema)
        if "description" in param and "description" not in properties[key]:
            properties[key]["description"] = param["description"]
        if param.get("required") or loc == "path":
            required.append(key)

    body = op.get("requestBody")
    if isinstance(body, dict):
        content = body.get("content", {})
        if isinstance(content, dict):
            json_media = content.get("application/json")
            if isinstance(json_media, dict):
                body_schema = json_media.get("schema") or {"type": "object"}
                properties["body"] = body_schema
                if body.get("required"):
                    required.append("body")

    return {
        "type": "object",
        "properties": properties,
        "required": required,
    }


__all__ = ["openapi_to_tools"]