diff --git a/requirements.txt b/requirements.txt index 9074148..fcd937a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ -fastmcp>=2.0.0 \ No newline at end of file +fastmcp>=2.0.0 +pyyaml>=6.0 \ No newline at end of file diff --git a/server.py b/server.py index 4d157c9..832e478 100644 --- a/server.py +++ b/server.py @@ -15,6 +15,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any +import yaml from fastmcp import FastMCP # ── Config ──────────────────────────────────────────────────────────────── @@ -599,6 +600,149 @@ def move_document(vault_name: str, src_path: str, dest_path: str) -> dict[str, A } +@mcp.tool() +def read_frontmatter(vault_name: str, doc_path: str, key: str | None = None) -> dict[str, Any]: + """Read YAML frontmatter from a markdown document. + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + key: Optional specific key to read. If None, returns all frontmatter. + + Returns the frontmatter dict or specific key value. + """ + target = _resolve_doc(vault_name, doc_path) + if not target.exists(): + return {"error": f"Document not found: {doc_path}", "vault": vault_name} + + content = target.read_text(encoding="utf-8", errors="replace") + fm = _parse_frontmatter(content) + if fm is None: + return {"vault": vault_name, "path": doc_path, "frontmatter": {}, "note": "No frontmatter found"} + + if key: + return {"vault": vault_name, "path": doc_path, "key": key, "value": fm.get(key)} + return {"vault": vault_name, "path": doc_path, "frontmatter": fm} + + +@mcp.tool() +def write_frontmatter(vault_name: str, doc_path: str, key: str, value: Any = None) -> dict[str, Any]: + """Set a YAML frontmatter key on a markdown document. Creates frontmatter if absent. + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + key: Frontmatter key to set. + value: Value to set (string, number, boolean, list, or dict). Pass null/None to delete. + + Returns the updated frontmatter state. + """ + target = _resolve_doc(vault_name, doc_path) + if not target.exists(): + return {"error": f"Document not found: {doc_path}", "vault": vault_name} + + content = target.read_text(encoding="utf-8", errors="replace") + fm = _parse_frontmatter(content) + + if fm is None: + # No frontmatter — create one + if value is None: + return {"vault": vault_name, "path": doc_path, "action": "noop", "note": "No frontmatter to delete from"} + new_fm = {key: value} + new_content = "---\n" + yaml.dump(new_fm, default_flow_style=False, allow_unicode=True).strip() + "\n---\n\n" + content + target.write_text(new_content, encoding="utf-8") + return {"vault": vault_name, "path": doc_path, "key": key, "value": value, "action": "created"} + + # Existing frontmatter + if value is None: + fm.pop(key, None) + action = "deleted" + else: + fm[key] = value + action = "set" + + new_fm_block = "---\n" + yaml.dump(fm, default_flow_style=False, allow_unicode=True).strip() + "\n---" + body = _body_after_frontmatter(content) + target.write_text(new_fm_block + "\n\n" + body, encoding="utf-8") + + return {"vault": vault_name, "path": doc_path, "key": key, "value": value, "action": action} + + +@mcp.tool() +def append_to_document(vault_name: str, doc_path: str, content: str) -> dict[str, Any]: + """Append content to the end of a markdown document. Creates the file if it doesn't exist. + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + content: Markdown content to append. + + Returns the updated document info. + """ + if not doc_path.lower().endswith(".md"): + doc_path = doc_path.rstrip("/\\") + ".md" + + target = _resolve_doc(vault_name, doc_path) + + if not target.exists(): + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content + "\n", encoding="utf-8") + action = "created" + else: + existing = target.read_text(encoding="utf-8", errors="replace") + separator = "\n" if existing.endswith("\n") else "\n\n" + target.write_text(existing + separator + content + "\n", encoding="utf-8") + action = "appended" + + return { + "vault": vault_name, + "path": doc_path.lstrip("/\\").replace("\\", "/"), + "action": action, + "size_bytes": target.stat().st_size, + } + + +@mcp.tool() +def delete_document(vault_name: str, doc_path: str) -> dict[str, Any]: + """Delete a document from a vault. + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + + Returns deletion confirmation. + """ + target = _resolve_doc(vault_name, doc_path) + if not target.exists(): + return {"error": f"Document not found: {doc_path}", "vault": vault_name} + + target.unlink() + return {"vault": vault_name, "path": doc_path.lstrip("/\\").replace("\\", "/"), "action": "deleted"} + + +def _parse_frontmatter(content: str) -> dict[str, Any] | None: + """Parse YAML frontmatter from markdown content. Returns None if not found.""" + if not content.startswith("---"): + return None + parts = content.split("---", 2) + if len(parts) < 3: + return None + try: + return yaml.safe_load(parts[1]) or {} + except yaml.YAMLError: + return None + + +def _body_after_frontmatter(content: str) -> str: + """Return the body content after YAML frontmatter.""" + if not content.startswith("---"): + return content + parts = content.split("---", 2) + if len(parts) < 3: + return content + return parts[2].lstrip("\n") + + # ── Entrypoint ──────────────────────────────────────────────────────────── def main():