From 9a3feccc9033a175c0cbbfd3750afd4403f8ed91 Mon Sep 17 00:00:00 2001 From: Samuel Amar Date: Wed, 2 Sep 2026 11:39:07 +0200 Subject: [PATCH] Add hierarchical key-scoped catalog API for machine consumers GET /catalog walks agents through progressive disclosure: available APIs with descriptions, then one API's tags, then a tag's endpoints, then full detail for a single operation with $refs resolved inline. Oversized tags (tag-poor upstreams) fall back to path-prefix groups, drillable with ?prefix= and compressed through single-child chains. Responses are filtered to the key's grants, carry ETags for cheap revalidation, and reuse the discovery spec cache. Key auth is shared with the docs portal via portal.resolve_api_key; 'catalog' and 'mcp' are now reserved slugs. Co-Authored-By: Claude Fable 5 --- README.md | 6 + app/admin/routes.py | 3 +- app/catalog.py | 331 ++++++++++++++++++++++++++++++++++++++++++++ app/main.py | 3 +- app/portal.py | 13 +- 5 files changed, 352 insertions(+), 4 deletions(-) create mode 100644 app/catalog.py diff --git a/README.md b/README.md index eabdaea..ef4f8fa 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,12 @@ with a built-in web management console. key holds a grant on, across all services, rewritten to the gateway's routes. `GET /docs` serves a Swagger UI portal around it — paste a key, browse and try exactly what that key can call. +- **Machine catalog** — a hierarchical, key-scoped catalog for agents/MCP + clients: `GET /catalog` (available APIs with descriptions), then + `/catalog/{slug}` (tags), then `/catalog/{slug}/tags/{tag}` (endpoints; + oversized tags group by path prefix, drill with `?prefix=`), then + `/catalog/{slug}/operation?method=&path=` (full schema, `$ref`s resolved + inline). Authenticated with `X-API-Key`; responses carry ETags. - **Retention** — a background purge runs every 6 hours: request/response payloads are blanked after 7 days (the log row stays inspectable), whole log rows are deleted after 90 days, and freed pages are returned to the diff --git a/app/admin/routes.py b/app/admin/routes.py index e626abd..caba076 100644 --- a/app/admin/routes.py +++ b/app/admin/routes.py @@ -13,7 +13,8 @@ router = APIRouter(prefix="/admin") templates = Jinja2Templates(directory=str(config.BASE_DIR / "app" / "templates")) # Slugs the proxy catch-all must never claim -RESERVED_SLUGS = {"admin", "static", "health", "docs", "redoc", "openapi.json"} +RESERVED_SLUGS = {"admin", "static", "health", "docs", "redoc", "openapi.json", + "catalog", "mcp"} def build_tree(endpoints: list[Endpoint]) -> dict: diff --git a/app/catalog.py b/app/catalog.py new file mode 100644 index 0000000..f465229 --- /dev/null +++ b/app/catalog.py @@ -0,0 +1,331 @@ +"""Hierarchical, key-scoped endpoint catalog for machine consumers (MCP). + +Designed for progressive disclosure so an agent is never flooded: + + GET /catalog -> the APIs this key can reach + GET /catalog/{slug} -> one API's tags (logical sections) + GET /catalog/{slug}/tags/{tag} -> endpoints of one tag; when a tag + is too large the response groups + by path prefix (?prefix= drills) + GET /catalog/{slug}/operation?method=&path= + -> full detail for one operation, + with $refs resolved inline + +Everything is authenticated with X-API-Key and filtered to the key's grants. +Responses carry ETags so a polling MCP can revalidate cheaply. +""" +import hashlib +import json +from urllib.parse import quote, urlencode + +from fastapi import APIRouter, Depends, Request, Response +from fastapi.responses import JSONResponse +from sqlalchemy.orm import Session + +from app import discovery +from app.database import get_db +from app.models import ApiKey, Endpoint, Service +from app.portal import _HTTP_METHODS, _grant_matches, resolve_api_key +from app.proxy import _error + +router = APIRouter() + +MAX_LISTING = 50 # above this, a tag listing groups by path prefix +MAX_REF_DEPTH = 10 + + +def _respond(request: Request, payload: dict) -> Response: + body = json.dumps(payload, ensure_ascii=False).encode() + etag = '"' + hashlib.md5(body).hexdigest() + '"' + if request.headers.get("if-none-match") == etag: + return Response(status_code=304, headers={"ETag": etag}) + return Response(body, media_type="application/json", + headers={"ETag": etag, "Cache-Control": "private, must-revalidate"}) + + +def _grants_by_service(api_key: ApiKey) -> dict[int, list[Endpoint]]: + grouped: dict[int, list[Endpoint]] = {} + for grant in api_key.endpoints: + if grant.service.is_active: + grouped.setdefault(grant.service_id, []).append(grant) + return grouped + + +def _granted_operations(db: Session, service: Service, + grants: list[Endpoint]) -> tuple[list[dict], dict | None]: + """The operations this key may call on one service, as + [{method, path (gateway form), tags, summary, operationId}]. Falls back to + the endpoint catalog when the upstream has no OpenAPI document.""" + spec = discovery.get_spec(db, service) + ops: list[dict] = [] + if spec: + for path, operations in spec.get("paths", {}).items(): + if not isinstance(operations, dict): + continue + for method, op in operations.items(): + if method.lower() not in _HTTP_METHODS or not isinstance(op, dict): + continue + if any(_grant_matches(g, method.upper(), path) for g in grants): + ops.append({ + "method": method.upper(), + "path": f"/{service.slug}{path}", + "tags": list(op.get("tags") or ["General"]), + "summary": op.get("summary") or op.get("description", "")[:200], + "operationId": op.get("operationId"), + }) + return ops, spec + + for g in grants: + ops.append({ + "method": "ANY" if g.method == "*" else g.method, + "path": f"/{service.slug}{g.path}", + "tags": ["General"], + "summary": g.description or "", + "operationId": None, + }) + return ops, None + + +def _service_summary(service: Service, spec: dict | None) -> dict: + info = (spec or {}).get("info") or {} + return { + "slug": service.slug, + "name": service.name, + "title": info.get("title", ""), + # The admin-entered description overrides the upstream's own. + "description": service.description or info.get("description", ""), + "route_prefix": f"/{service.slug}", + } + + +def _tag_summaries(ops: list[dict], spec: dict | None, slug: str) -> list[dict]: + order: list[str] = [] + counts: dict[str, int] = {} + for op in ops: + for tag in op["tags"]: + if tag not in counts: + order.append(tag) + counts[tag] = counts.get(tag, 0) + 1 + descriptions = {} + for tag in (spec or {}).get("tags") or []: + if isinstance(tag, dict) and tag.get("name"): + descriptions[tag["name"]] = tag.get("description", "") + return [{"name": name, + "description": descriptions.get(name, ""), + "endpoints": counts[name], + "url": f"/catalog/{slug}/tags/{quote(name, safe='')}"} + for name in order] + + +def _group_by_prefix(ops: list[dict], prefix: str) -> tuple[list[dict], list[dict]]: + """Bucket operations by the next path segment after `prefix`, compressing + single-child chains (/api + /v1 -> /api/v1). Returns (groups, operations + that sit exactly at the prefix).""" + at_prefix: list[dict] = [] + buckets: dict[str, list[dict]] = {} + base = prefix.rstrip("/") + for op in ops: + rest = op["path"][len(base):].lstrip("/") + if not rest: + at_prefix.append(op) + continue + buckets.setdefault(rest.split("/")[0], []).append(op) + + groups = [] + for segment, items in sorted(buckets.items()): + group_prefix = f"{base}/{segment}" + while True: # chain compression + rests = {op["path"][len(group_prefix):].lstrip("/").split("/")[0] + for op in items} + rests.discard("") + if len(rests) == 1 and all(len(op["path"]) > len(group_prefix) for op in items): + group_prefix = f"{group_prefix}/{rests.pop()}" + else: + break + groups.append({"prefix": group_prefix, "endpoints": len(items)}) + return groups, at_prefix + + +def _endpoint_entry(op: dict, slug: str) -> dict: + detail = urlencode({"method": op["method"], "path": op["path"]}) + return {"method": op["method"], "path": op["path"], "summary": op["summary"], + **({"operationId": op["operationId"]} if op["operationId"] else {}), + "url": f"/catalog/{slug}/operation?{detail}"} + + +def _resolve_refs(node, spec: dict, depth: int = 0, seen: frozenset = frozenset()): + """Inline every local $ref so a consumer never has to chase references. + Cycles and excessive depth degrade to a named placeholder.""" + if depth > MAX_REF_DEPTH: + return {"$truncated": True} + if isinstance(node, dict): + ref = node.get("$ref") + if isinstance(ref, str) and ref.startswith("#/"): + if ref in seen: + return {"$circular": ref.rsplit("/", 1)[-1]} + target = spec + for part in ref[2:].split("/"): + target = target.get(part) if isinstance(target, dict) else None + if target is None: + return node # dangling ref: leave as-is + resolved = _resolve_refs(target, spec, depth + 1, seen | {ref}) + if isinstance(resolved, dict): + extras = {k: v for k, v in node.items() if k != "$ref"} + return {**resolved, **extras} + return resolved + return {k: _resolve_refs(v, spec, depth + 1, seen) for k, v in node.items()} + if isinstance(node, list): + return [_resolve_refs(item, spec, depth + 1, seen) for item in node] + return node + + +def _find_service(db: Session, api_key: ApiKey, slug: str): + grouped = _grants_by_service(api_key) + for service_id, grants in grouped.items(): + service = grants[0].service + if service.slug == slug: + return service, grants + return None, None + + +# ---------- level 1: available APIs ---------- + +@router.get("/catalog", include_in_schema=False) +def catalog_root(request: Request, db: Session = Depends(get_db)): + api_key = resolve_api_key(request, db) + if isinstance(api_key, JSONResponse): + return api_key + + apis = [] + for grants in sorted(_grants_by_service(api_key).values(), + key=lambda g: g[0].service.name): + service = grants[0].service + ops, spec = _granted_operations(db, service, grants) + if not ops: + continue + tags = _tag_summaries(ops, spec, service.slug) + apis.append({**_service_summary(service, spec), + "tags": len(tags), "endpoints": len(ops), + "url": f"/catalog/{service.slug}"}) + return _respond(request, { + "key": api_key.name, + "apis": apis, + "hint": "GET the api url for its tags, then a tag url for its endpoints, " + "then an endpoint url for the full schema. Call endpoints through " + "this gateway with the X-API-Key header.", + }) + + +# ---------- level 2: one API's tags ---------- + +@router.get("/catalog/{slug}", include_in_schema=False) +def catalog_service(slug: str, request: Request, db: Session = Depends(get_db)): + api_key = resolve_api_key(request, db) + if isinstance(api_key, JSONResponse): + return api_key + service, grants = _find_service(db, api_key, slug) + if service is None: + return _error(404, "unknown_api", + f"No API '{slug}' is available to this key. GET /catalog for the list.") + + ops, spec = _granted_operations(db, service, grants) + return _respond(request, { + **_service_summary(service, spec), + "endpoints": len(ops), + "tags": _tag_summaries(ops, spec, slug), + }) + + +# ---------- level 3: one tag's endpoints ---------- + +@router.get("/catalog/{slug}/tags/{tag}", include_in_schema=False) +def catalog_tag(slug: str, tag: str, request: Request, + prefix: str | None = None, db: Session = Depends(get_db)): + api_key = resolve_api_key(request, db) + if isinstance(api_key, JSONResponse): + return api_key + service, grants = _find_service(db, api_key, slug) + if service is None: + return _error(404, "unknown_api", + f"No API '{slug}' is available to this key. GET /catalog for the list.") + + ops, _ = _granted_operations(db, service, grants) + ops = [op for op in ops if tag in op["tags"]] + if not ops: + return _error(404, "unknown_tag", + f"No tag '{tag}' on '{slug}'. GET /catalog/{slug} for its tags.") + if prefix: + ops = [op for op in ops if op["path"].startswith(prefix)] + + payload = {"api": slug, "tag": tag, "total": len(ops)} + if prefix: + payload["prefix"] = prefix + if len(ops) > MAX_LISTING: + groups, at_prefix = _group_by_prefix(ops, prefix or f"/{slug}") + # A single group would force a pointless extra round trip — split it. + while len(groups) == 1 and not at_prefix: + groups, at_prefix = _group_by_prefix(ops, groups[0]["prefix"]) + payload["endpoints"] = [_endpoint_entry(op, slug) for op in at_prefix] + payload["groups"] = [{**g, "url": f"/catalog/{slug}/tags/{quote(tag, safe='')}" + f"?{urlencode({'prefix': g['prefix']})}"} + for g in groups] + payload["hint"] = (f"{len(ops)} endpoints — too many to list. " + "GET a group url to drill into that path prefix.") + else: + payload["endpoints"] = [_endpoint_entry(op, slug) for op in ops] + return _respond(request, payload) + + +# ---------- level 4: one operation, full detail ---------- + +@router.get("/catalog/{slug}/operation", include_in_schema=False) +def catalog_operation(slug: str, method: str, path: str, request: Request, + db: Session = Depends(get_db)): + api_key = resolve_api_key(request, db) + if isinstance(api_key, JSONResponse): + return api_key + service, grants = _find_service(db, api_key, slug) + if service is None: + return _error(404, "unknown_api", + f"No API '{slug}' is available to this key. GET /catalog for the list.") + + # Accept the gateway form (/slug/...) as listed, or the upstream path. + upstream_path = path[len(f"/{slug}"):] if path.startswith(f"/{slug}/") else path + method = method.upper() + + spec = discovery.get_spec(db, service) + op = None + if spec: + operations = spec.get("paths", {}).get(upstream_path) + if isinstance(operations, dict): + candidate = operations.get(method.lower()) + if isinstance(candidate, dict): + op = candidate + + if not any(_grant_matches(g, method, upstream_path) for g in grants): + return _error(404, "unknown_operation", + f"{method} {path} is not available to this key on '{slug}'.") + + gateway_path = f"/{slug}{upstream_path}" + detail = { + "method": method, + "path": gateway_path, + "api": slug, + "auth": "Send the API key in the X-API-Key header.", + } + if op: + for field in ("operationId", "summary", "description", "tags", + "parameters", "requestBody", "responses", "deprecated"): + if field in op: + detail[field] = _resolve_refs(op[field], spec) + # path-level parameters apply to every operation of the path + shared = spec["paths"][upstream_path].get("parameters") + if shared: + detail.setdefault("parameters", []) + detail["parameters"] = _resolve_refs(shared, spec) + detail["parameters"] + else: + grant = next(g for g in grants if _grant_matches(g, method, upstream_path)) + detail["summary"] = grant.description or "" + detail["note"] = ("The upstream publishes no OpenAPI document; " + "no schema information is available.") + return _respond(request, detail) diff --git a/app/main.py b/app/main.py index 818285e..55178ae 100644 --- a/app/main.py +++ b/app/main.py @@ -6,7 +6,7 @@ from fastapi.staticfiles import StaticFiles from fastapi.responses import RedirectResponse from sqlalchemy import text -from app import config, portal, proxy, retention, security +from app import catalog, config, portal, proxy, retention, security from app.admin import routes as admin_routes from app.admin import stats as admin_stats from app.admin.deps import LoginRequired, login_redirect_handler @@ -131,6 +131,7 @@ def health(): app.include_router(admin_routes.router) app.include_router(admin_stats.router) app.include_router(portal.router) +app.include_router(catalog.router) app.mount("/static", StaticFiles(directory=str(config.BASE_DIR / "app" / "static")), name="static") # The proxy catch-all (/{slug}/...) must come last so it never shadows # /admin, /static, /docs or /health. diff --git a/app/portal.py b/app/portal.py index 161fdd1..49aa87d 100644 --- a/app/portal.py +++ b/app/portal.py @@ -183,8 +183,9 @@ def build_key_spec(db: Session, api_key: ApiKey, gateway_url: str) -> dict: } -@router.get("/openapi.json", include_in_schema=False) -def key_scoped_openapi(request: Request, db: Session = Depends(get_db)): +def resolve_api_key(request: Request, db: Session) -> ApiKey | JSONResponse: + """Authenticate a consumer-facing request by X-API-Key. Returns the key, + or the error response to send back.""" plain_key = request.headers.get(config.API_KEY_HEADER) if not plain_key: return _error(401, "missing_api_key", @@ -194,6 +195,14 @@ def key_scoped_openapi(request: Request, db: Session = Depends(get_db)): .one_or_none()) if api_key is None or not api_key.is_active or not api_key.user.is_active: return _error(403, "invalid_api_key", "API key is unknown or has been revoked.") + return api_key + + +@router.get("/openapi.json", include_in_schema=False) +def key_scoped_openapi(request: Request, db: Session = Depends(get_db)): + api_key = resolve_api_key(request, db) + if isinstance(api_key, JSONResponse): + return api_key spec = build_key_spec(db, api_key, str(request.base_url).rstrip("/")) return JSONResponse(spec, headers={"Cache-Control": "no-store"})