258 lines
7.9 KiB
Python
258 lines
7.9 KiB
Python
"""Tool registry and invocation helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Mapping
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
def _find_repo_root() -> Path:
|
|
# Prefer cwd so local dev runs from the repo root without packaging.
|
|
candidates = (Path.cwd(), Path(__file__).resolve())
|
|
for base in candidates:
|
|
for parent in (base, *base.parents):
|
|
if (parent / "docs" / "constitution.md").exists():
|
|
return parent
|
|
if (parent / ".git").exists():
|
|
return parent
|
|
return Path(__file__).resolve().parents[3]
|
|
|
|
|
|
RepoPath = _find_repo_root()
|
|
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, object, 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 show_constitution(_: Mapping[str, str]) -> tuple[str, str, list[str]]:
|
|
"""Return the AI constitution contents."""
|
|
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 the list of available playbooks."""
|
|
return ("ok", _list_playbooks(), [])
|
|
|
|
|
|
def show_reference(_: Mapping[str, str]) -> tuple[str, str, list[str]]:
|
|
"""Return the reference index contents."""
|
|
return ("ok", _list_reference_topics(), [])
|
|
|
|
|
|
def search_docs(params: Mapping[str, str]) -> tuple[str, str, list[str]]:
|
|
"""Search docs for a query string and return matches."""
|
|
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]]:
|
|
"""Return MCP task list contents."""
|
|
tasks_file = RepoPath / "specs" / "001-mcp-server" / "tasks.md"
|
|
return ("ok", _read_text(tasks_file) or "Tasks not found.", [])
|
|
|
|
|
|
def sync_docs(_: Mapping[str, str]) -> tuple[str, object, list[str]]:
|
|
"""Return catalog vs docs drift report."""
|
|
from mcp_server.docs_sync import check_catalog_parity
|
|
|
|
return ("ok", check_catalog_parity(), [])
|
|
|
|
|
|
def _tool_handlers() -> dict[str, Callable[[Mapping[str, str]], tuple[str, object, list[str]]]]:
|
|
return {
|
|
"show-constitution": show_constitution,
|
|
"list-playbooks": list_playbooks,
|
|
"show-reference": show_reference,
|
|
"search-docs": search_docs,
|
|
"list-mcp-tasks": list_tasks,
|
|
"sync-docs": sync_docs,
|
|
}
|
|
|
|
|
|
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",
|
|
)
|
|
anchor_sync = DocsAnchor(
|
|
path=DocsPath / "reference" / "mcp-server.md",
|
|
anchor="sync-docs",
|
|
summary="Compare tool catalog against documented anchors",
|
|
)
|
|
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"],
|
|
),
|
|
Tool(
|
|
name="sync-docs",
|
|
description="Compare tool catalog against docs reference anchors.",
|
|
inputs=(),
|
|
docs_anchor=anchor_sync,
|
|
handler=handlers["sync-docs"],
|
|
),
|
|
)
|
|
|
|
|
|
def list_tools_payload() -> dict[str, object]:
|
|
"""Render tool catalog payload for tool discovery."""
|
|
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 tools/list to see supported tools"],
|
|
"docsAnchor": {},
|
|
}
|
|
status, output, actions = tool.handler(args)
|
|
return {
|
|
"status": status,
|
|
"output": output,
|
|
"actions": actions,
|
|
"docsAnchor": tool.docs_anchor.as_dict(),
|
|
}
|