239 lines
9.6 KiB
Python
239 lines
9.6 KiB
Python
"""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"
|
|
)
|