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()