"""Standards MCP server. Serves the hand-authored `standards` bundle (the application blueprint plus the library registry and the runnable template) and adds discovery/scaffolding helpers so an agent can find the right internal library — and pull exact template code — before writing any infrastructure. Resources: doc://index, doc://nav, doc://modules/{module} (from the base server) registry://libraries (the structured catalog) registry://components (the template component index) Tools: ping (from the base server) find_library (capability -> matching libraries + MCP endpoints) list_app_components (the template component kinds) get_app_component (kind -> exact template file, seed tokens renamed) Prompts: start_app (collect the two axes, then emit the build prompt) """ from __future__ import annotations import json import re from pathlib import Path from typing import Any from config import MCPBundle from mcp_server import MCPServer def _pascal(name: str) -> str: """Convert a snake/kebab name to PascalCase.""" return "".join(part.capitalize() for part in re.split(r"[_\-\s]+", name) if part) class StandardsMCPServer(MCPServer): """MCP server for the `standards` bundle with discovery/scaffolding tools. Attributes: mcp_root (Path): Directory containing the generated bundle (index.json, nav.json, modules/, libraries.json, components.json, and templates/). """ def __init__(self, bundle: MCPBundle) -> None: super().__init__(bundle) self._register_standards() # ------------------------------------------------------------------ # Bundle readers # ------------------------------------------------------------------ def _catalog(self) -> dict[str, Any]: """Load the machine-readable library registry.""" path = Path(self.mcp_root) / "libraries.json" if not path.exists(): return {"error": "not_found", "path": str(path), "libraries": []} return json.loads(path.read_bytes()) def _components(self) -> dict[str, Any]: """Load the template component manifest.""" path = Path(self.mcp_root) / "components.json" if not path.exists(): return {"error": "not_found", "path": str(path), "components": []} return json.loads(path.read_bytes()) def _find_component(self, kind: str) -> dict[str, Any] | None: for entry in self._components().get("components", []): if entry.get("kind") == kind: return entry return None def _read_component( self, entry: dict[str, Any], resource: str | None, family: str | None, ) -> dict[str, Any]: """Read a vendored template file, renaming seed tokens when asked.""" rel = entry["path"] path = Path(self.mcp_root) / "templates" / "app" / rel if not path.exists(): return {"error": "not_found", "path": str(path)} text = path.read_text(encoding="utf-8") placeholder = entry.get("placeholder") if resource and placeholder == "resource": text = text.replace("Item", _pascal(resource)).replace("item", resource) if family and placeholder == "family": text = text.replace("Example", _pascal(family)).replace("example", family) return { "kind": entry["kind"], "path": rel, "placeholder": placeholder, "content": text, } # ------------------------------------------------------------------ # Standards surface # ------------------------------------------------------------------ def _register_standards(self) -> None: @self.app.resource("registry://libraries") def libraries() -> dict[str, Any]: return self._catalog() @self.app.resource("registry://components") def components() -> dict[str, Any]: return self._components() @self.app.tool() def find_library(capability: str) -> list[dict[str, Any]]: """Find internal libraries matching a capability keyword. Use this before writing infrastructure: it returns the library to use, its import name, what it replaces, and its MCP endpoint. Args: capability (str): A keyword such as "database", "pdf", "auth", "pipeline", "openapi", or "email". Returns: list[dict[str, Any]]: Matching library entries (empty when nothing matches). """ needle = (capability or "").strip().lower() if not needle: return self._catalog().get("libraries", []) matches: list[dict[str, Any]] = [] for lib in self._catalog().get("libraries", []): haystack = " ".join( [ str(lib.get("name", "")), str(lib.get("import", "")), str(lib.get("purpose", "")), str(lib.get("use_instead_of", "")), " ".join(lib.get("capabilities", [])), ] ).lower() if needle in haystack: matches.append(lib) return matches @self.app.tool() def list_app_components() -> list[dict[str, Any]]: """List the reusable component kinds shipped in the app template. Returns: list[dict[str, Any]]: Entries with ``kind``, ``path``, ``placeholder`` and ``description``. Fetch one with :func:`get_app_component`. """ return self._components().get("components", []) @self.app.tool() def get_app_component( kind: str, resource: str | None = None, family: str | None = None, ) -> dict[str, Any]: """Return the exact template file for a component kind. Use this to copy current, app-neutral skeleton code instead of inventing it. See :func:`list_app_components` for valid kinds. Args: kind (str): Component kind, e.g. "routes", "pipeline-impl". resource (str | None): Rename the resource seed ("item"/"Item") to this name (snake_case), for per-resource kinds. family (str | None): Rename the family seed ("example"/"Example") to this name (snake_case), for per-family kinds. Returns: dict[str, Any]: ``{kind, path, placeholder, content}`` or an error entry when the kind is unknown. """ entry = self._find_component(kind) if entry is None: return { "error": "unknown_kind", "kind": kind, "available": [c["kind"] for c in self._components().get("components", [])], } return self._read_component(entry, resource, family) @self.app.prompt() def start_app( app_name: str = "", resources: str = "", pipelines: str = "", auth: str = "no", ) -> str: """Entry point to build a new app from the standards blueprint. Call with no arguments to get the intake questions; call again with the answers to get the deterministic build prompt. Args: app_name (str): App name + one-line purpose. resources (str): Axis 1 — resources, fields, relationships (or a pasted openapi.yaml). pipelines (str): Axis 2 — pipeline families + input formats. auth (str): Whether auth is needed now ("yes"/"no"). Returns: str: The intake questions, or the build prompt when complete. """ if not app_name or not resources: return ( "# Start a new app — intake\n\n" "Gather the following, then call `start_app` again with the values:\n\n" "1. App name + one-line purpose.\n" "2. Axis 1 — resources: each resource, its fields, and relationships\n" " (or paste `openapi.yaml`).\n" "3. Axis 2 — pipelines: multi-step inputs? families + formats (or `none`).\n" "4. Auth needed now? (yes/no; auth is deferred — `py-jwt` only).\n" "5. Integrations: Mongo / ingestion formats / email / LLM provider.\n\n" "Then: `start_app(app_name=..., resources=..., pipelines=..., auth=...)`.\n" ) return ( f"# Build: {app_name}\n\n" "## Spec\n" f"- resources: {resources}\n" f"- pipelines: {pipelines or 'none'}\n" f"- auth: {auth}\n\n" "## Method (deterministic)\n" "1. Read `doc://modules/build_recipe` and follow it step by step.\n" "2. Read `doc://modules/genericity_matrix` (Tier A/B/C) and " "`doc://modules/layering_rules`.\n" "3. Call `find_library(capability)` for db, di, routing, ingestion, " "and pipeline; never reimplement a Tier A library.\n" "4. Pull exact files with `list_app_components()` and " "`get_app_component(kind, resource=..., family=...)`.\n" "5. Author only Tier C; keep Tier B as shipped.\n" )