kong-k8s-platform/tools/generate-mcp-tools.py

#!/usr/bin/env -S uv run --quiet --with pyyaml --script
"""Generate a domain's ai-mcp-proxy KongPlugin from its OpenAPI spec.

    ./generate-mcp-tools.py ../../catalog-api          # write
    ./generate-mcp-tools.py ../../catalog-api --check  # diff only, exit 1 if stale

`deck file openapi2mcp` does the conversion. This wraps it and re-injects the
five things it cannot produce. Every step below exists because of a specific
gap in Kong's generator, documented with its evidence in MCP-TOOL-GENERATION.md
— when a Kong release closes one, delete that step and nothing else changes.

Inputs are the domain's `openapi.yaml` and `kong/mcp/tools-policy.yaml`. The
policy is the allow-list *and* the ACL: an operation becomes a tool only if the
policy names it. That is deliberate — the specs are extracted from the live
backend, so anything expressed in the spec (`x-kong-mcp-exclude`) would be
overwritten at the next extraction, and a spec that grows silently must not
grow the MCP surface with it.

Requires `deck` on PATH. Run it from anywhere; paths are resolved from the
domain directory you pass.
"""

from __future__ import annotations

import argparse
import difflib
import json
import subprocess
import sys
from pathlib import Path
from typing import Any

import yaml

PLUGIN_CONFIG = {"mode": "conversion-listener", "include_consumer_groups": True}

# Gap 4: `servers` in the spec is the gateway's own public URL — it has to be,
# the spec is also the developer portal's documentation. Kong takes it as the
# upstream target, which would make the MCP proxy loop back into the gateway.
UPSTREAM_HOST = "${REST_HOSTNAME}"
UPSTREAM_SCHEME = "http"

# Gap 5: DRF declares json, form-urlencoded and multipart for every write, with
# the identical schema. Kong copies all three, tripling the definition the model
# has to read on every turn, for a body the plugin sends as JSON.
KEEP_CONTENT_TYPE = "application/json"


class Stale(Exception):
    """--check found a difference."""


def run_deck(spec: Path) -> list[dict[str, Any]]:
    out = subprocess.run(
        ["deck", "file", "openapi2mcp", "-s", str(spec), "--no-id", "-o", "-", "--format", "json"],
        capture_output=True,
        text=True,
        check=True,
    ).stdout
    doc = json.loads(out)
    for service in doc.get("services", []):
        for route in service.get("routes", []):
            for plugin in route.get("plugins", []):
                if plugin.get("name") == "ai-mcp-proxy":
                    return plugin["config"]["tools"]
    raise SystemExit("deck produced no ai-mcp-proxy plugin — check the spec's paths")


def annotations_for(method: str, title: str) -> dict[str, Any]:
    """Gap 2: Kong emits only `title`.

    The other four are in the MCP specification and follow mechanically from the
    method. They are not decoration — a client reads `destructive_hint` to decide
    whether to ask its user before calling.
    """
    m = method.upper()
    return {
        "title": title,
        "read_only_hint": m == "GET",
        "idempotent_hint": m in {"GET", "PUT", "DELETE"},
        "destructive_hint": m == "DELETE",
        "open_world_hint": False,
    }


def responses_for(spec: dict[str, Any], method: str, path: str) -> dict[str, Any] | None:
    """Gap 3: the plugin validates the upstream response against this and maps it
    to structuredOutput. Kong's generator never emits it, so every tool returns
    raw JSON today and a backend that changes shape is caught by nobody.

    The plugin accepts one non-error response and does not resolve `$ref`, so
    take the lowest 2xx and inline it.
    """
    op = (spec.get("paths", {}).get(path, {}) or {}).get(method.lower())
    if not op:
        return None
    ok = sorted(c for c in (op.get("responses") or {}) if c.isdigit() and int(c) < 400)
    if not ok:
        return None
    body = as_json_schema(resolve_refs(op["responses"][ok[0]], spec))
    content = body.get("content")
    if not content:
        return None
    kept = {k: v for k, v in content.items() if k == KEEP_CONTENT_TYPE}
    return {ok[0]: {"content": kept}} if kept else None


def as_json_schema(node: Any) -> Any:
    """Translate OpenAPI 3.0's `nullable: true` into JSON Schema's null type.

    The plugin's fields are documented as "OpenAPI format", and it validates them
    with a JSON Schema validator (`jsonschema_rs`). `nullable` is not a JSON
    Schema keyword, so it is ignored and `type: string` is enforced — meaning any
    3.0 spec with a nullable field produces a tool that rejects the backend's own
    valid response.

    Seen for real: `list-products` returns `previous: null` on the first page, and
    the call failed with *null is not of type "string", /previous* — after the
    backend had already done the work, and only visible in the agent's log.

    A 3.1 spec expresses this as `type: [x, "null"]` and needs none of this.
    """
    if isinstance(node, dict):
        out = {k: as_json_schema(v) for k, v in node.items() if k != "nullable"}
        if node.get("nullable") is True and "type" in out:
            t = out["type"]
            out["type"] = [*t, "null"] if isinstance(t, list) else [t, "null"]
        return out
    if isinstance(node, list):
        return [as_json_schema(v) for v in node]
    return node


def resolve_refs(node: Any, spec: dict[str, Any], seen: frozenset[str] = frozenset()) -> Any:
    """Inline `$ref`s. The plugin's schema states it does not resolve them.

    Circular references are cut rather than followed — a self-referencing schema
    would otherwise expand forever, and no spec in this estate has one.
    """
    if isinstance(node, dict):
        ref = node.get("$ref")
        if ref:
            if ref in seen:
                return {"type": "object"}
            target = spec
            for part in ref.lstrip("#/").split("/"):
                target = target[part]
            return resolve_refs(target, spec, seen | {ref})
        return {k: resolve_refs(v, spec, seen) for k, v in node.items()}
    if isinstance(node, list):
        return [resolve_refs(v, spec, seen) for v in node]
    return node


def build(domain: Path) -> str:
    spec_path = domain / "openapi.yaml"
    policy_path = domain / "kong" / "mcp" / "tools-policy.yaml"
    for p in (spec_path, policy_path):
        if not p.exists():
            raise SystemExit(f"missing {p}")

    spec = yaml.safe_load(spec_path.read_text())
    policy = yaml.safe_load(policy_path.read_text())
    wanted: dict[str, Any] = policy["tools"]

    generated = {t["name"]: t for t in run_deck(spec_path)}

    # Loud rather than silent: a policy key that matches nothing means Kong's
    # naming changed or an operation left the spec. Either way the ACL it
    # carried is no longer applied, and that must not pass quietly.
    unknown = sorted(set(wanted) - set(generated))
    if unknown:
        raise SystemExit(
            f"{domain.name}: policy names tools that the spec does not produce: {unknown}\n"
            f"  produced: {sorted(generated)}"
        )

    tools = []
    for name in wanted:
        tool = dict(generated[name])
        method, path = tool["method"], tool["path"]

        tool["annotations"] = annotations_for(method, tool.get("annotations", {}).get("title", name))

        # Gap 1: Kong derives per-tool ACLs only from an oauth2 scheme carrying
        # x-kong-mcp-acl, keyed on OAuth scopes. Our authorization is realm roles
        # mapped to consumer groups. Kong produces nothing, warns about nothing,
        # and exits 0 — a regeneration without this line publishes every write
        # operation to every caller.
        tool["acl"] = {"allow": list(wanted[name]["allow"])}

        tool["scheme"] = UPSTREAM_SCHEME
        tool["host"] = UPSTREAM_HOST

        if body := tool.get("request_body"):
            kept = {
                k: as_json_schema(v)
                for k, v in body["content"].items()
                if k == KEEP_CONTENT_TYPE
            }
            if not kept:
                raise SystemExit(f"{name}: no {KEEP_CONTENT_TYPE} request body in the spec")
            body["content"] = kept

        if responses := responses_for(spec, method, path):
            tool["responses"] = responses

        tools.append(tool)

    plugin = {
        "apiVersion": "configuration.konghq.com/v1",
        "kind": "KongPlugin",
        "metadata": {
            "name": "${NAME}-" + policy["mcp_name"] + "-ai-mcp-proxy",
            "namespace": "${NAMESPACE}",
        },
        "plugin": "ai-mcp-proxy",
        "config": {**PLUGIN_CONFIG, "tools": tools},
    }

    header = (
        "# GENERATED — do not edit. Edit openapi.yaml or tools-policy.yaml, then:\n"
        "#   k8s-platform/tools/generate-mcp-tools.py <this domain>\n"
        "#\n"
        "# Produced by `deck file openapi2mcp`, then post-processed to add the ACLs,\n"
        "# the MCP behavioural annotations, the response schemas, the real upstream\n"
        "# host and the JSON-only bodies that Kong's generator does not emit. Each of\n"
        "# those is a documented gap — see k8s-platform/MCP-TOOL-GENERATION.md.\n"
    )
    return header + yaml.safe_dump(plugin, sort_keys=False, width=100, allow_unicode=True)


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("domain", type=Path, help="a <domain>-api directory")
    ap.add_argument("--check", action="store_true", help="diff only; exit 1 if the file is stale")
    args = ap.parse_args()

    domain = args.domain.resolve()
    target = domain / "kong" / "mcp" / "ai-mcp-proxy.yaml"
    new = build(domain)
    old = target.read_text() if target.exists() else ""

    if new == old:
        print(f"{domain.name}: up to date")
        return 0
    if args.check:
        sys.stdout.writelines(
            difflib.unified_diff(
                old.splitlines(keepends=True), new.splitlines(keepends=True),
                fromfile=str(target), tofile="generated",
            )
        )
        print(f"{domain.name}: STALE", file=sys.stderr)
        return 1

    target.write_text(new)
    print(f"{domain.name}: written ({new.count(chr(10))} lines)")
    return 0


if __name__ == "__main__":
    sys.exit(main())