Vishesh 'ironeagle' Bangotra
8cc9631fe7
Add lib page 11 (SemVer contract, pre-1.0 rule, tag==version invariant, Keep a Changelog), the releasing wiki runbook, and canonical release templates (CHANGELOG.md, VERSION). Vendor them into the MCP bundle as templates/release + release.json and expose registry://release with list/get tools. Add a changelog component to the app template. Enforce the tag<->version contract in both Drone shapes: Type A check-version and the backfill loop verify pyproject version; Type B adds a VERSION guard step.
547 lines
22 KiB
Python
547 lines
22 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:
|
|
docs://index, docs://nav, docs://modules/{module} (from the base server)
|
|
docs://spec (the pinned auth-server OpenAPI contract)
|
|
registry://libraries (the structured catalog)
|
|
registry://components (the template component index)
|
|
registry://drone (the CI template index)
|
|
registry://release (the release template 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)
|
|
get_app_template (the whole app skeleton as a path -> content map)
|
|
list_drone_templates (the two CI shapes: PyPI package / Docker deployment)
|
|
get_drone_template (kind -> exact .drone.yml template)
|
|
list_release_templates (the changelog + version-file artefacts)
|
|
get_release_template (kind -> exact release template file)
|
|
|
|
Prompts:
|
|
intake (the questions to gather before building)
|
|
build (the two axes -> deterministic build prompt)
|
|
start_app (combined entry: intake, then build)
|
|
trial (self-contained trial build into a scratch dir)
|
|
"""
|
|
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)
|
|
|
|
|
|
_INTAKE_TEXT = (
|
|
"# Start a new app — intake\n\n"
|
|
"Gather the following, then call `build` (or `start_app`) 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). If yes, the auth-server service provides it —\n"
|
|
" read `docs://spec` + `docs://modules/auth_integration` (no auth code to write).\n"
|
|
"5. Integrations: Mongo / ingestion formats / email / LLM provider.\n\n"
|
|
"Then: `build(app_name=..., resources=..., pipelines=..., auth=...)`.\n"
|
|
)
|
|
|
|
|
|
def _build_text(app_name: str, resources: str, pipelines: str, auth: str) -> str:
|
|
"""Render the deterministic build prompt from the two axes.
|
|
|
|
Args:
|
|
app_name (str): App name + one-line purpose.
|
|
resources (str): Axis 1 — resources, fields, relationships.
|
|
pipelines (str): Axis 2 — pipeline families + input formats.
|
|
auth (str): Whether auth is needed now ("yes"/"no").
|
|
|
|
Returns:
|
|
str: The build prompt.
|
|
"""
|
|
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 `docs://modules/build_recipe` and follow it step by step.\n"
|
|
"2. Read `docs://modules/genericity_matrix` (Tier A/B/C) and "
|
|
"`docs://modules/layering_rules`.\n"
|
|
"3. Call `find_library(capability)` for db, di, routing, ingestion, "
|
|
"and pipeline; never reimplement a Tier A library.\n"
|
|
"4. Scaffold with `get_app_template()` (whole tree) or pull single "
|
|
"files with `list_app_components()` and "
|
|
"`get_app_component(kind, resource=..., family=...)`.\n"
|
|
"5. If auth is needed, read `docs://spec` (the auth-server contract) and "
|
|
"`docs://modules/auth_integration`; declare `securitySchemes.bearerAuth` "
|
|
"and `security` in your spec — do not write auth code.\n"
|
|
"6. Author only Tier C; keep Tier B as shipped.\n"
|
|
)
|
|
|
|
|
|
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 _spec(self) -> dict[str, Any]:
|
|
"""Load the pinned OpenAPI contract (auth-server)."""
|
|
path = Path(self.mcp_root) / "specs" / "auth-server.json"
|
|
if not path.exists():
|
|
return {"error": "not_found", "path": str(path)}
|
|
return json.loads(path.read_bytes())
|
|
|
|
def _drone(self) -> dict[str, Any]:
|
|
"""Load the CI (Drone) template manifest."""
|
|
path = Path(self.mcp_root) / "drone.json"
|
|
if not path.exists():
|
|
return {"error": "not_found", "path": str(path), "components": []}
|
|
return json.loads(path.read_bytes())
|
|
|
|
def _release(self) -> dict[str, Any]:
|
|
"""Load the release (versioning + changelog) template manifest."""
|
|
path = Path(self.mcp_root) / "release.json"
|
|
if not path.exists():
|
|
return {"error": "not_found", "path": str(path), "components": []}
|
|
return json.loads(path.read_bytes())
|
|
|
|
def _find_release_component(self, kind: str) -> dict[str, Any] | None:
|
|
for entry in self._release().get("components", []):
|
|
if entry.get("kind") == kind:
|
|
return entry
|
|
return None
|
|
|
|
def _read_release_component(self, entry: dict[str, Any]) -> dict[str, Any]:
|
|
"""Read a vendored release standard file."""
|
|
rel = entry["path"]
|
|
path = Path(self.mcp_root) / "templates" / "release" / rel
|
|
if not path.exists():
|
|
return {"error": "not_found", "path": str(path)}
|
|
return {
|
|
"kind": entry["kind"],
|
|
"path": rel,
|
|
"placeholder": entry.get("placeholder"),
|
|
"content": path.read_text(encoding="utf-8"),
|
|
}
|
|
|
|
def _find_drone_component(self, kind: str) -> dict[str, Any] | None:
|
|
for entry in self._drone().get("components", []):
|
|
if entry.get("kind") == kind:
|
|
return entry
|
|
return None
|
|
|
|
def _read_drone_component(self, entry: dict[str, Any]) -> dict[str, Any]:
|
|
"""Read a vendored drone template file."""
|
|
rel = entry["path"]
|
|
path = Path(self.mcp_root) / "templates" / "drone" / rel
|
|
if not path.exists():
|
|
return {"error": "not_found", "path": str(path)}
|
|
return {
|
|
"kind": entry["kind"],
|
|
"path": rel,
|
|
"placeholder": entry.get("placeholder"),
|
|
"content": path.read_text(encoding="utf-8"),
|
|
}
|
|
|
|
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 _template_files(self, prefix: str | None = None) -> dict[str, str]:
|
|
"""Read the vendored template tree as a path -> content map.
|
|
|
|
Args:
|
|
prefix (str | None): Optional path prefix to filter by (POSIX
|
|
separators, for example ``"infra/"``).
|
|
|
|
Returns:
|
|
dict[str, str]: Relative POSIX path -> UTF-8 file content, excluding
|
|
caches and bytecode.
|
|
"""
|
|
root = Path(self.mcp_root) / "templates" / "app"
|
|
files: dict[str, str] = {}
|
|
if not root.is_dir():
|
|
return files
|
|
for path in sorted(root.rglob("*")):
|
|
if not path.is_file():
|
|
continue
|
|
if "__pycache__" in path.parts or path.suffix == ".pyc":
|
|
continue
|
|
rel = path.relative_to(root).as_posix()
|
|
if prefix and not rel.startswith(prefix):
|
|
continue
|
|
files[rel] = path.read_text(encoding="utf-8")
|
|
return files
|
|
|
|
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.resource("registry://drone")
|
|
def drone() -> dict[str, Any]:
|
|
return self._drone()
|
|
|
|
@self.app.resource("registry://release")
|
|
def release() -> dict[str, Any]:
|
|
return self._release()
|
|
|
|
@self.app.resource("docs://spec")
|
|
def spec() -> dict[str, Any]:
|
|
"""Return the pinned auth-server OpenAPI contract.
|
|
|
|
Use this (instead of a live fetch) to learn the auth-server client
|
|
contract when wiring authentication into an app.
|
|
"""
|
|
return self._spec()
|
|
|
|
@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", "")),
|
|
str(lib.get("spec", "")),
|
|
str(lib.get("recipe", "")),
|
|
" ".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.tool()
|
|
def get_app_template(prefix: str = "") -> dict[str, Any]:
|
|
"""Return the whole app skeleton as a path -> content map.
|
|
|
|
Use this to scaffold in one shot instead of pulling components one
|
|
by one; combine with :func:`list_app_components` for the kind and
|
|
placeholder metadata. Seed tokens (``item`` / ``example``) are left
|
|
as-is here — fetch a renamed file with :func:`get_app_component`.
|
|
|
|
Args:
|
|
prefix (str): Optional path prefix filter with POSIX
|
|
separators, for example ``"infra/"`` or ``"pipelines/"``.
|
|
Empty returns every file.
|
|
|
|
Returns:
|
|
dict[str, Any]: ``{root, file_count, files: {relpath: content}}``.
|
|
"""
|
|
files = self._template_files(prefix or None)
|
|
return {
|
|
"root": "templates/app",
|
|
"file_count": len(files),
|
|
"files": files,
|
|
}
|
|
|
|
@self.app.tool()
|
|
def list_drone_templates() -> list[dict[str, Any]]:
|
|
"""List the CI (Drone) template shapes.
|
|
|
|
There are exactly two types of ``.drone.yml`` in the org: a Python
|
|
package published to the private PyPI index, and a Docker
|
|
deployment that builds, pushes, and runs a container.
|
|
|
|
Returns:
|
|
list[dict[str, Any]]:
|
|
Entries with ``kind``, ``path``, ``placeholder`` and
|
|
``description``. Fetch one with :func:`get_drone_template`.
|
|
"""
|
|
return self._drone().get("components", [])
|
|
|
|
@self.app.tool()
|
|
def get_drone_template(kind: str) -> dict[str, Any]:
|
|
"""Return the exact CI template for a component kind.
|
|
|
|
Use this to copy the current ``.drone.yml`` shape instead of
|
|
inventing one. See :func:`list_drone_templates` for valid kinds
|
|
("drone-pypi", "drone-service").
|
|
|
|
Args:
|
|
kind (str): Component kind, e.g. "drone-pypi" or
|
|
"drone-service".
|
|
|
|
Returns:
|
|
dict[str, Any]: ``{kind, path, placeholder, content}`` or an
|
|
error entry when the kind is unknown.
|
|
"""
|
|
entry = self._find_drone_component(kind)
|
|
if entry is None:
|
|
return {
|
|
"error": "unknown_kind",
|
|
"kind": kind,
|
|
"available": [c["kind"] for c in self._drone().get("components", [])],
|
|
}
|
|
return self._read_drone_component(entry)
|
|
|
|
@self.app.tool()
|
|
def list_release_templates() -> list[dict[str, Any]]:
|
|
"""List the release/versioning template artefacts.
|
|
|
|
These are the files every repo carries for releases: a Keep a
|
|
Changelog ``CHANGELOG.md`` and, for services, a root ``VERSION``
|
|
file. See standards "11_versioning_and_changelog".
|
|
|
|
Returns:
|
|
list[dict[str, Any]]:
|
|
Entries with ``kind``, ``path``, ``placeholder`` and
|
|
``description``. Fetch one with :func:`get_release_template`.
|
|
"""
|
|
return self._release().get("components", [])
|
|
|
|
@self.app.tool()
|
|
def get_release_template(kind: str) -> dict[str, Any]:
|
|
"""Return the exact release standard file for a component kind.
|
|
|
|
Use this to drop the canonical changelog / version file into a repo
|
|
instead of inventing one. See :func:`list_release_templates` for
|
|
valid kinds ("changelog", "version-file").
|
|
|
|
Args:
|
|
kind (str): Component kind, e.g. "changelog" or "version-file".
|
|
|
|
Returns:
|
|
dict[str, Any]: ``{kind, path, placeholder, content}`` or an
|
|
error entry when the kind is unknown.
|
|
"""
|
|
entry = self._find_release_component(kind)
|
|
if entry is None:
|
|
return {
|
|
"error": "unknown_kind",
|
|
"kind": kind,
|
|
"available": [c["kind"] for c in self._release().get("components", [])],
|
|
}
|
|
return self._read_release_component(entry)
|
|
|
|
@self.app.prompt()
|
|
def intake() -> str:
|
|
"""Return the questions to gather before building an app.
|
|
|
|
Use this (or `start_app` with no arguments) to collect the two axes
|
|
and the integration list. Answer them, then call `build`.
|
|
|
|
Returns:
|
|
str: The intake questions.
|
|
"""
|
|
return _INTAKE_TEXT
|
|
|
|
@self.app.prompt()
|
|
def build(
|
|
app_name: str = "",
|
|
resources: str = "",
|
|
pipelines: str = "",
|
|
auth: str = "no",
|
|
) -> str:
|
|
"""Emit the deterministic build prompt from the two axes.
|
|
|
|
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 build prompt, or the intake questions when the axes
|
|
are missing.
|
|
"""
|
|
if not app_name or not resources:
|
|
return _INTAKE_TEXT
|
|
return _build_text(app_name, resources, pipelines, auth)
|
|
|
|
@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 _INTAKE_TEXT
|
|
return _build_text(app_name, resources, pipelines, auth)
|
|
|
|
@self.app.prompt()
|
|
def trial(app_name: str = "", build_path: str = "") -> str:
|
|
"""Prompt a self-contained trial build into a scratch directory.
|
|
|
|
Use this to validate the standard end-to-end without touching a real
|
|
app: generate into a throwaway path, prove it builds, and record
|
|
findings.
|
|
|
|
Args:
|
|
app_name (str): App to build for the trial.
|
|
build_path (str): Where to write the trial app.
|
|
|
|
Returns:
|
|
str: The trial prompt.
|
|
"""
|
|
target = build_path or f"apps/{app_name or '<app>'}-trial"
|
|
return (
|
|
f"# Trial: {app_name or '<app>'}\n\n"
|
|
f"Build into `{target}` (a throwaway path) and validate the "
|
|
"standard end-to-end.\n\n"
|
|
"## Steps\n"
|
|
"1. Run the `build` prompt to get the deterministic build prompt.\n"
|
|
f"2. Scaffold into `{target}` only — never an existing app.\n"
|
|
"3. Prove it: `python -c \"import main\"` and `pytest`.\n"
|
|
"4. Record every friction point as a finding: missing component, "
|
|
"wrong codegen shape, ambiguous doc, or missing prompt.\n"
|
|
"5. Do not fix the standard from inside the trial; report findings "
|
|
"back so the standard can be amended.\n\n"
|
|
f"Delete `{target}` once you are satisfied.\n"
|
|
)
|