Add local MCP server for public REST reads

.gitignore

@@ -68,6 +68,7 @@ standards_cache.sqlite
 *.md
 !AGENTS.md
 !docs/faq.md
+!docs/api/mcp.md
 !docs/Mid_eval_blog_gsoc2026/module_B_mideval_blog.md
 
 ### Dev DBDumps

README.md

@@ -288,6 +288,8 @@ http://127.0.0.1:5000
 
 See [the myOpenCRE user guide](docs/my-opencre-user-guide.md) on using the OpenCRE API to, for example, add your own security guidelines and standards.
 
+For a local stdio MCP server over public REST reads, see [docs/api/mcp.md](docs/api/mcp.md).
+
 ## Docker building and running
 
 You can build the production or the development docker images with:

application/mcp/__init__.py

@@ -0,0 +1,5 @@
+"""Local stdio MCP server exposing public OpenCRE REST reads (issue #1003 v1)."""
+
+from application.mcp.catalog import PUBLIC_TOOLS, get_tool, list_tool_names
+
+__all__ = ["PUBLIC_TOOLS", "get_tool", "list_tool_names"]

application/mcp/__main__.py

@@ -0,0 +1,22 @@
+"""Entry point: python -m application.mcp"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import sys
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        stream=sys.stderr,
+        format="%(levelname)s %(name)s: %(message)s",
+    )
+    from application.mcp.server import run_stdio
+
+    asyncio.run(run_stdio())
+
+
+if __name__ == "__main__":
+    main()

application/mcp/catalog.py

@@ -0,0 +1,96 @@
+"""Static security allowlist for OpenCRE MCP v1 public-read tools.
+
+Exposure is determined only by this allowlist — never by enumerating OpenAPI.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, List, Tuple
+
+
+@dataclass(frozen=True)
+class ToolSpec:
+    """One MCP tool bound to a fixed public REST GET operation."""
+
+    name: str
+    method: str
+    path_template: str
+    summary: str
+
+    @property
+    def openapi_identity(self) -> Tuple[str, str]:
+        """(HTTP method lowercased, OpenAPI path) used to resolve schemas."""
+        return (self.method.lower(), self.path_template)
+
+
+# Exact v1 surface approved for the first MCP PR (public JSON reads only).
+PUBLIC_TOOLS: Tuple[ToolSpec, ...] = (
+    ToolSpec(
+        name="get_cre_by_id",
+        method="GET",
+        path_template="/rest/v1/id/{creid}",
+        summary="Get a CRE by ID",
+    ),
+    ToolSpec(
+        name="get_cre_by_name",
+        method="GET",
+        path_template="/rest/v1/name/{crename}",
+        summary="Get a CRE by name",
+    ),
+    ToolSpec(
+        name="get_node",
+        method="GET",
+        path_template="/rest/v1/{ntype}/{name}",
+        summary="Get nodes by type and name",
+    ),
+    ToolSpec(
+        name="get_documents_by_tag",
+        method="GET",
+        path_template="/rest/v1/tags",
+        summary="Get documents by tag",
+    ),
+    ToolSpec(
+        name="text_search",
+        method="GET",
+        path_template="/rest/v1/text_search",
+        summary="Text search",
+    ),
+    ToolSpec(
+        name="list_root_cres",
+        method="GET",
+        path_template="/rest/v1/root_cres",
+        summary="Get root CREs",
+    ),
+    ToolSpec(
+        name="list_all_cres",
+        method="GET",
+        path_template="/rest/v1/all_cres",
+        summary="List all CREs (paginated)",
+    ),
+    ToolSpec(
+        name="list_standards",
+        method="GET",
+        path_template="/rest/v1/standards",
+        summary="List standards",
+    ),
+    ToolSpec(
+        name="list_ga_standards",
+        method="GET",
+        path_template="/rest/v1/ga_standards",
+        summary="Standards eligible for gap analysis",
+    ),
+)
+
+_TOOLS_BY_NAME: Dict[str, ToolSpec] = {tool.name: tool for tool in PUBLIC_TOOLS}
+
+
+def list_tool_names() -> List[str]:
+    return [tool.name for tool in PUBLIC_TOOLS]
+
+
+def get_tool(name: str) -> ToolSpec:
+    try:
+        return _TOOLS_BY_NAME[name]
+    except KeyError as exc:
+        raise KeyError(f"Unknown MCP tool: {name}") from exc

application/mcp/openapi_loader.py

@@ -0,0 +1,165 @@
+"""Load MCP tool input schemas from the committed OpenAPI document.
+
+OpenAPI is the source of truth for selected allowlisted operations' parameters.
+This module does not enumerate OpenAPI to decide which tools exist.
+"""
+
+from __future__ import annotations
+
+import copy
+import os
+from functools import lru_cache
+from typing import Any, Dict, List, Optional, Set
+
+import yaml
+
+from application.mcp.catalog import PUBLIC_TOOLS, ToolSpec, get_tool
+
+# Response-format switchers that can leave JSON; omit from MCP v1 inputs.
+_JSON_UNSAFE_QUERY_PARAMS: Set[str] = {"format"}
+
+REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+DEFAULT_OPENAPI_PATH = os.path.join(REPO_ROOT, "docs", "api", "openapi.yaml")
+
+
+class OpenAPILookupError(LookupError):
+    """Raised when an allowlisted method/path is missing or unusable in OpenAPI."""
+
+
+def _load_spec(path: str) -> Dict[str, Any]:
+    with open(path, "r", encoding="utf-8") as handle:
+        data = yaml.safe_load(handle)
+    if not isinstance(data, dict) or "paths" not in data:
+        raise OpenAPILookupError(f"Invalid OpenAPI document: {path}")
+    return data
+
+
+def _resolve_ref(
+    spec: Dict[str, Any],
+    node: Any,
+    *,
+    _ref_stack: Optional[frozenset[str]] = None,
+) -> Any:
+    """Resolve local #/components/... refs; reject cycles on the current ref chain."""
+    if not isinstance(node, dict):
+        return node
+    stack = _ref_stack or frozenset()
+    if "$ref" not in node:
+        # Sibling branches share the parent chain only — not each other's refs.
+        return {
+            key: _resolve_ref(spec, value, _ref_stack=stack)
+            for key, value in node.items()
+        }
+    ref = node["$ref"]
+    if not isinstance(ref, str) or not ref.startswith("#/"):
+        raise OpenAPILookupError(f"Unsupported OpenAPI $ref: {ref}")
+    if ref in stack:
+        raise OpenAPILookupError(f"Circular OpenAPI $ref: {ref}")
+    current: Any = spec
+    for part in ref[2:].split("/"):
+        if not isinstance(current, dict) or part not in current:
+            raise OpenAPILookupError(f"Unresolved OpenAPI $ref: {ref}")
+        current = current[part]
+    # Merge sibling keywords onto the resolved target (OpenAPI 3.0.3).
+    resolved = copy.deepcopy(current)
+    next_stack = stack | {ref}
+    if isinstance(resolved, dict):
+        for key, value in node.items():
+            if key == "$ref":
+                continue
+            resolved[key] = value
+        return _resolve_ref(spec, resolved, _ref_stack=next_stack)
+    return resolved
+
+
+def _parameter_to_json_schema(
+    spec: Dict[str, Any], param: Dict[str, Any]
+) -> Dict[str, Any]:
+    schema = param.get("schema")
+    if schema is None and "content" in param:
+        raise OpenAPILookupError(
+            f"Unsupported content-based parameter '{param.get('name')}'"
+        )
+    if schema is None:
+        schema = {"type": "string"}
+    resolved = _resolve_ref(spec, schema)
+    if not isinstance(resolved, dict):
+        raise OpenAPILookupError(
+            f"Parameter '{param.get('name')}' schema must be an object"
+        )
+    out = copy.deepcopy(resolved)
+    description = param.get("description")
+    if description and "description" not in out:
+        out["description"] = description
+    return out
+
+
+def operation_input_schema(
+    tool: ToolSpec, *, spec: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
+    """Build a JSON Schema object for MCP tool arguments from OpenAPI parameters."""
+    document = spec if spec is not None else load_openapi_spec()
+    method, path = tool.openapi_identity
+    paths = document.get("paths") or {}
+    path_item = paths.get(path)
+    if not isinstance(path_item, dict):
+        raise OpenAPILookupError(
+            f"Allowlisted OpenAPI path missing: {method.upper()} {path}"
+        )
+    operation = path_item.get(method)
+    if not isinstance(operation, dict):
+        raise OpenAPILookupError(
+            f"Allowlisted OpenAPI operation missing: {method.upper()} {path}"
+        )
+
+    properties: Dict[str, Any] = {}
+    required: List[str] = []
+    for raw_param in operation.get("parameters") or []:
+        param = _resolve_ref(document, raw_param)
+        if not isinstance(param, dict):
+            raise OpenAPILookupError(f"Invalid parameter on {method.upper()} {path}")
+        location = param.get("in")
+        name = param.get("name")
+        if not name or location not in ("path", "query"):
+            continue
+        if location == "query" and name in _JSON_UNSAFE_QUERY_PARAMS:
+            continue
+        properties[name] = _parameter_to_json_schema(document, param)
+        if param.get("required"):
+            required.append(name)
+
+    schema: Dict[str, Any] = {
+        "type": "object",
+        "properties": properties,
+        "additionalProperties": False,
+    }
+    if required:
+        schema["required"] = required
+    return schema
+
+
+@lru_cache(maxsize=1)
+def load_openapi_spec(path: str = DEFAULT_OPENAPI_PATH) -> Dict[str, Any]:
+    return _load_spec(path)
+
+
+def clear_openapi_cache() -> None:
+    load_openapi_spec.cache_clear()
+
+
+def input_schema_for_tool(name: str) -> Dict[str, Any]:
+    return operation_input_schema(get_tool(name))
+
+
+def all_tool_input_schemas() -> Dict[str, Dict[str, Any]]:
+    """Resolve input schemas for every allowlisted tool (fails if OpenAPI drifts)."""
+    spec = load_openapi_spec()
+    return {tool.name: operation_input_schema(tool, spec=spec) for tool in PUBLIC_TOOLS}
+
+
+def openapi_has_operation(method: str, path: str) -> bool:
+    spec = load_openapi_spec()
+    path_item = (spec.get("paths") or {}).get(path)
+    if not isinstance(path_item, dict):
+        return False
+    return isinstance(path_item.get(method.lower()), dict)