mcp

scripts/mcp-server/src/mcp_server/tools.py

@@ -0,0 +1,214 @@
+"""Tool registry and invocation helpers."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from pathlib import Path
+
+RepoPath = Path(__file__).resolve().parents[3]
+DocsPath = RepoPath / "docs"
+
+
+@dataclass(frozen=True)
+class DocsAnchor:
+    """Documentation pointer for a tool."""
+
+    path: Path
+    anchor: str
+    summary: str
+
+    def as_dict(self) -> dict[str, str]:
+        """Serialize the anchor for transport."""
+        return {"path": str(self.path), "anchor": self.anchor, "summary": self.summary}
+
+
+@dataclass(frozen=True)
+class InputParam:
+    """Input parameter definition."""
+
+    name: str
+    type: str
+    required: bool
+    description: str
+
+    def as_dict(self) -> dict[str, str | bool]:
+        """Serialize the input parameter for transport."""
+        return {
+            "name": self.name,
+            "type": self.type,
+            "required": self.required,
+            "description": self.description,
+        }
+
+
+@dataclass(frozen=True)
+class Tool:
+    """Tool metadata and handler binding."""
+
+    name: str
+    description: str
+    inputs: tuple[InputParam, ...]
+    docs_anchor: DocsAnchor
+    handler: Callable[[Mapping[str, str]], tuple[str, str, list[str]]]
+
+    def as_dict(self) -> dict[str, object]:
+        """Serialize tool metadata for transport."""
+        return {
+            "name": self.name,
+            "description": self.description,
+            "inputs": list(map(InputParam.as_dict, self.inputs)),
+            "docsAnchor": self.docs_anchor.as_dict(),
+        }
+
+
+def _read_text(path: Path) -> str:
+    if not path.exists():
+        return ""
+    return path.read_text()
+
+
+def _list_playbooks() -> str:
+    playbooks = sorted((DocsPath / "playbooks").glob("*.md"))
+    if not playbooks:
+        return "No playbooks found."
+    return "\n".join(p.name for p in playbooks)
+
+
+def _list_reference_topics() -> str:
+    reference = DocsPath / "reference" / "index.md"
+    return _read_text(reference) or "Reference index is empty."
+
+
+def _search_docs(term: str) -> str:
+    files = sorted(DocsPath.rglob("*.md"))
+    matches = []
+    for path in files:
+        content = path.read_text()
+        if term.lower() in content.lower():
+            matches.append(f"{path}: {term}")
+    if not matches:
+        return "No matches found."
+    return "\n".join(matches[:20])
+
+
+def _tool_handlers() -> dict[str, Callable[[Mapping[str, str]], tuple[str, str, list[str]]]]:
+    def show_constitution(_: Mapping[str, str]) -> tuple[str, str, list[str]]:
+        text = _read_text(DocsPath / "constitution.md")
+        return ("ok", text or "Constitution not found.", [])
+
+    def list_playbooks(_: Mapping[str, str]) -> tuple[str, str, list[str]]:
+        return ("ok", _list_playbooks(), [])
+
+    def show_reference(_: Mapping[str, str]) -> tuple[str, str, list[str]]:
+        return ("ok", _list_reference_topics(), [])
+
+    def search_docs(params: Mapping[str, str]) -> tuple[str, str, list[str]]:
+        term = params.get("query", "")
+        if not term:
+            return ("invalid_input", "Missing query", [])
+        return ("ok", _search_docs(term), [])
+
+    def list_tasks(_: Mapping[str, str]) -> tuple[str, str, list[str]]:
+        tasks_file = RepoPath / "specs" / "001-mcp-server" / "tasks.md"
+        return ("ok", _read_text(tasks_file) or "Tasks not found.", [])
+
+    return {
+        "show-constitution": show_constitution,
+        "list-playbooks": list_playbooks,
+        "show-reference": show_reference,
+        "search-docs": search_docs,
+        "list-mcp-tasks": list_tasks,
+    }
+
+
+def tool_catalog() -> tuple[Tool, ...]:
+    """Return the available MCP tools and their metadata."""
+    handlers = _tool_handlers()
+    anchor_constitution = DocsAnchor(
+        path=DocsPath / "constitution.md",
+        anchor="ai-constitution-for-the-nixos-repository",
+        summary="Authoritative rules and workflows",
+    )
+    anchor_playbooks = DocsAnchor(
+        path=DocsPath / "playbooks" / "template.md",
+        anchor="playbook-template",
+        summary="Playbook index and template reference",
+    )
+    anchor_reference = DocsAnchor(
+        path=DocsPath / "reference" / "index.md",
+        anchor="reference-index",
+        summary="Navigation map for repository docs",
+    )
+    anchor_search = DocsAnchor(
+        path=DocsPath,
+        anchor="docs-search",
+        summary="Search across docs for maintenance topics",
+    )
+    anchor_tasks = DocsAnchor(
+        path=RepoPath / "specs" / "001-mcp-server" / "tasks.md",
+        anchor="tasks",
+        summary="Implementation tasks for MCP feature",
+    )
+    return (
+        Tool(
+            name="show-constitution",
+            description="Display repository AI constitution for rule lookup.",
+            inputs=(),
+            docs_anchor=anchor_constitution,
+            handler=handlers["show-constitution"],
+        ),
+        Tool(
+            name="list-playbooks",
+            description="List available playbooks under docs/playbooks.",
+            inputs=(),
+            docs_anchor=anchor_playbooks,
+            handler=handlers["list-playbooks"],
+        ),
+        Tool(
+            name="show-reference",
+            description="Show docs/reference/index.md for navigation guidance.",
+            inputs=(),
+            docs_anchor=anchor_reference,
+            handler=handlers["show-reference"],
+        ),
+        Tool(
+            name="search-docs",
+            description="Search docs for a query string.",
+            inputs=(InputParam("query", "string", True, "Term to search for"),),
+            docs_anchor=anchor_search,
+            handler=handlers["search-docs"],
+        ),
+        Tool(
+            name="list-mcp-tasks",
+            description="Show MCP feature task list from specs.",
+            inputs=(),
+            docs_anchor=anchor_tasks,
+            handler=handlers["list-mcp-tasks"],
+        ),
+    )
+
+
+def list_tools_payload() -> dict[str, object]:
+    """Render tool catalog payload for listTools."""
+    return {"tools": [tool.as_dict() for tool in tool_catalog()]}
+
+
+def invoke_tool(name: str, args: Mapping[str, str]) -> dict[str, object]:
+    """Invoke a tool and return standardized result payload."""
+    registry: dict[str, Tool] = {tool.name: tool for tool in tool_catalog()}
+    tool = registry.get(name)
+    if not tool:
+        return {
+            "status": "unsupported",
+            "output": f"Tool '{name}' is not available.",
+            "actions": ["call listTools to see supported tools"],
+            "docsAnchor": {},
+        }
+    status, output, actions = tool.handler(args)
+    return {
+        "status": status,
+        "output": output,
+        "actions": actions,
+        "docsAnchor": tool.docs_anchor.as_dict(),
+    }