From fe6025fa630787b59e31de09174714a12b410628 Mon Sep 17 00:00:00 2001 From: Labyricorn Date: Sun, 23 Aug 2026 10:48:33 -0700 Subject: [PATCH] Add local repository MCP server --- README.md | 99 ++++++ labyricorn_mcp/__init__.py | 3 + labyricorn_mcp/__main__.py | 36 ++ labyricorn_mcp/domain.py | 561 +++++++++++++++++++++++++++++++ labyricorn_mcp/errors.py | 27 ++ labyricorn_mcp/gitops.py | 269 +++++++++++++++ labyricorn_mcp/safety.py | 126 +++++++ labyricorn_mcp/server.py | 151 +++++++++ labyricorn_mcp/tools.py | 219 ++++++++++++ labyricorn_mcp/validation.py | 149 ++++++++ scripts/labyricorn_mcp_server.py | 17 + tests/test_labyricorn_mcp.py | 266 +++++++++++++++ 12 files changed, 1923 insertions(+) create mode 100644 labyricorn_mcp/__init__.py create mode 100644 labyricorn_mcp/__main__.py create mode 100644 labyricorn_mcp/domain.py create mode 100644 labyricorn_mcp/errors.py create mode 100644 labyricorn_mcp/gitops.py create mode 100644 labyricorn_mcp/safety.py create mode 100644 labyricorn_mcp/server.py create mode 100644 labyricorn_mcp/tools.py create mode 100644 labyricorn_mcp/validation.py create mode 100644 scripts/labyricorn_mcp_server.py create mode 100644 tests/test_labyricorn_mcp.py diff --git a/README.md b/README.md index e55af2b..a929839 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,8 @@ assets/static/ CSS, filtering JavaScript, favicon, and static assets configs/project-sources.ini Approved public remote-project registry scripts/project_sources.py Remote-content validator and importer scripts/build_with_projects.py Isolated build/preview entry point +scripts/labyricorn_mcp_server.py Local stdio MCP launcher +labyricorn_mcp/ Repository-scoped MCP protocol and domain adapters scripts/trigger_project_refresh.py Authenticated remote-project refresh client ops/deploy-labyricorn Versioned copy of the production deployment command tests/ Importer validation tests @@ -320,6 +322,103 @@ test -f build/index.html The project currently has no third-party Lektor packages or plugins. +### Local Labyricorn MCP server + +Labyricorn MCP is a local, stdio-based, repository-scoped management interface +for this checkout. An MCP-capable AI client launches it as a child process and +receives structured tools for repository information, models, content, tracked +tags, validated project snapshots, allowlisted design files, local validation, +and constrained Git operations. + +It is not the Labyricorn website server, a deployment or remote-administration +service, a general filesystem MCP, a shell MCP, or a production infrastructure +controller. It creates no listener, HTTP endpoint, OAuth flow, persistent +service, scheduled job, or additional deployment mechanism. The boundary stays: + +```text +MCP-capable AI client + | + stdio + | +Labyricorn MCP process + | +local Labyricorn Lektor repository + | + explicit ordinary Git push + | + Gitea + | +existing external update/deployment system + | +live site +``` + +The implementation uses Python's standard library and the repository's existing +article, blog, tracked-tag, project-import, build, and Git helpers. It adds no +Python package dependency. Launch it from a terminal with: + +```bash +python scripts/labyricorn_mcp_server.py --repository /absolute/path/to/labyricorn-site +``` + +On Windows, use the existing Lektor pipx Python when available. Project-cache +inspection and local builds need Lektor in the server runtime; the usual pipx +location is: + +```text +%LOCALAPPDATA%\pipx\pipx\venvs\lektor\Scripts\python.exe +``` + +For an MCP client that accepts the common `mcpServers` configuration shape, use +absolute paths. A Windows example is: + +```json +{ + "mcpServers": { + "labyricorn": { + "command": "C:\\Users\\YOUR-NAME\\AppData\\Local\\pipx\\pipx\\venvs\\lektor\\Scripts\\python.exe", + "args": [ + "C:\\absolute\\path\\to\\labyricorn-site\\scripts\\labyricorn_mcp_server.py", + "--repository", + "C:\\absolute\\path\\to\\labyricorn-site" + ] + } + } +} +``` + +The client must launch the process directly; stdout is reserved exclusively for +newline-delimited MCP JSON-RPC messages and diagnostics go to stderr. The server +supports MCP protocol revisions `2025-11-25`, `2025-06-18`, `2025-03-26`, and +`2024-11-05`. + +The focused v1 tool surface includes: + +- repository/model/content reads plus safe article and blog create/update; +- tracked-tag list, usage, create, update, and explicit untrack; +- allowlisted project sources and locally cached snapshots validated through the + existing importer, plus an explicit networked source-validation check; +- inspection of templates and static text assets, with hash-guarded updates only + under `templates/` and `assets/static/`; +- the existing isolated project-aware build, unit tests, diff check, and a + consolidated validation result; +- Git status, diff, and log; clean fast-forward pull; reviewed-path-only commit; + and normal upstream push. + +Mutations use repository-relative path containment and reject traversal, +out-of-tree resolution, and linked-path writes. Content and design updates +require the SHA-256 returned by the preceding read, so a concurrent human edit +causes `FILE_CHANGED` instead of being overwritten. Commits and pushes are +separate explicit actions. Commit accepts only reviewed changed paths, refuses +credential-like files and unrelated staged changes, and never pushes. Push +cannot force or rewrite history. A push from `main` requires an explicit +acknowledgement that Gitea's existing external site-update workflow may run. +The MCP itself never calls, inspects, restarts, or reconfigures production. + +Content deletion and tracked-tag slug renaming are intentionally absent from v1. +Untracking a tag removes only its registry record and preserves every content +reference, matching the existing tracked-tag editor. + ### Local management launcher To check the checkout's Git synchronization state and open the focused desktop diff --git a/labyricorn_mcp/__init__.py b/labyricorn_mcp/__init__.py new file mode 100644 index 0000000..d7bfe57 --- /dev/null +++ b/labyricorn_mcp/__init__.py @@ -0,0 +1,3 @@ +"""Local, repository-scoped MCP management interface for Labyricorn.""" + +__version__ = "0.1.0" diff --git a/labyricorn_mcp/__main__.py b/labyricorn_mcp/__main__.py new file mode 100644 index 0000000..7e24ec5 --- /dev/null +++ b/labyricorn_mcp/__main__.py @@ -0,0 +1,36 @@ +"""Command-line entry point for the stdio MCP server.""" + +from __future__ import annotations + +import argparse +import logging +import sys + +from .errors import LabyricornMcpError +from .safety import RepositoryGuard +from .server import StdioMcpServer + + +def main() -> int: + parser = argparse.ArgumentParser(description="Run the local repository-scoped Labyricorn MCP server over stdio.") + parser.add_argument( + "--repository", + help="Labyricorn repository root. Defaults to LABYRICORN_REPOSITORY, then the current directory.", + ) + parser.add_argument("--verbose", action="store_true", help="Write diagnostic detail to stderr.") + arguments = parser.parse_args() + logging.basicConfig( + stream=sys.stderr, + level=logging.INFO if arguments.verbose else logging.WARNING, + format="labyricorn-mcp: %(levelname)s: %(message)s", + ) + try: + guard = RepositoryGuard.discover(arguments.repository) + return StdioMcpServer(guard).serve_forever() + except LabyricornMcpError as exc: + print(f"labyricorn-mcp: ERROR: {exc.code}: {exc.message}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/labyricorn_mcp/domain.py b/labyricorn_mcp/domain.py new file mode 100644 index 0000000..125e5e5 --- /dev/null +++ b/labyricorn_mcp/domain.py @@ -0,0 +1,561 @@ +"""Domain adapters over the site's existing editors and importer.""" + +from __future__ import annotations + +import configparser +from dataclasses import asdict +from datetime import date +import hashlib +import json +import os +from pathlib import Path +import re +import sys +import tempfile +from typing import Any + +from .errors import LabyricornMcpError +from .safety import RepositoryGuard + + +DESIGN_ROOTS = ("templates", "assets/static") +DESIGN_SUFFIXES = {".html", ".css", ".js", ".json", ".svg", ".txt", ".xml"} +CONTENT_MODELS = {"article", "blog", "project", "devlog"} + + +def _hash(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def _load_site_modules(root: Path) -> dict[str, Any]: + scripts = str(root / "scripts") + if scripts not in sys.path: + sys.path.insert(0, scripts) + try: + import article_editor + import blog_editor + import project_editor + import tag_editor + except (ImportError, OSError) as exc: + raise LabyricornMcpError( + "INVALID_LABYRICORN_REPOSITORY", "The repository's management modules could not be loaded." + ) from exc + return { + "article": article_editor, + "blog": blog_editor, + "project": project_editor, + "tag": tag_editor, + } + + +class LabyricornDomain: + def __init__(self, guard: RepositoryGuard): + self.guard = guard + self.modules = _load_site_modules(guard.root) + + def _article_repository(self) -> Any: + repository = self.modules["article"].ArticleRepository(self.guard.root) + self._require_contained(repository.article_root, repository.tags_root) + return repository + + def _blog_repository(self) -> Any: + repository = self.modules["blog"].BlogRepository(self.guard.root) + self._require_contained(repository.blog_root, repository.tags_root) + return repository + + def _tag_repository(self) -> Any: + repository = self.modules["tag"].TagRepository(self.guard.root) + self._require_contained(repository.tags_root) + return repository + + def _project_registry(self) -> Any: + return self.modules["project"].ProjectSourceRegistry(self.guard.root) + + def _require_contained(self, *paths: Path) -> None: + if any(not path.resolve().is_relative_to(self.guard.root) for path in paths): + raise LabyricornMcpError( + "PATH_OUTSIDE_REPOSITORY", "A required content path resolves outside the repository." + ) + + def _translate(self, exc: Exception, *, not_found: bool = False) -> LabyricornMcpError: + message = str(exc).replace(str(self.guard.root), ".") + folded = message.casefold() + if "changed on disk" in folded or "reload" in folded: + code = "FILE_CHANGED" + elif not_found or "does not exist" in folded: + code = "CONTENT_NOT_FOUND" + elif "already exists" in folded: + code = "WORKTREE_CONFLICT" + elif "unsafe" in folded or "outside" in folded or "linked" in folded: + code = "PATH_OUTSIDE_REPOSITORY" + else: + code = "VALIDATION_FAILED" + return LabyricornMcpError(code, message) + + def site_structure(self) -> dict[str, Any]: + return { + "content": { + "articles": {"path": "content/articles", "model": "entry", "kicker": "Article", "ownership": "site"}, + "blog": {"path": "content/blog", "model": "entry", "kicker": "Blog", "ownership": "site"}, + "tags": {"path": "content/tags", "model": "tag", "ownership": "site"}, + "projects": { + "path": "content/projects", + "models": ["project", "devlog", "devlog-entry"], + "ownership": "validated remote sources", + "source_registry": "configs/project-sources.ini", + }, + }, + "design": { + "inspectable_roots": list(DESIGN_ROOTS), + "hash_guarded_text_mutation_roots": list(DESIGN_ROOTS), + "binary_assets": "metadata and hash inspection only", + }, + "models_path": "models", + "build_entry_point": "scripts/build_with_projects.py", + "transport": "stdio", + } + + def describe_model(self, model: str) -> dict[str, Any]: + if not isinstance(model, str) or not re.fullmatch(r"[a-z0-9-]+", model): + raise LabyricornMcpError("INVALID_CONTENT_MODEL", "Model names must use lowercase letters, numbers, and hyphens.") + path = self.guard.resolve(f"models/{model}.ini", allowed_roots=("models",), must_exist=True) + parser = configparser.ConfigParser(interpolation=None) + try: + parser.read(path, encoding="utf-8") + except configparser.Error as exc: + raise LabyricornMcpError("VALIDATION_FAILED", "The model definition is malformed.") from exc + fields: list[dict[str, Any]] = [] + for section in parser.sections(): + if not section.startswith("fields."): + continue + values = dict(parser.items(section)) + fields.append({"name": section.removeprefix("fields."), **values}) + return { + "model": model, + "name": parser.get("model", "name", fallback=model), + "label": parser.get("model", "label", fallback=""), + "hidden": parser.getboolean("model", "hidden", fallback=False), + "protected": parser.getboolean("model", "protected", fallback=False), + "children": dict(parser.items("children")) if parser.has_section("children") else None, + "fields": fields, + "source": self.guard.relative(path), + "sha256": _hash(path.read_bytes()), + } + + def _entry_result(self, model: str, entry: Any, repository: Any, *, include_body: bool) -> dict[str, Any]: + values = repository.values_for(entry.document) + if not include_body: + values.pop("body", None) + return { + "model": model, + "slug": entry.slug, + "path": self.guard.relative(entry.path), + "fields": values, + "sha256": _hash(entry.original_bytes), + } + + def list_content(self, filters: dict[str, Any]) -> dict[str, Any]: + requested = filters.get("model") + models = [requested] if requested else ["article", "blog", "project", "devlog"] + if any(model not in CONTENT_MODELS for model in models): + raise LabyricornMcpError("INVALID_CONTENT_MODEL", "Supported content models are article, blog, project, and devlog.") + items: list[dict[str, Any]] = [] + if "article" in models: + repository = self._article_repository() + items.extend(self._entry_result("article", item, repository, include_body=False) for item in repository.list_entries()) + if "blog" in models: + repository = self._blog_repository() + items.extend(self._entry_result("blog", item, repository, include_body=False) for item in repository.list_entries()) + if "project" in models or "devlog" in models: + for project in self.list_projects()["projects"]: + if "project" in models: + items.append({"model": "project", **project}) + if "devlog" in models: + try: + items.extend(self.list_project_devlogs(project["project_id"])["entries"]) + except LabyricornMcpError as exc: + if exc.code not in {"CONTENT_NOT_FOUND", "RUNTIME_MISSING"}: + raise + + tag = str(filters.get("tag") or "").strip() + project_id = str(filters.get("project") or "").strip() + slug = str(filters.get("slug") or "").strip() + date_value = str(filters.get("date") or "").strip() + search = str(filters.get("search") or "").strip().casefold() + + def matches(item: dict[str, Any]) -> bool: + fields = item.get("fields", {}) + tags = fields.get("tags", []) + if isinstance(tags, str): + tags = [part.strip() for part in re.split(r"[,\n]", tags) if part.strip()] + if tag and tag not in tags and tag not in fields.get("topic_tag_slugs", []): + return False + if project_id and item.get("project_id") != project_id: + return False + if slug and item.get("slug") != slug: + return False + if date_value and fields.get("date") != date_value: + return False + if search and search not in json.dumps(item, ensure_ascii=False).casefold(): + return False + return True + + filtered = [item for item in items if matches(item)] + return {"count": len(filtered), "items": filtered} + + def get_content(self, model: str, slug: str, project: str | None = None) -> dict[str, Any]: + try: + if model == "article": + repository = self._article_repository() + return self._entry_result(model, repository.load_entry(slug), repository, include_body=True) + if model == "blog": + repository = self._blog_repository() + return self._entry_result(model, repository.load_entry(slug), repository, include_body=True) + if model == "project": + return self.get_project(slug) + if model == "devlog": + if not project: + raise LabyricornMcpError("VALIDATION_FAILED", "The project argument is required for devlog content.") + return self.get_devlog(project, slug) + except LabyricornMcpError: + raise + except Exception as exc: + raise self._translate(exc, not_found=True) from exc + raise LabyricornMcpError("INVALID_CONTENT_MODEL", "Supported content models are article, blog, project, and devlog.") + + def _create_entry(self, model: str, slug: str, fields: dict[str, Any]) -> dict[str, Any]: + repository = self._article_repository() if model == "article" else self._blog_repository() + kicker = "Article" if model == "article" else "Blog" + values = { + "title": str(fields.get("title", "")), + "date": str(fields.get("date", date.today().isoformat())), + "updated": str(fields.get("updated", "")), + "author": str(fields.get("author", "")), + "tags": self._tag_input(fields.get("tags", "")), + "kicker": kicker, + "summary": str(fields.get("summary", "")), + "body": str(fields.get("body", "")), + "external_url": str(fields.get("external_url", "")), + } + if model == "article": + values["published_urls"] = self._lines_input(fields.get("published_urls", "")) + try: + entry = repository.create_entry(slug, values) + return self._entry_result(model, entry, repository, include_body=True) + except Exception as exc: + raise self._translate(exc) from exc + + def _update_entry(self, model: str, slug: str, fields: dict[str, Any], expected_hash: str) -> dict[str, Any]: + repository = self._article_repository() if model == "article" else self._blog_repository() + try: + entry = repository.load_entry(slug) + actual_hash = _hash(entry.original_bytes) + if actual_hash != expected_hash: + raise LabyricornMcpError( + "FILE_CHANGED", "The content changed after it was read; reload it before updating.", + {"expected_sha256": expected_hash, "actual_sha256": actual_hash}, + ) + values = repository.values_for(entry.document) + for key, value in fields.items(): + if key not in values: + raise LabyricornMcpError("VALIDATION_FAILED", f"Unsupported {model} field: {key}") + if key == "tags": + values[key] = self._tag_input(value) + elif key == "published_urls": + values[key] = self._lines_input(value) + else: + values[key] = str(value) + saved = repository.save_entry(entry, values) + return self._entry_result(model, saved, repository, include_body=True) + except LabyricornMcpError: + raise + except Exception as exc: + raise self._translate(exc, not_found=True) from exc + + @staticmethod + def _tag_input(value: Any) -> str: + if isinstance(value, list): + return ", ".join(str(item) for item in value) + return str(value) + + @staticmethod + def _lines_input(value: Any) -> str: + if isinstance(value, list): + return "\n".join(str(item) for item in value) + return str(value) + + def create_article(self, slug: str, fields: dict[str, Any]) -> dict[str, Any]: + return self._create_entry("article", slug, fields) + + def update_article(self, slug: str, fields: dict[str, Any], expected_hash: str) -> dict[str, Any]: + return self._update_entry("article", slug, fields, expected_hash) + + def create_blog_post(self, slug: str, fields: dict[str, Any]) -> dict[str, Any]: + return self._create_entry("blog", slug, fields) + + def update_blog_post(self, slug: str, fields: dict[str, Any], expected_hash: str) -> dict[str, Any]: + return self._update_entry("blog", slug, fields, expected_hash) + + def list_tags(self) -> dict[str, Any]: + try: + tags = self._tag_repository().list_tags() + except Exception as exc: + raise self._translate(exc) from exc + return { + "count": len(tags), + "tags": [ + {"slug": tag.slug, "title": tag.title, "summary": tag.summary, "sha256": _hash(tag.original_bytes)} + for tag in tags + ], + } + + def get_tag_usage(self, slug: str) -> dict[str, Any]: + usages: list[dict[str, str]] = [] + for model in ("article", "blog"): + repository = self._article_repository() if model == "article" else self._blog_repository() + for entry in repository.list_entries(): + tags = [part.strip() for part in re.split(r"[,\n]", entry.document.get("tags")) if part.strip()] + if slug in tags: + usages.append({"model": model, "slug": entry.slug}) + for project in self.list_projects()["projects"]: + try: + snapshot = self._cached_project_snapshot(project["project_id"]) + except LabyricornMcpError: + continue + project_record = snapshot["records"][".labyricorn/project/contents.lr"] + if slug in self._split(project_record.get("tags", "")): + usages.append({"model": "project", "slug": project["project_id"]}) + for path, record in snapshot["records"].items(): + if path.startswith(".labyricorn/devlog/") and path.count("/") == 3 and slug in self._split(record.get("tags", "")): + usages.append({"model": "devlog", "slug": Path(path).parent.name, "project": project["project_id"]}) + tracked = any(tag["slug"] == slug for tag in self.list_tags()["tags"]) + return {"slug": slug, "tracked": tracked, "count": len(usages), "usages": usages} + + def create_tag(self, slug: str, title: str, summary: str) -> dict[str, Any]: + try: + tag = self._tag_repository().create_tag(slug, title, summary) + except Exception as exc: + raise self._translate(exc) from exc + return {"slug": tag.slug, "title": tag.title, "summary": tag.summary, "sha256": _hash(tag.original_bytes)} + + def update_tag(self, slug: str, title: str, summary: str, expected_hash: str) -> dict[str, Any]: + repository = self._tag_repository() + try: + tag = repository.load_tag(slug) + actual = _hash(tag.original_bytes) + if actual != expected_hash: + raise LabyricornMcpError("FILE_CHANGED", "The tracked tag changed after it was read.", {"actual_sha256": actual}) + saved = repository.save_tag(tag, title, summary) + except LabyricornMcpError: + raise + except Exception as exc: + raise self._translate(exc, not_found=True) from exc + return {"slug": saved.slug, "title": saved.title, "summary": saved.summary, "sha256": _hash(saved.original_bytes)} + + def untrack_tag(self, slug: str, expected_hash: str) -> dict[str, Any]: + repository = self._tag_repository() + try: + tag = repository.load_tag(slug) + actual = _hash(tag.original_bytes) + if actual != expected_hash: + raise LabyricornMcpError("FILE_CHANGED", "The tracked tag changed after it was read.", {"actual_sha256": actual}) + usage = self.get_tag_usage(slug) + repository.untrack_tag(tag) + except LabyricornMcpError: + raise + except Exception as exc: + raise self._translate(exc, not_found=True) from exc + return {"slug": slug, "untracked": True, "preserved_content_references": usage["count"]} + + def list_projects(self) -> dict[str, Any]: + try: + sources = self._project_registry().list_sources() + except Exception as exc: + raise self._translate(exc) from exc + state = self._project_state() + projects = [] + for source in sources: + cached = state.get("projects", {}).get(source.project_id, {}) + projects.append( + { + "project_id": source.project_id, + "repository": source.web_url, + "branch": source.branch, + "featured_order": source.featured_order, + "cached_commit": cached.get("commit"), + "synchronized_at": cached.get("synchronized_at"), + "metadata": cached.get("metadata", {}), + } + ) + return {"count": len(projects), "projects": projects} + + def get_project(self, project_id: str) -> dict[str, Any]: + snapshot = self._cached_project_snapshot(project_id) + record = snapshot["records"][".labyricorn/project/contents.lr"] + return { + "model": "project", + "project_id": project_id, + "slug": project_id, + "fields": record, + "commit": snapshot["commit"], + "source": snapshot["source"], + "metadata": snapshot["state"].get("metadata", {}), + } + + def list_project_devlogs(self, project_id: str) -> dict[str, Any]: + snapshot = self._cached_project_snapshot(project_id) + entries = [] + for path, record in snapshot["records"].items(): + pure = Path(path.replace("/", "\\")) if sys.platform == "win32" else Path(path) + if not path.startswith(".labyricorn/devlog/") or path.count("/") != 3: + continue + slug = pure.parent.name + entries.append( + {"model": "devlog", "project_id": project_id, "slug": slug, "fields": record, "commit": snapshot["commit"]} + ) + entries.sort(key=lambda item: (item["fields"].get("date", ""), item["slug"]), reverse=True) + return {"project_id": project_id, "count": len(entries), "entries": entries} + + def get_devlog(self, project_id: str, slug: str) -> dict[str, Any]: + for entry in self.list_project_devlogs(project_id)["entries"]: + if entry["slug"] == slug: + return entry + raise LabyricornMcpError("CONTENT_NOT_FOUND", "The requested project devlog entry does not exist.") + + def list_project_sources(self) -> dict[str, Any]: + projects = self.list_projects()["projects"] + return { + "count": len(projects), + "sources": [ + {key: value for key, value in project.items() if key != "metadata"} + for project in projects + ], + } + + def inspect_project_source(self, project_id: str) -> dict[str, Any]: + snapshot = self._cached_project_snapshot(project_id) + return { + "project_id": project_id, + "source": snapshot["source"], + "cached_commit": snapshot["commit"], + "validated_files": sorted(snapshot["files"]), + "records": snapshot["records"], + "network_accessed": False, + } + + def validate_project_source(self, project_id: str) -> dict[str, Any]: + try: + source = self._project_registry().load_source(project_id) + readiness = self.modules["project"].check_devlog_readiness(source) + except Exception as exc: + raise self._translate(exc, not_found=True) from exc + return {"project_id": project_id, **asdict(readiness), "network_accessed": True, "repository_modified": False} + + def _project_state(self) -> dict[str, Any]: + path = self.guard.resolve(".cache/project-sources/state.json") + if not path.is_file(): + return {"version": 1, "projects": {}} + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise LabyricornMcpError("VALIDATION_FAILED", "The local project cache state is malformed.") from exc + + def _cached_project_snapshot(self, project_id: str) -> dict[str, Any]: + try: + source = self._project_registry().load_source(project_id) + except Exception as exc: + raise self._translate(exc, not_found=True) from exc + state = self._project_state().get("projects", {}).get(project_id) + if not state or not state.get("commit"): + raise LabyricornMcpError("CONTENT_NOT_FOUND", "No validated local snapshot is cached for this project.") + try: + import project_sources + except ModuleNotFoundError as exc: + raise LabyricornMcpError( + "RUNTIME_MISSING", "Inspecting cached project content requires the existing Lektor Python environment." + ) from exc + mirror = self.guard.resolve(f".cache/project-sources/repos/{project_id}.git", must_exist=True) + try: + files, records = project_sources.validate_snapshot(source.importer_source(), mirror, state["commit"]) + except Exception as exc: + raise LabyricornMcpError("VALIDATION_FAILED", f"The cached project snapshot failed validation: {exc}") from exc + return { + "source": {key: value for key, value in source.importer_source().items() if key != "api_url"}, + "commit": state["commit"], + "state": state, + "files": files, + "records": records, + } + + @staticmethod + def _split(value: str) -> list[str]: + return [part.strip() for part in re.split(r"[,\n]", value) if part.strip()] + + def list_design_files(self, area: str | None = None) -> dict[str, Any]: + roots = DESIGN_ROOTS + if area: + if area not in DESIGN_ROOTS: + raise LabyricornMcpError("PATH_NOT_ALLOWED", "Area must be templates or assets/static.") + roots = (area,) + files = [] + for root in roots: + base = self.guard.resolve(root, allowed_roots=DESIGN_ROOTS, must_exist=True) + for path in sorted(base.rglob("*")): + if not path.resolve().is_relative_to(self.guard.root): + raise LabyricornMcpError( + "PATH_OUTSIDE_REPOSITORY", "A design path resolves outside the repository." + ) + if path.is_file() and not path.is_symlink(): + files.append( + { + "path": self.guard.relative(path), + "size": path.stat().st_size, + "sha256": _hash(path.read_bytes()), + "text_readable": path.suffix.casefold() in DESIGN_SUFFIXES, + } + ) + return {"count": len(files), "files": files} + + def get_design_file(self, relative: str) -> dict[str, Any]: + path = self.guard.resolve(relative, allowed_roots=DESIGN_ROOTS, must_exist=True) + if path.suffix.casefold() not in DESIGN_SUFFIXES or not path.is_file(): + raise LabyricornMcpError("PATH_NOT_ALLOWED", "Only allowlisted text design files can be read.") + try: + payload = path.read_bytes() + content = payload.decode("utf-8") + except (OSError, UnicodeError) as exc: + raise LabyricornMcpError("VALIDATION_FAILED", "The design file is not readable UTF-8 text.") from exc + return {"path": self.guard.relative(path), "content": content, "sha256": _hash(payload)} + + def update_design_file(self, relative: str, content: str, expected_hash: str) -> dict[str, Any]: + path = self.guard.resolve(relative, allowed_roots=DESIGN_ROOTS, must_exist=True, reject_links=True) + if path.suffix.casefold() not in DESIGN_SUFFIXES or not path.is_file(): + raise LabyricornMcpError("PATH_NOT_ALLOWED", "Only existing allowlisted text design files can be updated.") + current = path.read_bytes() + actual = _hash(current) + if actual != expected_hash: + raise LabyricornMcpError("FILE_CHANGED", "The design file changed after it was read.", {"actual_sha256": actual}) + payload = content.encode("utf-8") + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", prefix=f".{path.name}.labyricorn-mcp-", dir=path.parent, delete=False + ) as output: + temporary = Path(output.name) + output.write(payload) + output.flush() + os.fsync(output.fileno()) + if path.read_bytes() != current: + temporary.unlink(missing_ok=True) + raise LabyricornMcpError( + "FILE_CHANGED", "The design file changed while the update was being prepared." + ) + os.chmod(temporary, path.stat().st_mode) + os.replace(temporary, path) + except LabyricornMcpError: + raise + except OSError as exc: + if temporary is not None: + temporary.unlink(missing_ok=True) + raise LabyricornMcpError("VALIDATION_FAILED", "The design file could not be updated.") from exc + return {"path": self.guard.relative(path), "sha256": _hash(payload), "changed": payload != current} diff --git a/labyricorn_mcp/errors.py b/labyricorn_mcp/errors.py new file mode 100644 index 0000000..9dc3a0c --- /dev/null +++ b/labyricorn_mcp/errors.py @@ -0,0 +1,27 @@ +"""Structured errors safe to return to MCP clients.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class LabyricornMcpError(Exception): + code: str + message: str + details: dict[str, Any] = field(default_factory=dict) + + def __str__(self) -> str: + return self.message + + def as_dict(self) -> dict[str, Any]: + result: dict[str, Any] = {"code": self.code, "message": self.message} + if self.details: + result["details"] = self.details + return result + + +def require(condition: bool, code: str, message: str, **details: Any) -> None: + if not condition: + raise LabyricornMcpError(code, message, details) diff --git a/labyricorn_mcp/gitops.py b/labyricorn_mcp/gitops.py new file mode 100644 index 0000000..d2b2382 --- /dev/null +++ b/labyricorn_mcp/gitops.py @@ -0,0 +1,269 @@ +"""Deliberately constrained Git operations for the local checkout.""" + +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import re +import sys +from typing import Any + +from .errors import LabyricornMcpError +from .safety import RepositoryGuard + + +PROHIBITED_COMMIT_PARTS = {".git", ".cache", ".lektor", "build", "dist"} +SENSITIVE_NAMES = {".env", ".netrc", "id_rsa", "id_ed25519"} + + +class GitOperations: + def __init__(self, guard: RepositoryGuard): + self.guard = guard + scripts = str(guard.root / "scripts") + if scripts not in sys.path: + sys.path.insert(0, scripts) + import git_repo + + self.module = git_repo + self.repository = git_repo.GitRepository(guard.root) + + def _run(self, *args: str, timeout: int = 30, allow_prompt: bool = False) -> Any: + try: + result = self.module._run_git( + self.guard.root, *args, check=False, timeout=timeout, allow_prompt=allow_prompt + ) + except Exception as exc: + raise LabyricornMcpError("GIT_ERROR", str(exc)) from exc + if result.returncode != 0: + detail = self.module._error_detail(result) + detail = re.sub(r"(https?://)[^/\s@]+@", r"\1", detail) + raise LabyricornMcpError("GIT_ERROR", detail) + return result + + def _status_text(self) -> str: + return self._run("status", "--porcelain=v1", "--untracked-files=normal").stdout.rstrip() + + def status(self) -> dict[str, Any]: + text = self._status_text() + branch = self._run("symbolic-ref", "--quiet", "--short", "HEAD").stdout.strip() + commit = self._run("rev-parse", "HEAD").stdout.strip() + upstream_result = self.module._run_git( + self.guard.root, + "rev-parse", + "--abbrev-ref", + "--symbolic-full-name", + "@{upstream}", + check=False, + ) + upstream = upstream_result.stdout.strip() if upstream_result.returncode == 0 else None + entries = self._parse_status(text) + status_snapshot = json.dumps( + { + "branch": branch, + "commit": commit, + "upstream": upstream, + "porcelain": text, + "tracked_diff_sha256": hashlib.sha256( + self._run("diff", "--binary", "HEAD").stdout.encode("utf-8") + ).hexdigest(), + "untracked_fingerprint": self._untracked_fingerprint(entries), + }, + sort_keys=True, + separators=(",", ":"), + ) + return { + "branch": branch, + "commit": commit, + "upstream": upstream, + "clean": not entries, + "changes": entries, + "status_sha256": hashlib.sha256(status_snapshot.encode("utf-8")).hexdigest(), + "main_push_warning": ( + "Pushing main triggers the repository's external site update workflow. " + "The MCP performs only a Git push." + if branch == "main" + else None + ), + } + + def _untracked_fingerprint(self, entries: list[dict[str, str]]) -> list[tuple[str, int, int]]: + fingerprint: list[tuple[str, int, int]] = [] + for entry in entries: + if entry["index"] != "?": + continue + relative = entry["path"].rstrip("/") + candidate = self.guard.resolve(relative) + if candidate.is_dir() and (candidate / ".git").exists(): + stat = candidate.stat() + fingerprint.append((entry["path"], stat.st_size, stat.st_mtime_ns)) + continue + paths = [candidate] + if candidate.is_dir(): + paths = [] + for directory, names, filenames in os.walk(candidate): + for name in names: + linked = Path(directory) / name + if (linked.is_symlink() or getattr(linked, "is_junction", lambda: False)()) and not linked.resolve().is_relative_to(self.guard.root): + raise LabyricornMcpError( + "PATH_OUTSIDE_REPOSITORY", "An untracked linked path resolves outside the repository." + ) + names[:] = [name for name in names if name not in {".git", "__pycache__"}] + paths.extend(Path(directory) / filename for filename in filenames) + if len(paths) > 2_000: + break + for path in sorted(paths): + if (path.is_symlink() or getattr(path, "is_junction", lambda: False)()) and not path.resolve().is_relative_to(self.guard.root): + raise LabyricornMcpError( + "PATH_OUTSIDE_REPOSITORY", "An untracked linked path resolves outside the repository." + ) + if not path.is_file(): + continue + stat = path.stat() + fingerprint.append((self.guard.relative(path), stat.st_size, stat.st_mtime_ns)) + return fingerprint + + @staticmethod + def _parse_status(text: str) -> list[dict[str, str]]: + entries = [] + for line in text.splitlines(): + if len(line) < 4: + continue + raw_path = line[3:] + if " -> " in raw_path: + original, path = raw_path.split(" -> ", 1) + entries.append({"index": line[0], "worktree": line[1], "path": path.strip('"'), "original_path": original.strip('"')}) + else: + entries.append({"index": line[0], "worktree": line[1], "path": raw_path.strip('"')}) + return entries + + def diff(self, *, staged: bool = False, path: str | None = None, max_chars: int = 100_000) -> dict[str, Any]: + args = ["diff"] + if staged: + args.append("--cached") + if path: + safe = self.guard.relative(self.guard.resolve(path)) + args.extend(("--", safe)) + output = self._run(*args).stdout + truncated = len(output) > max_chars + return {"staged": staged, "path": path, "diff": output[:max_chars], "truncated": truncated} + + def log(self, limit: int = 20) -> dict[str, Any]: + if not isinstance(limit, int) or not 1 <= limit <= 100: + raise LabyricornMcpError("VALIDATION_FAILED", "Git log limit must be between 1 and 100.") + separator = "\x1f" + output = self._run( + "log", f"-{limit}", f"--format=%H{separator}%h{separator}%aI{separator}%an{separator}%s" + ).stdout + commits = [] + for line in output.splitlines(): + parts = line.split(separator, 4) + if len(parts) == 5: + commits.append(dict(zip(("commit", "short_commit", "date", "author", "subject"), parts))) + return {"count": len(commits), "commits": commits} + + def pull(self, expected_status_hash: str) -> dict[str, Any]: + before = self.status() + self._expect_status(before, expected_status_hash) + if not before["clean"]: + raise LabyricornMcpError("WORKTREE_CONFLICT", "A fast-forward pull requires a clean working tree.") + try: + state = self.repository.refresh() + updated = self.repository.pull(state) + except Exception as exc: + raise LabyricornMcpError("GIT_ERROR", str(exc)) from exc + return {"pulled": True, "branch": updated.branch, "commit": self._run("rev-parse", "HEAD").stdout.strip(), "status": self.status()} + + def commit(self, message: str, paths: list[str], expected_status_hash: str) -> dict[str, Any]: + before = self.status() + self._expect_status(before, expected_status_hash) + message = " ".join(str(message).splitlines()).strip() + if not message: + raise LabyricornMcpError("VALIDATION_FAILED", "A non-empty commit message is required.") + if not isinstance(paths, list) or not paths: + raise LabyricornMcpError("VALIDATION_FAILED", "At least one reviewed changed path is required.") + changed = {entry["path"] for entry in before["changes"]} + untracked_directories = { + entry["path"] for entry in before["changes"] + if entry["index"] == "?" and entry["path"].endswith("/") + } + selected: list[str] = [] + for value in paths: + path = self.guard.relative(self.guard.resolve(str(value), reject_links=True)) + parts = set(Path(path).parts) + if parts & PROHIBITED_COMMIT_PARTS or Path(path).name.casefold() in SENSITIVE_NAMES: + raise LabyricornMcpError("PATH_NOT_ALLOWED", f"The path is not eligible for MCP commits: {path}") + if self._inside_nested_repository(path): + raise LabyricornMcpError("PATH_NOT_ALLOWED", f"Nested repositories are outside this MCP's commit scope: {path}") + candidate = self.guard.root / path + if candidate.exists() and not candidate.is_file(): + raise LabyricornMcpError("PATH_NOT_ALLOWED", "Commit paths must identify individual files.") + summarized = any(path.startswith(directory) for directory in untracked_directories) + if path not in changed and not summarized: + raise LabyricornMcpError("WORKTREE_CONFLICT", f"The reviewed path is not a current working-tree change: {path}") + if path not in selected: + selected.append(path) + unrelated_staged = [ + entry["path"] for entry in before["changes"] if entry["index"] not in {" ", "?"} and entry["path"] not in selected + ] + if unrelated_staged: + raise LabyricornMcpError( + "WORKTREE_CONFLICT", "Unrelated staged changes must be reviewed outside this commit.", + {"staged_paths": unrelated_staged}, + ) + self._run("add", "--", *selected) + staged = [line for line in self._run("diff", "--cached", "--name-only").stdout.splitlines() if line] + if set(staged) != set(selected): + raise LabyricornMcpError( + "WORKTREE_CONFLICT", "Staging did not produce exactly the reviewed path set; no commit was created.", + {"reviewed_paths": selected, "staged_paths": staged}, + ) + self._run("commit", "-m", message, "--", *selected, timeout=60) + commit = self._run("rev-parse", "HEAD").stdout.strip() + return {"committed": True, "commit": commit, "message": message, "paths": selected, "status": self.status()} + + def _inside_nested_repository(self, relative: str) -> bool: + current = self.guard.root + for part in Path(relative).parts[:-1]: + current /= part + if current != self.guard.root and (current / ".git").exists(): + return True + return False + + def push(self, expected_head: str, acknowledge_main_update: bool = False) -> dict[str, Any]: + before = self.status() + if before["commit"] != expected_head: + raise LabyricornMcpError( + "WORKTREE_CONFLICT", "HEAD changed after it was reviewed; inspect Git status again.", + {"actual_head": before["commit"]}, + ) + if before["branch"] == "main" and not acknowledge_main_update: + raise LabyricornMcpError( + "MAIN_PUSH_ACK_REQUIRED", + "Pushing main triggers the repository's external site update workflow. Set acknowledge_main_update=true to confirm the Git push.", + ) + try: + state = self.repository.refresh() + if state.branch != before["branch"]: + raise LabyricornMcpError("WORKTREE_CONFLICT", "The branch changed while preparing the push.") + updated = self.repository.push(state) + except LabyricornMcpError: + raise + except Exception as exc: + raise LabyricornMcpError("GIT_ERROR", str(exc)) from exc + return { + "pushed": True, + "branch": updated.branch, + "commit": expected_head, + "external_update_may_run": updated.branch == "main", + "production_contacted_by_mcp": False, + } + + @staticmethod + def _expect_status(status: dict[str, Any], expected: str) -> None: + if status["status_sha256"] != expected: + raise LabyricornMcpError( + "WORKTREE_CONFLICT", "The working tree changed after it was reviewed; inspect Git status again.", + {"actual_status_sha256": status["status_sha256"]}, + ) diff --git a/labyricorn_mcp/safety.py b/labyricorn_mcp/safety.py new file mode 100644 index 0000000..05deb86 --- /dev/null +++ b/labyricorn_mcp/safety.py @@ -0,0 +1,126 @@ +"""Repository discovery and filesystem containment.""" + +from __future__ import annotations + +import os +from pathlib import Path, PurePath +import subprocess + +from .errors import LabyricornMcpError + + +class RepositoryGuard: + """Resolve every path against one verified Labyricorn Git checkout.""" + + REQUIRED_PATHS = ( + "Labyricorn.lektorproject", + "content/articles/contents.lr", + "content/blog/contents.lr", + "content/projects/contents.lr", + "content/tags/contents.lr", + "models/entry.ini", + "models/project.ini", + "models/devlog-entry.ini", + "configs/project-sources.ini", + "scripts/build_with_projects.py", + ) + + def __init__(self, root: Path): + try: + self.root = root.expanduser().resolve(strict=True) + except OSError as exc: + raise LabyricornMcpError( + "REPOSITORY_NOT_FOUND", "The configured repository directory does not exist." + ) from exc + if not self.root.is_dir(): + raise LabyricornMcpError( + "REPOSITORY_NOT_FOUND", "The configured repository path is not a directory." + ) + self._validate_repository() + + @classmethod + def discover(cls, configured: str | None = None) -> "RepositoryGuard": + selected = configured or os.environ.get("LABYRICORN_REPOSITORY") or os.getcwd() + return cls(Path(selected)) + + def _validate_repository(self) -> None: + missing = [relative for relative in self.REQUIRED_PATHS if not (self.root / relative).is_file()] + if missing: + raise LabyricornMcpError( + "INVALID_LABYRICORN_REPOSITORY", + "The selected directory is not a compatible Labyricorn repository.", + {"missing": missing}, + ) + escaped = [ + relative + for relative in self.REQUIRED_PATHS + if not (self.root / relative).resolve().is_relative_to(self.root) + ] + if escaped: + raise LabyricornMcpError( + "INVALID_LABYRICORN_REPOSITORY", + "Required Labyricorn paths must not resolve outside the repository.", + {"escaped": escaped}, + ) + try: + result = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], + cwd=self.root, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=15, + check=False, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise LabyricornMcpError( + "INVALID_LABYRICORN_REPOSITORY", "The selected directory is not an accessible Git repository." + ) from exc + try: + git_root = Path(result.stdout.strip()).resolve(strict=True) + except OSError: + git_root = Path() + if result.returncode != 0 or git_root != self.root: + raise LabyricornMcpError( + "INVALID_LABYRICORN_REPOSITORY", + "The selected directory must be the root of the Labyricorn Git repository.", + ) + + def resolve( + self, + relative: str, + *, + allowed_roots: tuple[str, ...] = (), + must_exist: bool = False, + reject_links: bool = False, + ) -> Path: + if not isinstance(relative, str) or not relative.strip(): + raise LabyricornMcpError("INVALID_PATH", "A non-empty repository-relative path is required.") + raw = PurePath(relative) + if raw.is_absolute() or raw.drive or ".." in raw.parts: + raise LabyricornMcpError("PATH_OUTSIDE_REPOSITORY", "Only repository-relative paths are allowed.") + candidate = self.root.joinpath(*raw.parts) + try: + resolved = candidate.resolve(strict=must_exist) + except OSError as exc: + raise LabyricornMcpError("CONTENT_NOT_FOUND", "The requested repository path does not exist.") from exc + if not resolved.is_relative_to(self.root): + raise LabyricornMcpError("PATH_OUTSIDE_REPOSITORY", "The requested path resolves outside the repository.") + if allowed_roots and not any( + resolved.is_relative_to((self.root / allowed).resolve()) for allowed in allowed_roots + ): + raise LabyricornMcpError( + "PATH_NOT_ALLOWED", "The requested path is outside the allowlisted repository areas.", + {"allowed_roots": list(allowed_roots)}, + ) + if reject_links: + current = self.root + for part in raw.parts: + current = current / part + if current.exists() and (current.is_symlink() or getattr(current, "is_junction", lambda: False)()): + raise LabyricornMcpError("PATH_LINK_REJECTED", "Linked paths cannot be modified.") + return resolved + + def relative(self, path: Path) -> str: + return path.resolve().relative_to(self.root).as_posix() diff --git a/labyricorn_mcp/server.py b/labyricorn_mcp/server.py new file mode 100644 index 0000000..5575625 --- /dev/null +++ b/labyricorn_mcp/server.py @@ -0,0 +1,151 @@ +"""Dependency-free MCP stdio transport for the local Labyricorn repository.""" + +from __future__ import annotations + +import json +import logging +import sys +from typing import Any, TextIO + +from . import __version__ +from .errors import LabyricornMcpError +from .safety import RepositoryGuard +from .tools import ToolRegistry + + +SUPPORTED_PROTOCOLS = ("2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05") +LATEST_PROTOCOL = SUPPORTED_PROTOCOLS[0] + + +class StdioMcpServer: + def __init__(self, guard: RepositoryGuard, *, input_stream: TextIO | None = None, output_stream: TextIO | None = None): + self.guard = guard + self.registry = ToolRegistry(guard) + self.input = input_stream or sys.stdin + self.output = output_stream or sys.stdout + self.initialized = False + self.client_initialized = False + self.protocol_version: str | None = None + self.log = logging.getLogger("labyricorn_mcp") + + def serve_forever(self) -> int: + self.log.info("server started for repository %s", self.guard.root) + for raw_line in self.input: + line = raw_line.rstrip("\r\n") + if not line: + continue + try: + message = json.loads(line) + except json.JSONDecodeError: + self._write_error(None, -32700, "Parse error") + continue + if not isinstance(message, dict) or isinstance(message.get("id"), (dict, list)): + self._write_error(message.get("id") if isinstance(message, dict) else None, -32600, "Invalid Request") + continue + self._handle(message) + self.log.info("server stopped after stdin closed") + return 0 + + def _handle(self, message: dict[str, Any]) -> None: + method = message.get("method") + request_id = message.get("id") + is_notification = "id" not in message + if message.get("jsonrpc") != "2.0" or not isinstance(method, str): + if not is_notification: + self._write_error(request_id, -32600, "Invalid Request") + return + params = message.get("params", {}) + if not isinstance(params, dict): + if not is_notification: + self._write_error(request_id, -32602, "Invalid params") + return + + if is_notification: + if method == "notifications/initialized" and self.initialized: + self.client_initialized = True + return + + if method == "initialize": + self._initialize(request_id, params) + return + if not self.initialized: + self._write_error(request_id, -32002, "Server is not initialized") + return + if method == "ping": + self._write_result(request_id, {}) + elif method == "tools/list": + self._write_result(request_id, {"tools": self.registry.tools}) + elif method == "tools/call": + self._call_tool(request_id, params) + else: + self._write_error(request_id, -32601, "Method not found") + + def _initialize(self, request_id: Any, params: dict[str, Any]) -> None: + if self.initialized: + self._write_error(request_id, -32600, "Server is already initialized") + return + requested = params.get("protocolVersion") + self.protocol_version = requested if requested in SUPPORTED_PROTOCOLS else LATEST_PROTOCOL + self.initialized = True + self._write_result( + request_id, + { + "protocolVersion": self.protocol_version, + "capabilities": {"tools": {"listChanged": False}}, + "serverInfo": {"name": "labyricorn-mcp", "version": __version__}, + "instructions": ( + "This server manages only the configured local Labyricorn Git checkout. " + "It has no production administration capability. Commits and pushes are separate explicit tools." + ), + }, + ) + + def _call_tool(self, request_id: Any, params: dict[str, Any]) -> None: + name = params.get("name") + if not isinstance(name, str): + self._write_error(request_id, -32602, "Tool name is required") + return + self.log.info("tool invocation: %s", name) + try: + result = self.registry.call(name, params.get("arguments")) + except LabyricornMcpError as exc: + self.log.warning("tool rejected: %s code=%s", name, exc.code) + payload = {"error": exc.as_dict()} + self._write_result( + request_id, + { + "content": [{"type": "text", "text": json.dumps(payload, ensure_ascii=False)}], + "structuredContent": payload, + "isError": True, + }, + ) + except Exception: + self.log.exception("unexpected tool failure: %s", name) + payload = {"error": {"code": "INTERNAL_ERROR", "message": "The tool failed unexpectedly; see local stderr diagnostics."}} + self._write_result( + request_id, + { + "content": [{"type": "text", "text": json.dumps(payload)}], + "structuredContent": payload, + "isError": True, + }, + ) + else: + self._write_result( + request_id, + { + "content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, default=str)}], + "structuredContent": result, + "isError": False, + }, + ) + + def _write_result(self, request_id: Any, result: dict[str, Any]) -> None: + self._write({"jsonrpc": "2.0", "id": request_id, "result": result}) + + def _write_error(self, request_id: Any, code: int, message: str) -> None: + self._write({"jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message}}) + + def _write(self, message: dict[str, Any]) -> None: + self.output.write(json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n") + self.output.flush() diff --git a/labyricorn_mcp/tools.py b/labyricorn_mcp/tools.py new file mode 100644 index 0000000..4acd5af --- /dev/null +++ b/labyricorn_mcp/tools.py @@ -0,0 +1,219 @@ +"""MCP tool declarations and dispatch.""" + +from __future__ import annotations + +import configparser +import json +from pathlib import Path +from typing import Any, Callable + +from .domain import LabyricornDomain +from .errors import LabyricornMcpError +from .gitops import GitOperations +from .safety import RepositoryGuard +from .validation import ValidationOperations + + +Handler = Callable[[dict[str, Any]], dict[str, Any]] + + +def obj(properties: dict[str, Any] | None = None, required: list[str] | None = None) -> dict[str, Any]: + schema: dict[str, Any] = {"type": "object", "properties": properties or {}, "additionalProperties": False} + if required: + schema["required"] = required + return schema + + +def string(description: str = "") -> dict[str, Any]: + result: dict[str, Any] = {"type": "string"} + if description: + result["description"] = description + return result + + +def boolean(description: str = "") -> dict[str, Any]: + result: dict[str, Any] = {"type": "boolean"} + if description: + result["description"] = description + return result + + +def integer(description: str = "") -> dict[str, Any]: + result: dict[str, Any] = {"type": "integer"} + if description: + result["description"] = description + return result + + +def string_array(description: str = "") -> dict[str, Any]: + result: dict[str, Any] = {"type": "array", "items": {"type": "string"}} + if description: + result["description"] = description + return result + + +class ToolRegistry: + def __init__(self, guard: RepositoryGuard): + self.guard = guard + self.domain = LabyricornDomain(guard) + self.git = GitOperations(guard) + self.validation = ValidationOperations(guard, self.git) + self._handlers: dict[str, Handler] = {} + self._tools: list[dict[str, Any]] = [] + self._register_tools() + + def _add( + self, + name: str, + description: str, + schema: dict[str, Any], + handler: Handler, + *, + read_only: bool, + destructive: bool = False, + idempotent: bool = False, + open_world: bool = False, + ) -> None: + self._handlers[name] = handler + self._tools.append( + { + "name": name, + "title": name.replace("_", " ").title(), + "description": description, + "inputSchema": schema, + "annotations": { + "readOnlyHint": read_only, + "destructiveHint": destructive, + "idempotentHint": idempotent, + "openWorldHint": open_world, + }, + } + ) + + @property + def tools(self) -> list[dict[str, Any]]: + return list(self._tools) + + def call(self, name: str, arguments: dict[str, Any] | None) -> dict[str, Any]: + handler = self._handlers.get(name) + if handler is None: + raise LabyricornMcpError("TOOL_NOT_FOUND", f"Unknown Labyricorn MCP tool: {name}") + if arguments is None: + arguments = {} + if not isinstance(arguments, dict): + raise LabyricornMcpError("VALIDATION_FAILED", "Tool arguments must be an object.") + tool = next(item for item in self._tools if item["name"] == name) + self._validate_arguments(tool["inputSchema"], arguments) + return handler(arguments) + + @staticmethod + def _validate_arguments(schema: dict[str, Any], arguments: dict[str, Any]) -> None: + properties = schema.get("properties", {}) + unknown = sorted(set(arguments) - set(properties)) + if unknown: + raise LabyricornMcpError("VALIDATION_FAILED", "Unknown tool arguments were supplied.", {"unknown": unknown}) + missing = [name for name in schema.get("required", []) if name not in arguments] + if missing: + raise LabyricornMcpError("VALIDATION_FAILED", "Required tool arguments are missing.", {"missing": missing}) + type_map = { + "string": str, + "boolean": bool, + "integer": int, + "object": dict, + "array": list, + } + for name, value in arguments.items(): + expected_name = properties[name].get("type") + expected = type_map.get(expected_name) + if expected and (not isinstance(value, expected) or expected is int and isinstance(value, bool)): + raise LabyricornMcpError("VALIDATION_FAILED", f"Argument {name!r} must be {expected_name}.") + if expected is list and properties[name].get("items", {}).get("type") == "string": + if not all(isinstance(item, str) for item in value): + raise LabyricornMcpError("VALIDATION_FAILED", f"Argument {name!r} must contain only strings.") + + def _site_info(self) -> dict[str, Any]: + project_file = self.guard.root / "Labyricorn.lektorproject" + parser = configparser.ConfigParser(interpolation=None) + parser.read(project_file, encoding="utf-8") + status = self.git.status() + return { + "repository_root": str(self.guard.root), + "branch": status["branch"], + "commit": status["commit"], + "working_tree_clean": status["clean"], + "changed_paths": [change["path"] for change in status["changes"]], + "project": dict(parser.items("project")), + "project_file": project_file.name, + "models": sorted(path.stem for path in (self.guard.root / "models").glob("*.ini")), + "remote_project_sources": self.domain.list_project_sources()["count"], + "scope": "local repository only", + "production_access": False, + } + + def _mutation(self, operation: Callable[[], dict[str, Any]]) -> dict[str, Any]: + before = self.git.status() + result = operation() + after = self.git.status() + result["worktree_context"] = { + "before_status_sha256": before["status_sha256"], + "before_changed_paths": [item["path"] for item in before["changes"]], + "after_status_sha256": after["status_sha256"], + "after_changed_paths": [item["path"] for item in after["changes"]], + } + return result + + def _register_tools(self) -> None: + empty = obj() + fields_schema = {"type": "object", "description": "Model field values to create or update."} + self._add("site_info", "Describe the configured local Labyricorn repository; this is not production status.", empty, lambda _a: self._site_info(), read_only=True) + self._add("site_structure", "Describe the local content, model, design, build, and remote-project organization.", empty, lambda _a: self.domain.site_structure(), read_only=True) + self._add("describe_model", "Read a Lektor model definition from models/*.ini.", obj({"model": string()}, ["model"]), lambda a: self.domain.describe_model(a["model"]), read_only=True) + self._add( + "list_content", + "List article, blog, project, and devlog content with simple optional filters.", + obj({key: string() for key in ("model", "tag", "project", "date", "slug", "search")}), + lambda a: self.domain.list_content(a), + read_only=True, + ) + self._add( + "get_content", + "Get one article, blog post, project, or project devlog record.", + obj({"model": string(), "slug": string(), "project": string()}, ["model", "slug"]), + lambda a: self.domain.get_content(a["model"], a["slug"], a.get("project")), + read_only=True, + ) + for name, description, handler in ( + ("create_article", "Create a native article entry in content/articles.", lambda a: self._mutation(lambda: self.domain.create_article(a["slug"], a["fields"]))), + ("create_blog_post", "Create a native blog entry in content/blog.", lambda a: self._mutation(lambda: self.domain.create_blog_post(a["slug"], a["fields"]))), + ): + self._add(name, description, obj({"slug": string(), "fields": fields_schema}, ["slug", "fields"]), handler, read_only=False) + for name, description, handler in ( + ("update_article", "Update an article only if its previously returned SHA-256 still matches.", lambda a: self._mutation(lambda: self.domain.update_article(a["slug"], a["fields"], a["expected_sha256"]))), + ("update_blog_post", "Update a blog post only if its previously returned SHA-256 still matches.", lambda a: self._mutation(lambda: self.domain.update_blog_post(a["slug"], a["fields"], a["expected_sha256"]))), + ): + self._add(name, description, obj({"slug": string(), "fields": fields_schema, "expected_sha256": string()}, ["slug", "fields", "expected_sha256"]), handler, read_only=False, idempotent=True) + self._add("list_tags", "List tracked tags from content/tags.", empty, lambda _a: self.domain.list_tags(), read_only=True) + self._add("get_tag_usage", "Find article, blog, cached project, and cached devlog uses of a tag slug.", obj({"slug": string()}, ["slug"]), lambda a: self.domain.get_tag_usage(a["slug"]), read_only=True) + self._add("create_tag", "Create a tracked tag without rewriting existing content references.", obj({"slug": string(), "title": string(), "summary": string()}, ["slug", "title", "summary"]), lambda a: self._mutation(lambda: self.domain.create_tag(a["slug"], a["title"], a["summary"])), read_only=False) + self._add("update_tag", "Update tracked-tag display data if its SHA-256 still matches; slug renaming is not supported.", obj({"slug": string(), "title": string(), "summary": string(), "expected_sha256": string()}, ["slug", "title", "summary", "expected_sha256"]), lambda a: self._mutation(lambda: self.domain.update_tag(a["slug"], a["title"], a["summary"], a["expected_sha256"])), read_only=False, idempotent=True) + self._add("untrack_tag", "Remove only a tracked tag record, preserving every content reference.", obj({"slug": string(), "expected_sha256": string()}, ["slug", "expected_sha256"]), lambda a: self._mutation(lambda: self.domain.untrack_tag(a["slug"], a["expected_sha256"])), read_only=False, destructive=True) + self._add("list_projects", "List allowlisted remote projects and local last-known-good metadata.", empty, lambda _a: self.domain.list_projects(), read_only=True) + self._add("get_project", "Read a project record from its validated local cached mirror.", obj({"project_id": string()}, ["project_id"]), lambda a: self.domain.get_project(a["project_id"]), read_only=True) + self._add("list_project_devlogs", "List devlog records from a project's validated local cached mirror.", obj({"project_id": string()}, ["project_id"]), lambda a: self.domain.list_project_devlogs(a["project_id"]), read_only=True) + self._add("get_devlog", "Read one devlog record from a validated local cached mirror.", obj({"project_id": string(), "slug": string()}, ["project_id", "slug"]), lambda a: self.domain.get_devlog(a["project_id"], a["slug"]), read_only=True) + self._add("list_project_sources", "List the repository's allowlisted public project sources.", empty, lambda _a: self.domain.list_project_sources(), read_only=True) + self._add("inspect_project_source", "Validate and inspect an existing local cached project snapshot without network access.", obj({"project_id": string()}, ["project_id"]), lambda a: self.domain.inspect_project_source(a["project_id"]), read_only=True) + self._add("validate_project_source", "Run the existing read-only public-provider and disposable-mirror validation for one allowlisted source.", obj({"project_id": string()}, ["project_id"]), lambda a: self.domain.validate_project_source(a["project_id"]), read_only=True, open_world=True) + self._add("list_design_files", "List allowlisted template and static text design files.", obj({"area": string()}), lambda a: self.domain.list_design_files(a.get("area")), read_only=True) + self._add("get_design_file", "Read an allowlisted template or static text design file with a concurrency hash.", obj({"path": string()}, ["path"]), lambda a: self.domain.get_design_file(a["path"]), read_only=True) + self._add("update_design_file", "Update an existing allowlisted template or static text file only if its hash still matches.", obj({"path": string(), "content": string(), "expected_sha256": string()}, ["path", "content", "expected_sha256"]), lambda a: self._mutation(lambda: self.domain.update_design_file(a["path"], a["content"], a["expected_sha256"])), read_only=False, idempotent=True) + self._add("build_site", "Run the repository's isolated build_with_projects.py validation; output is temporary and production is not contacted.", empty, lambda _a: self.validation.build_site(), read_only=False, idempotent=True, open_world=True) + self._add("run_tests", "Run the repository unittest suite with the server's Python runtime.", empty, lambda _a: self.validation.run_tests(), read_only=False, idempotent=True) + self._add("check_diff", "Run git diff --check and report changed files.", empty, lambda _a: self.validation.check_diff(), read_only=True) + self._add("validate_changes", "Consolidate the supported local build, tests, and diff checks.", empty, lambda _a: self.validation.validate_changes(), read_only=False, idempotent=True, open_world=True) + self._add("git_status", "Read branch, HEAD, upstream, working-tree changes, and a status concurrency hash without fetching.", empty, lambda _a: self.git.status(), read_only=True) + self._add("git_diff", "Read a constrained Git diff, optionally staged or limited to one repository path.", obj({"staged": boolean(), "path": string()}), lambda a: self.git.diff(staged=a.get("staged", False), path=a.get("path")), read_only=True) + self._add("git_log", "Read recent local Git history (maximum 100 commits).", obj({"limit": integer()}), lambda a: self.git.log(a.get("limit", 20)), read_only=True) + self._add("git_pull", "Perform only a clean, reviewed, fast-forward pull through the existing Git safety helper.", obj({"expected_status_sha256": string()}, ["expected_status_sha256"]), lambda a: self.git.pull(a["expected_status_sha256"]), read_only=False, open_world=True) + self._add("commit_changes", "Commit only explicitly listed reviewed paths; never pushes and rejects unrelated staged changes.", obj({"message": string(), "paths": string_array(), "expected_status_sha256": string()}, ["message", "paths", "expected_status_sha256"]), lambda a: self.git.commit(a["message"], a["paths"], a["expected_status_sha256"]), read_only=False) + self._add("push_changes", "Push the reviewed HEAD normally to its configured origin upstream; force push is impossible. Pushing main requires acknowledgement.", obj({"expected_head": string(), "acknowledge_main_update": boolean()}, ["expected_head"]), lambda a: self.git.push(a["expected_head"], a.get("acknowledge_main_update", False)), read_only=False, open_world=True) diff --git a/labyricorn_mcp/validation.py b/labyricorn_mcp/validation.py new file mode 100644 index 0000000..5437476 --- /dev/null +++ b/labyricorn_mcp/validation.py @@ -0,0 +1,149 @@ +"""Structured wrappers around the repository's supported validation commands.""" + +from __future__ import annotations + +from pathlib import Path +import shutil +import subprocess +import sys +import time +from typing import Any +import uuid + +from .errors import LabyricornMcpError +from .gitops import GitOperations +from .safety import RepositoryGuard + + +class ValidationOperations: + BUILD_SOURCE_DIRECTORIES = ("assets", "configs", "content", "models", "scripts", "templates") + + def __init__(self, guard: RepositoryGuard, git: GitOperations): + self.guard = guard + self.git = git + + def _run(self, command: list[str], *, timeout: int) -> dict[str, Any]: + started = time.monotonic() + try: + result = subprocess.run( + command, + cwd=self.guard.root, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired as exc: + return { + "passed": False, + "returncode": None, + "duration_seconds": round(time.monotonic() - started, 3), + "stdout": (exc.stdout or "")[-100_000:] if isinstance(exc.stdout, str) else "", + "stderr": "Validation timed out.", + "timed_out": True, + } + except OSError as exc: + raise LabyricornMcpError("VALIDATION_FAILED", f"Validation could not start: {exc}") from exc + return { + "passed": result.returncode == 0, + "returncode": result.returncode, + "duration_seconds": round(time.monotonic() - started, 3), + "stdout": result.stdout[-100_000:], + "stderr": result.stderr[-100_000:], + "truncated": len(result.stdout) > 100_000 or len(result.stderr) > 100_000, + } + + def _lektor_executable(self) -> str: + adjacent = Path(sys.executable).with_name("lektor.exe" if sys.platform == "win32" else "lektor") + if adjacent.is_file(): + return str(adjacent) + discovered = shutil.which("lektor") + if discovered: + return discovered + raise LabyricornMcpError( + "RUNTIME_MISSING", + "Lektor is unavailable. Launch the MCP with the existing Lektor pipx Python environment or put lektor on PATH.", + ) + + def build_site(self) -> dict[str, Any]: + cache = self.guard.resolve(".cache") + cache.mkdir(exist_ok=True) + temporary = cache / f"labyricorn-mcp-build-{uuid.uuid4().hex}" + source = temporary / "source" + output = temporary / "output" + source.mkdir(parents=True) + try: + for name in self.BUILD_SOURCE_DIRECTORIES: + origin = self.guard.resolve(name, must_exist=True) + if origin.is_dir(): + self._validate_build_tree(origin) + shutil.copytree( + origin, + source / name, + ignore=shutil.ignore_patterns("__pycache__", "*.pyc"), + ) + shutil.copy2( + self.guard.root / "Labyricorn.lektorproject", + source / "Labyricorn.lektorproject", + ) + command = [ + sys.executable, + str(source / "scripts" / "build_with_projects.py"), + "--site-root", + str(source), + "--cache-path", + str(self.guard.resolve(".cache/project-sources")), + "--output-path", + str(output), + "--lektor", + self._lektor_executable(), + ] + result = self._run(command, timeout=600) + result["index_created"] = (output / "index.html").is_file() + result["output_retained"] = False + result["passed"] = result["passed"] and result["index_created"] + return result + finally: + shutil.rmtree(temporary, ignore_errors=True) + + def _validate_build_tree(self, root: Path) -> None: + if not root.resolve().is_relative_to(self.guard.root): + raise LabyricornMcpError("PATH_OUTSIDE_REPOSITORY", "A build source root resolves outside the repository.") + for path in root.rglob("*"): + if not path.resolve().is_relative_to(self.guard.root): + raise LabyricornMcpError( + "PATH_OUTSIDE_REPOSITORY", "A build source path resolves outside the repository." + ) + if path.is_symlink() or getattr(path, "is_junction", lambda: False)(): + if not path.resolve().is_relative_to(self.guard.root): + raise LabyricornMcpError( + "PATH_OUTSIDE_REPOSITORY", "A linked build source path resolves outside the repository." + ) + + def run_tests(self) -> dict[str, Any]: + return self._run( + [sys.executable, "-m", "unittest", "discover", "-s", "tests", "-v"], timeout=600 + ) + + def check_diff(self) -> dict[str, Any]: + result = self._run(["git", "diff", "--check", "HEAD"], timeout=60) + result["changed_files"] = [entry["path"] for entry in self.git.status()["changes"]] + return result + + def validate_changes(self) -> dict[str, Any]: + build = self.build_site() + tests = self.run_tests() + diff = self.check_diff() + return { + "passed": build["passed"] and tests["passed"] and diff["passed"], + "build": build, + "tests": tests, + "diff_check": diff, + "warnings": [ + line for result in (build, tests, diff) for line in result.get("stderr", "").splitlines() + if "warning" in line.casefold() + ], + "changed_files": diff["changed_files"], + } diff --git a/scripts/labyricorn_mcp_server.py b/scripts/labyricorn_mcp_server.py new file mode 100644 index 0000000..e50da47 --- /dev/null +++ b/scripts/labyricorn_mcp_server.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +"""Launch the repository-local Labyricorn MCP package over stdio.""" + +from __future__ import annotations + +from pathlib import Path +import sys + + +SITE_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(SITE_ROOT)) + +from labyricorn_mcp.__main__ import main # noqa: E402 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_labyricorn_mcp.py b/tests/test_labyricorn_mcp.py new file mode 100644 index 0000000..3311441 --- /dev/null +++ b/tests/test_labyricorn_mcp.py @@ -0,0 +1,266 @@ +from __future__ import annotations + +from io import StringIO +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import unittest +import uuid + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from labyricorn_mcp.domain import LabyricornDomain +from labyricorn_mcp.errors import LabyricornMcpError +from labyricorn_mcp.gitops import GitOperations +from labyricorn_mcp.safety import RepositoryGuard +from labyricorn_mcp.server import StdioMcpServer +from labyricorn_mcp.tools import ToolRegistry +from labyricorn_mcp.validation import ValidationOperations + + +ROOT = Path(__file__).resolve().parents[1] + + +class McpRepositoryFixture(unittest.TestCase): + def setUp(self) -> None: + self.base = ROOT / ".cache" / f"mcp-test-{uuid.uuid4().hex}" + self.site = self.base / "site" + for section, model in (("articles", "section"), ("blog", "section"), ("projects", "section")): + directory = self.site / "content" / section + directory.mkdir(parents=True) + (directory / "contents.lr").write_text( + f"_model: {model}\n---\ntitle: {section.title()}\n", encoding="utf-8" + ) + tags = self.site / "content" / "tags" + (tags / "writing").mkdir(parents=True) + (tags / "contents.lr").write_text("_model: tag-index\n---\ntitle: Tags\n", encoding="utf-8") + (tags / "writing" / "contents.lr").write_text( + "_model: tag\n---\ntitle: Writing\n---\nsummary: Writing and editing.\n", encoding="utf-8" + ) + (self.site / "models").mkdir() + for name in ("entry.ini", "tag.ini", "project.ini", "devlog.ini", "devlog-entry.ini", "section.ini", "tag-index.ini"): + shutil.copy2(ROOT / "models" / name, self.site / "models" / name) + (self.site / "configs").mkdir() + (self.site / "configs" / "project-sources.ini").write_text( + "[example]\nrepository = https://example.test/example.git\nweb_url = https://example.test/example\napi_url = https://example.test/api/example\nbranch = main\n", + encoding="utf-8", + ) + (self.site / "scripts").mkdir() + for name in ( + "article_editor.py", + "blog_editor.py", + "git_repo.py", + "narration_import.py", + "project_editor.py", + "project_providers.py", + "tag_editor.py", + ): + shutil.copy2(ROOT / "scripts" / name, self.site / "scripts" / name) + (self.site / "scripts" / "build_with_projects.py").write_text("raise SystemExit(0)\n", encoding="utf-8") + (self.site / "templates").mkdir() + (self.site / "templates" / "base.html").write_text("
Test
\n", encoding="utf-8") + (self.site / "assets" / "static").mkdir(parents=True) + (self.site / "assets" / "static" / "style.css").write_text("body {}\n", encoding="utf-8") + (self.site / "Labyricorn.lektorproject").write_text( + "[project]\nname = Fixture\nurl = https://example.test\n", encoding="utf-8" + ) + self.git("init", "--initial-branch=main") + self.git("config", "user.name", "MCP Test") + self.git("config", "user.email", "mcp@example.invalid") + self.git("add", ".") + self.git("commit", "-m", "Initial fixture") + self.guard = RepositoryGuard(self.site) + self.domain = LabyricornDomain(self.guard) + + def tearDown(self) -> None: + shutil.rmtree(self.base, onexc=self._make_writable) + + @staticmethod + def _make_writable(function: object, path: str, _error: BaseException) -> None: + os.chmod(path, 0o700) + function(path) + + def git(self, *args: str) -> subprocess.CompletedProcess[str]: + result = subprocess.run( + ["git", *args], cwd=self.site, capture_output=True, text=True, encoding="utf-8", check=False + ) + if result.returncode: + raise AssertionError(result.stderr or result.stdout) + return result + + @staticmethod + def article_fields(**changes: object) -> dict[str, object]: + fields: dict[str, object] = { + "title": "MCP Test Article", + "date": "2026-08-23", + "author": "Test", + "tags": ["writing", "Local Experiment"], + "summary": "A test article.", + "body": "Test body.", + } + fields.update(changes) + return fields + + +class RepositorySafetyTests(McpRepositoryFixture): + def test_repository_detection_rejects_wrong_directory(self) -> None: + wrong = self.base / "wrong" + wrong.mkdir() + with self.assertRaisesRegex(LabyricornMcpError, "compatible Labyricorn"): + RepositoryGuard(wrong) + + def test_path_traversal_and_absolute_paths_are_rejected(self) -> None: + for value in ("../outside.txt", str((self.base / "outside.txt").resolve())): + with self.subTest(value=value), self.assertRaisesRegex(LabyricornMcpError, "repository-relative"): + self.guard.resolve(value) + + def test_symlink_escape_is_rejected(self) -> None: + outside = self.base / "outside.txt" + outside.write_text("outside", encoding="utf-8") + link = self.site / "templates" / "escape.html" + try: + link.symlink_to(outside) + except OSError as exc: + self.skipTest(f"symlink creation is unavailable: {exc}") + with self.assertRaisesRegex(LabyricornMcpError, "outside the repository"): + self.guard.resolve("templates/escape.html", allowed_roots=("templates",), must_exist=True) + + +class ContentDomainTests(McpRepositoryFixture): + def test_content_create_read_update_and_conflict(self) -> None: + created = self.domain.create_article("mcp-test", self.article_fields()) + self.assertEqual("writing, local-experiment", created["fields"]["tags"]) + loaded = self.domain.get_content("article", "mcp-test") + updated = self.domain.update_article( + "mcp-test", {"summary": "Updated safely."}, loaded["sha256"] + ) + self.assertEqual("Updated safely.", updated["fields"]["summary"]) + with self.assertRaisesRegex(LabyricornMcpError, "changed after it was read"): + self.domain.update_article("mcp-test", {"summary": "Stale"}, loaded["sha256"]) + + def test_malformed_content_is_rejected(self) -> None: + created = self.domain.create_article("mcp-test", self.article_fields()) + with (self.site / created["path"]).open("a", encoding="utf-8") as output: + output.write("---\ntitle: Duplicate\n") + with self.assertRaises(LabyricornMcpError): + self.domain.get_content("article", "mcp-test") + + def test_model_description_is_derived_and_invalid_model_rejected(self) -> None: + model = self.domain.describe_model("entry") + self.assertIn("title", {field["name"] for field in model["fields"]}) + self.assertEqual("models/entry.ini", model["source"]) + with self.assertRaisesRegex(LabyricornMcpError, "Model names"): + self.domain.describe_model("../entry") + + def test_design_update_requires_matching_hash(self) -> None: + loaded = self.domain.get_design_file("templates/base.html") + updated = self.domain.update_design_file( + "templates/base.html", "
Updated
\n", loaded["sha256"] + ) + self.assertNotEqual(loaded["sha256"], updated["sha256"]) + with self.assertRaisesRegex(LabyricornMcpError, "changed after it was read"): + self.domain.update_design_file("templates/base.html", "stale", loaded["sha256"]) + + +class GitSafetyTests(McpRepositoryFixture): + def test_dirty_state_hash_detects_intervening_change(self) -> None: + operations = GitOperations(self.guard) + first = self.site / "first.txt" + first.write_text("first\n", encoding="utf-8") + status = operations.status() + (self.site / "second.txt").write_text("second\n", encoding="utf-8") + with self.assertRaisesRegex(LabyricornMcpError, "changed after it was reviewed"): + operations.commit("Test", ["first.txt"], status["status_sha256"]) + + def test_commit_is_limited_to_reviewed_paths(self) -> None: + operations = GitOperations(self.guard) + (self.site / "new").mkdir() + (self.site / "new" / "first.txt").write_text("first\n", encoding="utf-8") + (self.site / "second.txt").write_text("second\n", encoding="utf-8") + status = operations.status() + result = operations.commit("Commit one file", ["new/first.txt"], status["status_sha256"]) + self.assertEqual(["new/first.txt"], result["paths"]) + self.assertIn("second.txt", {item["path"] for item in result["status"]["changes"]}) + self.assertEqual("Commit one file", self.git("log", "-1", "--format=%s").stdout.strip()) + + def test_sensitive_file_cannot_be_committed(self) -> None: + operations = GitOperations(self.guard) + (self.site / ".env").write_text("TOKEN=secret\n", encoding="utf-8") + status = operations.status() + with self.assertRaisesRegex(LabyricornMcpError, "not eligible"): + operations.commit("Do not commit", [".env"], status["status_sha256"]) + + +class ProtocolAndValidationTests(McpRepositoryFixture): + def test_tool_registration_has_focused_surface_and_no_shell_or_deploy(self) -> None: + names = {tool["name"] for tool in ToolRegistry(self.guard).tools} + self.assertTrue({"site_info", "create_article", "validate_changes", "push_changes"} <= names) + self.assertFalse({"shell", "exec", "deploy", "publish_site"} & names) + + def test_mutation_reports_unrelated_worktree_context(self) -> None: + (self.site / "unrelated.txt").write_text("keep me\n", encoding="utf-8") + result = ToolRegistry(self.guard).call( + "create_article", {"slug": "mcp-test", "fields": self.article_fields()} + ) + context = result["worktree_context"] + self.assertIn("unrelated.txt", context["before_changed_paths"]) + self.assertIn("content/articles/mcp-test/", context["after_changed_paths"]) + + def test_stdio_protocol_initializes_lists_and_calls_tools(self) -> None: + messages = [ + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "test", "version": "1"}}}, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}, + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "site_structure", "arguments": {}}}, + ] + input_stream = StringIO("".join(json.dumps(message) + "\n" for message in messages)) + output_stream = StringIO() + server = StdioMcpServer(self.guard, input_stream=input_stream, output_stream=output_stream) + self.assertEqual(0, server.serve_forever()) + responses = [json.loads(line) for line in output_stream.getvalue().splitlines()] + self.assertEqual([1, 2, 3], [response["id"] for response in responses]) + self.assertEqual("2025-06-18", responses[0]["result"]["protocolVersion"]) + self.assertGreaterEqual(len(responses[1]["result"]["tools"]), 30) + self.assertFalse(responses[2]["result"]["isError"]) + + def test_validation_failure_is_structured(self) -> None: + git = GitOperations(self.guard) + validation = ValidationOperations(self.guard, git) + validation._lektor_executable = lambda: "lektor" # type: ignore[method-assign] + validation._run = lambda _command, timeout: { # type: ignore[method-assign] + "passed": False, "returncode": 1, "stdout": "", "stderr": "build failed", "duration_seconds": 0.1 + } + result = validation.build_site() + self.assertFalse(result["passed"]) + self.assertEqual("build failed", result["stderr"]) + self.assertFalse(result["output_retained"]) + + +class RealStdioStartupTests(unittest.TestCase): + def test_module_starts_over_stdio_without_protocol_noise(self) -> None: + request = json.dumps( + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "test", "version": "1"}}} + ) + "\n" + result = subprocess.run( + [sys.executable, "-m", "labyricorn_mcp", "--repository", str(ROOT)], + cwd=ROOT, + input=request, + capture_output=True, + text=True, + encoding="utf-8", + timeout=30, + check=False, + ) + self.assertEqual(0, result.returncode, result.stderr) + lines = result.stdout.splitlines() + self.assertEqual(1, len(lines)) + response = json.loads(lines[0]) + self.assertEqual("labyricorn-mcp", response["result"]["serverInfo"]["name"]) + + +if __name__ == "__main__": + unittest.main()