From a92fd72b1f6f6d02c2eae861b6c43495006ae693 Mon Sep 17 00:00:00 2001 From: Helm Date: Mon, 24 Aug 2026 15:01:35 -0400 Subject: [PATCH] obsidian-mcp: rename edit_document -> replace_document_content; add line-granular tools (read/insert/replace_document_lines) --- server.py | 680 ++++++++++++++++++++++++++++++++++++++++++------------ 1 file changed, 529 insertions(+), 151 deletions(-) diff --git a/server.py b/server.py index 832e478..370a97e 100644 --- a/server.py +++ b/server.py @@ -3,15 +3,25 @@ Obsidian MCP Server — Expose Obsidian vaults as MCP tools. Transport: StreamableHTTP (reachable cross-OS from WSL2 to Windows host). Run: python server.py --vaults-root "D:\\Obsidian" --port 8765 + +v3.1 — Git sync + auto-embedding: + - READ tools: git pull (2-min debounce) before accessing vault files + - WRITE tools: embed content via bge-micro-v2, write .smart-env/multi/*.ajson, git push + - smart-connections MCP auto-detects ajson changes via maybeReload() """ from __future__ import annotations import argparse +import datetime as dt import json import os import re -from datetime import datetime, timezone +import subprocess +import sys +import threading +import time +from datetime import timezone from pathlib import Path from typing import Any @@ -46,12 +56,198 @@ DEFAULT_EXCLUDE_EXT = {".webp", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".pdf", ".mp4", ".mp3", ".mov", ".excalidraw", ".bmp", ".ico", ".tiff", ".heic", ".avif"} +# ── Git Sync ─────────────────────────────────────────────────────────────── + +_git_lock = threading.Lock() +_last_git_pull: dict[str, float] = {} # vault_path -> timestamp +GIT_PULL_DEBOUNCE_S = 120 + + +def _git_pull_if_stale(vault_path: Path) -> None: + """Pull latest changes if last pull was > GIT_PULL_DEBOUNCE_S ago.""" + key = str(vault_path) + now = time.time() + if now - _last_git_pull.get(key, 0) < GIT_PULL_DEBOUNCE_S: + return + + git_dir = vault_path / ".git" + if not git_dir.is_dir(): + return + + with _git_lock: + # Re-check inside lock + if now - _last_git_pull.get(key, 0) < GIT_PULL_DEBOUNCE_S: + return + try: + subprocess.run( + ["git", "pull", "--ff-only"], + cwd=vault_path, + capture_output=True, + timeout=30, + check=False, + ) + except (subprocess.TimeoutExpired, OSError): + pass + _last_git_pull[key] = time.time() + + +def _git_commit_and_push(vault_path: Path, msg: str) -> None: + """Stage all changes, commit, and push.""" + git_dir = vault_path / ".git" + if not git_dir.is_dir(): + return + + with _git_lock: + try: + subprocess.run( + ["git", "add", "-A"], + cwd=vault_path, + capture_output=True, + timeout=30, + check=False, + ) + subprocess.run( + ["git", "commit", "-m", msg, "--allow-empty"], + cwd=vault_path, + capture_output=True, + timeout=30, + check=False, + ) + subprocess.run( + ["git", "push"], + cwd=vault_path, + capture_output=True, + timeout=60, + check=False, + ) + except (subprocess.TimeoutExpired, OSError): + pass + + +# ── Embedding (Smart Connections ajson format) ───────────────────────────── + +_embed_model = None +_embed_lock = threading.Lock() +EMBED_MODEL_KEY = "TaylorAI/bge-micro-v2" + + +def _get_embedder(): + """Lazy-load bge-micro-v2. Returns tuple (model, dim).""" + global _embed_model + if _embed_model is not None: + return _embed_model + + with _embed_lock: + if _embed_model is not None: + return _embed_model + try: + import numpy as np + from sentence_transformers import SentenceTransformer + model = SentenceTransformer(EMBED_MODEL_KEY) + dim = model.get_sentence_embedding_dimension() + _embed_model = (model, dim) + return _embed_model + except ImportError: + return None + except Exception: + return None + + +def _ajson_filename(doc_path: str) -> str: + """Convert doc path to ajson filename: '01 - Sessions/Session 01.md' -> '01 - Sessions_Session 01_md.ajson'""" + return doc_path.replace("/", "_").replace(".md", "_md") + ".ajson" + + +def _ajson_path(vault_path: Path, doc_path: str) -> Path: + """Get the full path to the ajson file for a note.""" + return vault_path / ".smart-env" / "multi" / _ajson_filename(doc_path) + + +def _write_ajson_entry(vault_path: Path, doc_path: str, content: str) -> bool: + """Embed note content and write/update its .smart-env/multi/*.ajson file. + + Returns True on success, False on failure (embedding unavailable, etc.) + """ + result = _get_embedder() + if result is None: + return False + + model, dim = result + try: + import numpy as np + vec = model.encode(content, normalize_embeddings=True) + vec_list = [float(v) for v in vec.tolist()] + except Exception: + return False + + entry = { + f"smart_sources:{doc_path}": { + "path": doc_path, + "embeddings": { + EMBED_MODEL_KEY: {"vec": vec_list} + }, + } + } + + ajson_file = _ajson_path(vault_path, doc_path) + ajson_file.parent.mkdir(parents=True, exist_ok=True) + + # Write: one JSON object per line (append-compatible, but we overwrite + # the whole file for simplicity — it's one note per file) + line = json.dumps(entry)[1:-1] + ",\n" # strip outer {}, add trailing comma+newline + ajson_file.write_text(line, encoding="utf-8") + return True + + +def _delete_ajson_entry(vault_path: Path, doc_path: str) -> bool: + """Remove a note's ajson file from .smart-env/multi/. Returns True if deleted.""" + ajson_file = _ajson_path(vault_path, doc_path) + if ajson_file.exists(): + ajson_file.unlink() + return True + return False + + +# ── Sync hooks ───────────────────────────────────────────────────────────── + +def _sync_pre_read(vault_name: str) -> None: + """Called before any read tool. Pulls git if stale.""" + try: + vault = _vault_path(vault_name) + _git_pull_if_stale(vault) + except (ValueError, FileNotFoundError, NotADirectoryError): + pass + + +def _sync_post_write(vault_name: str, doc_path: str, content: str | None = None) -> None: + """Called after any write tool. Embeds + pushes via git.""" + try: + vault = _vault_path(vault_name) + except (ValueError, FileNotFoundError, NotADirectoryError): + return + + if content is not None: + _write_ajson_entry(vault, doc_path, content) + + _git_commit_and_push(vault, f"[hermes] auto-sync: {doc_path}") + + +def _sync_post_delete(vault_name: str, doc_path: str) -> None: + """Called after delete_document. Removes ajson + pushes.""" + try: + vault = _vault_path(vault_name) + except (ValueError, FileNotFoundError, NotADirectoryError): + return + + _delete_ajson_entry(vault, doc_path) + _git_commit_and_push(vault, f"[hermes] auto-sync: delete {doc_path}") + # ── Helpers ─────────────────────────────────────────────────────────────── def _compact_date(timestamp: float) -> str: """Return YYYY-MM-DD (no time/microseconds/tz — 1/10 the token cost).""" - return datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d") + return dt.datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d") def _is_hidden_path(rel: Path) -> bool: @@ -62,26 +258,63 @@ def _is_hidden_path(rel: Path) -> bool: # ── Helpers ─────────────────────────────────────────────────────────────── def _vault_path(vault_name: str) -> Path: - """Get the absolute path to a vault directory, with safety check.""" + """Get the absolute path to a vault directory, with safety check. + + Falls back to case-insensitive matching if the exact name isn't found. + """ vault = (VAULTS_ROOT / vault_name).resolve() # Ensure it's actually inside VAULTS_ROOT (prevent traversal) if not str(vault).startswith(str(VAULTS_ROOT.resolve())): raise ValueError(f"Vault name escapes root: {vault_name}") + if not vault.exists(): - raise FileNotFoundError(f"Vault not found: {vault_name}") + # Case-insensitive fallback: scan siblings at the same level + parent = vault.parent + lower = vault_name.lower() + for candidate in parent.iterdir(): + if candidate.is_dir() and candidate.name.lower() == lower: + vault = candidate.resolve() + break + else: + raise FileNotFoundError(f"Vault not found: {vault_name}") + if not vault.is_dir(): raise NotADirectoryError(f"Not a directory: {vault_name}") return vault def _resolve_doc(vault_name: str, doc_path: str) -> Path: - """Resolve a document path within a vault, preventing traversal.""" + """Resolve a document path within a vault, preventing traversal. + + Falls back to case-insensitive path component matching on Linux. + """ vault = _vault_path(vault_name) # Normalize: strip leading slash, handle backslashes clean = doc_path.lstrip("/\\").replace("\\", "/") target = (vault / clean).resolve() if not str(target).startswith(str(vault)): raise ValueError(f"Document path escapes vault: {doc_path}") + + if not target.exists(): + # Case-insensitive path component fallback + parts = clean.split("/") + current = vault + for part in parts: + lower = part.lower() + found = None + for child in current.iterdir(): + if child.name.lower() == lower: + found = child + break + if found is None: + # Component not found — let the original target raise + break + current = found + else: + # All components resolved case-insensitively + if str(current.resolve()).startswith(str(vault)): + return current.resolve() + return target @@ -104,7 +337,7 @@ def _file_info_full(p: Path, relative_to: Path) -> dict[str, Any]: "size_bytes": stat.st_size if p.is_file() else None, "modified": _compact_date(stat.st_mtime), } - if p.is_file() and _is_markdown(p): + if p.is_file() and isinstance(info, dict): try: content = p.read_text(encoding="utf-8", errors="replace") except Exception: @@ -118,29 +351,6 @@ def _is_markdown(p: Path) -> bool: return p.suffix.lower() in MARKDOWN_EXT -def _search_file(filepath: Path, query: str) -> list[dict[str, Any]]: - """Search a single markdown file for query, return match contexts.""" - try: - text = filepath.read_text(encoding="utf-8", errors="replace") - except Exception: - return [] - results = [] - query_lower = query.lower() - for i, line in enumerate(text.splitlines(), start=1): - if query_lower in line.lower(): - # Trim context snippet - idx = line.lower().index(query_lower) - start = max(0, idx - 40) - end = min(len(line), idx + len(query) + 40) - snippet = line[start:end] - if start > 0: - snippet = "…" + snippet - if end < len(line): - snippet += "…" - results.append({"line": i, "snippet": snippet.strip()}) - return results - - def _parse_wikilinks(content: str) -> list[str]: """Extract [[wikilink]] targets from markdown content.""" # Match [[Target]] or [[Target|Alias]] @@ -230,6 +440,7 @@ def list_vault( total: combined count """ vault = _vault_path(vault_name) + _sync_pre_read(vault_name) # Normalize the inner path clean_path = path.lstrip("/\\").replace("\\", "/") target = (vault / clean_path).resolve() if clean_path else vault @@ -296,6 +507,7 @@ def get_vault_document(vault_name: str, doc_path: str) -> dict[str, Any]: Returns the document content with metadata. """ + _sync_pre_read(vault_name) target = _resolve_doc(vault_name, doc_path) if not target.exists(): return {"error": f"Document not found: {doc_path}", "vault": vault_name} @@ -329,56 +541,13 @@ def get_file_info(vault_name: str, doc_path: str) -> dict[str, Any]: Returns metadata dict with: name, path, type, size_bytes, modified (YYYY-MM-DD), and for markdown files: line_count, char_count. """ + _sync_pre_read(vault_name) target = _resolve_doc(vault_name, doc_path) if not target.exists(): return {"error": f"File not found: {doc_path}", "vault": vault_name} return _file_info_full(target, _vault_path(vault_name)) -@mcp.tool() -def search_vault(vault_name: str, query: str, max_results: int = 50) -> dict[str, Any]: - """FILESYSTEM TOOL (not memory) — Search markdown files on disk. Find notes about characters, locations, concepts. - - Args: - vault_name: Name of the vault to search. - query: Search query (case-insensitive substring match). - max_results: Maximum number of match results to return (default 50). - - Returns matches grouped by file with line numbers and context snippets. - """ - vault = _vault_path(vault_name) - results = [] - files_searched = 0 - - for md_file in vault.rglob("*.md"): - rel_path = md_file.relative_to(vault) - if _is_hidden_path(rel_path): - continue - - files_searched += 1 - matches = _search_file(md_file, query) - if matches: - results.append({ - "file": str(rel_path).replace("\\", "/"), - "matches": matches[:max_results], - "match_count": len(matches), - }) - - if len(results) >= max_results: - break - - # Sort by match count descending - results.sort(key=lambda r: r["match_count"], reverse=True) - - return { - "vault": vault_name, - "query": query, - "results": results[:max_results], - "total_matches": sum(r["match_count"] for r in results), - "files_searched": files_searched, - "files_with_matches": len(results), - } - @mcp.tool() def get_recent_changes(vault_name: str, limit: int = 20) -> dict[str, Any]: @@ -390,6 +559,7 @@ def get_recent_changes(vault_name: str, limit: int = 20) -> dict[str, Any]: Returns documents sorted by modification time, most recent first. """ + _sync_pre_read(vault_name) vault = _vault_path(vault_name) md_files = [] @@ -423,6 +593,7 @@ def get_backlinks(vault_name: str, doc_path: str) -> dict[str, Any]: Returns a list of files that link to it, with context snippets. """ + _sync_pre_read(vault_name) vault = _vault_path(vault_name) target = _resolve_doc(vault_name, doc_path) target_rel = str(target.relative_to(vault)).replace("\\", "/") @@ -447,8 +618,9 @@ def get_backlinks(vault_name: str, doc_path: str) -> dict[str, Any]: matching_links = [] for m in pattern.finditer(content): link_target = m.group(1).strip() - # Match if link_target matches stem, filename, or full path - if link_target == target_stem or link_target == target_rel or link_target == target.name: + # Case-insensitive match against stem, filename, or full path + link_lower = link_target.lower() + if link_lower == target_stem.lower() or link_lower == target_rel.lower() or link_lower == target.name.lower(): # Get line context line_no = content[:m.start()].count("\n") + 1 lines = content.splitlines() @@ -485,6 +657,7 @@ def create_document(vault_name: str, doc_path: str, content: str) -> dict[str, A Returns the created document info. """ + _sync_pre_read(vault_name) if not doc_path.lower().endswith(".md"): doc_path = doc_path.rstrip("/\\") + ".md" @@ -495,25 +668,33 @@ def create_document(vault_name: str, doc_path: str, content: str) -> dict[str, A target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") + # Sync + clean_path = doc_path.lstrip("/\\").replace("\\", "/") + _sync_post_write(vault_name, clean_path, content) + return { "vault": vault_name, - "path": doc_path.lstrip("/\\").replace("\\", "/"), + "path": clean_path, "action": "created", "size_bytes": target.stat().st_size, } @mcp.tool() -def edit_document(vault_name: str, doc_path: str, content: str) -> dict[str, Any]: - """Edit (overwrite) an existing markdown document in a vault. +def replace_document_content(vault_name: str, doc_path: str, content: str) -> dict[str, Any]: + """REPLACES THE ENTIRE FILE with `content` — this is a full overwrite, NOT a + patch or partial edit. Pass the COMPLETE new document text, not a fragment. + If you only need to add text at the end, use append_to_document. If you need + to change a few lines in place, use replace_document_lines or insert_document_lines. Args: vault_name: Name of the vault. doc_path: Relative path to the document. - content: New markdown content (replaces entire file). + content: COMPLETE new markdown content for the whole document. - Returns the edited document info. + Returns the replaced document info. """ + _sync_pre_read(vault_name) if not doc_path.lower().endswith(".md"): doc_path = doc_path.rstrip("/\\") + ".md" @@ -523,14 +704,235 @@ def edit_document(vault_name: str, doc_path: str, content: str) -> dict[str, Any target.write_text(content, encoding="utf-8") + # Sync + clean_path = doc_path.lstrip("/\\").replace("\\", "/") + _sync_post_write(vault_name, clean_path, content) + + return { + "vault": vault_name, + "path": clean_path, + "action": "replaced", + "size_bytes": target.stat().st_size, + } + + +@mcp.tool() +def read_document_lines(vault_name: str, doc_path: str, starting_line_number: int, total_lines: int) -> dict[str, Any]: + """Read a bounded window of lines from a markdown document. + + Lines are 1-indexed. `starting_line_number` is the first line returned; + `total_lines` is the number of lines to return. Returns the full document's + `total_lines` count so callers can page through a large note in chunks. + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + starting_line_number: 1-indexed line to start reading from (>= 1). + total_lines: Number of lines to return. + + Returns the line window plus the document's total line count. + """ + _sync_pre_read(vault_name) + 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") + lines = content.splitlines() + line_count = len(lines) + + start = max(1, starting_line_number) + end = start + max(0, total_lines) + window = lines[start - 1 : end] + return { "vault": vault_name, "path": doc_path.lstrip("/\\").replace("\\", "/"), - "action": "edited", + "content": "\n".join(window), + "starting_line_number": start, + "returned_lines": len(window), + "total_lines": line_count, + } + + +@mcp.tool() +def insert_document_lines(vault_name: str, doc_path: str, starting_line_number: int, new_content: str) -> dict[str, Any]: + """Insert lines into a markdown document at a specific position. + + 1-indexed, insert-AT semantics: `new_content`'s first line lands at + `starting_line_number`, and the line previously at that position shifts down. + To insert at the very end instead, set starting_line_number = total_lines + 1 + (or simply use append_to_document). + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + starting_line_number: 1-indexed line where new_content begins (>= 1). + new_content: Markdown text to insert (may be multiple lines). + + Returns the updated document info. + """ + _sync_pre_read(vault_name) + if not doc_path.lower().endswith(".md"): + doc_path = doc_path.rstrip("/\\") + ".md" + + target = _resolve_doc(vault_name, doc_path) + clean_path = doc_path.lstrip("/\\").replace("\\", "/") + + existing = target.read_text(encoding="utf-8", errors="replace") if target.exists() else "" + lines = existing.splitlines(keepends=True) + + new_lines = new_content.splitlines(keepends=True) + if not new_lines: + new_lines = ["\n"] + + # Clamp insert position into [1, len(lines) + 1] + insert_at = max(1, min(starting_line_number, len(lines) + 1)) + insert_idx = insert_at - 1 + + if lines and not lines[-1].endswith("\n"): + lines[-1] = lines[-1] + "\n" + if new_lines and not new_lines[-1].endswith("\n"): + new_lines[-1] = new_lines[-1] + "\n" + + new_file_lines = lines[:insert_idx] + new_lines + lines[insert_idx:] + full = "".join(new_file_lines) + + if not target.exists(): + target.parent.mkdir(parents=True, exist_ok=True) + + target.write_text(full, encoding="utf-8") + _sync_post_write(vault_name, clean_path, full) + + return { + "vault": vault_name, + "path": clean_path, + "action": "inserted", + "inserted_at_line": insert_at, "size_bytes": target.stat().st_size, } +@mcp.tool() +def replace_document_lines(vault_name: str, doc_path: str, starting_line_number: int, ending_line_number: int, new_content: str) -> dict[str, Any]: + """Replace an inclusive range of lines [starting_line_number, ending_line_number] + with `new_content`. Pass new_content="" to delete the range outright. + + 1-indexed, inclusive on both ends. This is the safe in-place edit primitive: + no read-splice-overwrite of the whole file required. + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + starting_line_number: 1-indexed first line of the range to replace (>= 1). + ending_line_number: 1-indexed last line of the range to replace (>= start). + new_content: Markdown text to substitute for the range (empty string deletes it). + + Returns the updated document info. + """ + _sync_pre_read(vault_name) + if not doc_path.lower().endswith(".md"): + doc_path = doc_path.rstrip("/\\") + ".md" + + target = _resolve_doc(vault_name, doc_path) + if not target.exists(): + return {"error": f"Document does not exist: {doc_path}", "vault": vault_name} + + clean_path = doc_path.lstrip("/\\").replace("\\", "/") + + existing = target.read_text(encoding="utf-8", errors="replace") + lines = existing.splitlines(keepends=True) + line_count = max(1, len(existing.splitlines())) + + start = max(1, starting_line_number) + end = max(start, min(ending_line_number, line_count)) + start_idx = start - 1 + end_idx = end # exclusive + + new_lines = new_content.splitlines(keepends=True) if new_content else [] + if new_lines and not new_lines[-1].endswith("\n"): + new_lines[-1] = new_lines[-1] + "\n" + + new_file_lines = lines[:start_idx] + new_lines + lines[end_idx:] + full = "".join(new_file_lines) + + target.write_text(full, encoding="utf-8") + _sync_post_write(vault_name, clean_path, full) + + return { + "vault": vault_name, + "path": clean_path, + "action": "replaced_lines", + "replaced_line_start": start, + "replaced_line_end": end, + "size_bytes": target.stat().st_size, + } + + +@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. + """ + _sync_pre_read(vault_name) + if not doc_path.lower().endswith(".md"): + doc_path = doc_path.rstrip("/\\") + ".md" + + target = _resolve_doc(vault_name, doc_path) + clean_path = doc_path.lstrip("/\\").replace("\\", "/") + + 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" + full = existing + separator + content + "\n" + target.write_text(full, encoding="utf-8") + action = "appended" + + # Sync — pass full content for embedding + full_content = target.read_text(encoding="utf-8", errors="replace") + _sync_post_write(vault_name, clean_path, full_content) + + return { + "vault": vault_name, + "path": clean_path, + "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. + """ + _sync_pre_read(vault_name) + target = _resolve_doc(vault_name, doc_path) + if not target.exists(): + return {"error": f"Document not found: {doc_path}", "vault": vault_name} + + target.unlink() + + clean_path = doc_path.lstrip("/\\").replace("\\", "/") + _sync_post_delete(vault_name, clean_path) + + return {"vault": vault_name, "path": clean_path, "action": "deleted"} + + @mcp.tool() def create_vault_directory(vault_name: str, dir_path: str) -> dict[str, Any]: """FILESYSTEM TOOL (not memory) — Create a new directory in a vault on disk. @@ -578,6 +980,7 @@ def move_document(vault_name: str, src_path: str, dest_path: str) -> dict[str, A Returns the move result with both old and new paths. """ + _sync_pre_read(vault_name) src = _resolve_doc(vault_name, src_path) if not src.exists(): return {"error": f"Source document does not exist: {src_path}", "vault": vault_name} @@ -592,9 +995,21 @@ def move_document(vault_name: str, src_path: str, dest_path: str) -> dict[str, A src.rename(dest) vault = _vault_path(vault_name) + clean_src = src_path.lstrip("/\\").replace("\\", "/") + clean_dest = dest_path.lstrip("/\\").replace("\\", "/") + + # Remove old ajson, embed new location + _delete_ajson_entry(vault, clean_src) + try: + content = dest.read_text(encoding="utf-8", errors="replace") + _write_ajson_entry(vault, clean_dest, content) + except Exception: + content = None + _git_commit_and_push(vault, f"[hermes] auto-sync: move {clean_src} -> {clean_dest}") + return { "vault": vault_name, - "old_path": src_path.lstrip("/\\").replace("\\", "/"), + "old_path": clean_src, "new_path": str(dest.relative_to(vault)).replace("\\", "/"), "action": "moved", } @@ -611,6 +1026,7 @@ def read_frontmatter(vault_name: str, doc_path: str, key: str | None = None) -> Returns the frontmatter dict or specific key value. """ + _sync_pre_read(vault_name) target = _resolve_doc(vault_name, doc_path) if not target.exists(): return {"error": f"Document not found: {doc_path}", "vault": vault_name} @@ -637,6 +1053,7 @@ def write_frontmatter(vault_name: str, doc_path: str, key: str, value: Any = Non Returns the updated frontmatter state. """ + _sync_pre_read(vault_name) target = _resolve_doc(vault_name, doc_path) if not target.exists(): return {"error": f"Document not found: {doc_path}", "vault": vault_name} @@ -651,73 +1068,26 @@ def write_frontmatter(vault_name: str, doc_path: str, key: str, value: Any = Non 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" + # 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") + new_fm_block = "---\n" + yaml.dump(fm, default_flow_style=False, allow_unicode=True).strip() + "\n---" + body = _body_after_frontmatter(content) + new_content = new_fm_block + "\n\n" + body + target.write_text(new_content, encoding="utf-8") - return {"vault": vault_name, "path": doc_path, "key": key, "value": value, "action": action} + clean_path = doc_path.lstrip("/\\").replace("\\", "/") + # Re-read the full content for embedding + full_content = target.read_text(encoding="utf-8", errors="replace") + _sync_post_write(vault_name, clean_path, full_content) - -@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"} + return {"vault": vault_name, "path": clean_path, "key": key, "value": value, "action": action} def _parse_frontmatter(content: str) -> dict[str, Any] | None: @@ -763,17 +1133,25 @@ def main(): default="0.0.0.0", help="Bind host (default: 0.0.0.0 for LAN access)", ) + parser.add_argument( + "--transport", + choices=["streamable-http", "stdio"], + default="streamable-http", + help="Transport mode: streamable-http (network) or stdio (subprocess) (default: streamable-http)", + ) args = parser.parse_args() global VAULTS_ROOT VAULTS_ROOT = resolve_vaults_root(args.vaults_root) - print(f"Obsidian MCP Server") - print(f" Vaults root: {VAULTS_ROOT}") - print(f" Listening: http://{args.host}:{args.port}/mcp") - print(f" Vaults found: {len(get_vaults()['vaults'])}") - - mcp.run(transport="streamable-http", host=args.host, port=args.port, path="/mcp") + if args.transport == "stdio": + mcp.run(transport="stdio") + else: + print(f"Obsidian MCP Server") + print(f" Vaults root: {VAULTS_ROOT}") + print(f" Listening: http://{args.host}:{args.port}/mcp") + print(f" Vaults found: {len(get_vaults()['vaults'])}") + mcp.run(transport="streamable-http", host=args.host, port=args.port, path="/mcp") if __name__ == "__main__":