obsidian-mcp-server/server.py

"""
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
import subprocess
import sys
import threading
import time
from datetime import timezone
from pathlib import Path
from typing import Any

import yaml
from fastmcp import FastMCP

# ── Config ────────────────────────────────────────────────────────────────

def resolve_vaults_root(cli_arg: str | None = None) -> Path:
    """Resolve vaults root: CLI arg > env var > config.json > cwd."""
    if cli_arg:
        return Path(cli_arg).expanduser().resolve()

    env_val = os.environ.get("OBSIDIAN_VAULTS_ROOT")
    if env_val:
        return Path(env_val).expanduser().resolve()

    config_path = Path(__file__).parent / "config.json"
    if config_path.exists():
        data = json.loads(config_path.read_text(encoding="utf-8"))
        if root := data.get("vaults_root"):
            return Path(root).expanduser().resolve()

    # Fallback: assume vaults/ is next to this script
    return (Path(__file__).parent / "vaults").resolve()


VAULTS_ROOT: Path = Path(".")  # set on startup
OBSIDIAN_HIDDEN = {".obsidian", ".trash", ".git", ".DS_Store", "node_modules"}
MARKDOWN_EXT = {".md", ".markdown"}
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 dt.datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d")


def _is_hidden_path(rel: Path) -> bool:
    """True if any path component is a dotfile/dotdir (.obsidian, .git, .trash, etc.)."""
    return any(part.startswith(".") for part in rel.parts)


# ── Helpers ───────────────────────────────────────────────────────────────

def _vault_path(vault_name: str) -> Path:
    """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():
        # 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.

    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


def _file_info_lean(p: Path, relative_to: Path) -> dict[str, Any]:
    """Bare index entry — name, path, type only. ~55 chars vs ~120."""
    return {
        "name": p.name,
        "path": str(p.relative_to(relative_to)).replace("\\", "/"),
        "type": "directory" if p.is_dir() else "file",
    }


def _file_info_full(p: Path, relative_to: Path) -> dict[str, Any]:
    """Full metadata for a single file — size, date, counts."""
    stat = p.stat()
    info: dict[str, Any] = {
        "name": p.name,
        "path": str(p.relative_to(relative_to)).replace("\\", "/"),
        "type": "directory" if p.is_dir() else "file",
        "size_bytes": stat.st_size if p.is_file() else None,
        "modified": _compact_date(stat.st_mtime),
    }
    if p.is_file() and isinstance(info, dict):
        try:
            content = p.read_text(encoding="utf-8", errors="replace")
        except Exception:
            content = ""
        info["line_count"] = content.count("\n") + (1 if content else 0)
        info["char_count"] = len(content)
    return info


def _is_markdown(p: Path) -> bool:
    return p.suffix.lower() in MARKDOWN_EXT


def _parse_wikilinks(content: str) -> list[str]:
    """Extract [[wikilink]] targets from markdown content."""
    # Match [[Target]] or [[Target|Alias]]
    pattern = re.compile(r"\[\[([^\]|#]+?)(?:[|#][^\]]+)?\]\]")
    return [m.group(1).strip() for m in pattern.finditer(content)]


# ── Server ────────────────────────────────────────────────────────────────

mcp = FastMCP("Obsidian Vaults")


# ── Tools ─────────────────────────────────────────────────────────────────

@mcp.tool()
def get_vaults() -> dict[str, Any]:
    """FILESYSTEM TOOL (not memory) — List Obsidian vaults on disk. Use to discover available vaults before browsing one.

    Returns a list of vaults with name, path, and note counts.
    """
    if not VAULTS_ROOT.exists():
        return {"vaults": [], "vaults_root": str(VAULTS_ROOT), "error": "Vaults root does not exist."}

    vaults = []
    for entry in sorted(VAULTS_ROOT.iterdir()):
        if not entry.is_dir():
            continue
        if entry.name.startswith("."):
            continue
        # Check if it looks like a vault (has .obsidian or contains .md files)
        has_obsidian = (entry / ".obsidian").is_dir()
        md_count = sum(1 for f in entry.rglob("*.md") if ".obsidian" not in str(f) and ".trash" not in str(f))
        if has_obsidian or md_count > 0:
            vaults.append({
                "name": entry.name,
                "path": str(entry),
                "note_count": md_count,
                "has_obsidian_config": has_obsidian,
            })

    return {"vaults": vaults, "vaults_root": str(VAULTS_ROOT), "count": len(vaults)}


@mcp.tool()
def list_vault(
    vault_name: str,
    path: str = "",
    recursive: bool = False,
    include_files: bool = True,
    include_directories: bool = True,
    file_extensions: list[str] | None = None,  # None = all; pass ['.md'] for notes only
    exclude_extensions: list[str] | None = None,  # None = default image blacklist (see below)
    include_hidden: bool = False,
) -> dict[str, Any]:
    """FILESYSTEM TOOL (not memory) — Index files and folders in vault on disk. Use to see structure before reading docs.

    Returns a lean listing (name + path + type only). Image/binary files (.webp,
    .png, .jpg, .gif, .svg, .pdf, .mp4, .mp3, .mov, .excalidraw) are excluded by
    default — the model can't read them anyway. Pass exclude_extensions=[] to see
    everything. Use get_file_info for size/date/counts on a specific file.
    Hidden dirs are excluded by default.

    Args:
        vault_name: Name of the vault (subdirectory name under vaults root)
        path: Subfolder inside the vault. Omit or pass "" for vault root.
            NOT the vault name — that's vault_name. Examples:
            "" or omit = vault root, "NPCs" = vault/NPCs/, "NPCs/Hostile" = deeper.
        recursive: If True, list all nested contents recursively.
        include_files: Include files in results (default True).
        include_directories: Include directories in results (default True).
        file_extensions: Filter files by extension. Pass ['.md'] for notes only,
            ['.md', '.canvas'] for notes + canvases, or omit/None for all files.
        exclude_extensions: Extensions to EXCLUDE. Defaults to binary/image types
            (webp, png, jpg, gif, svg, pdf, mp4, mp3, mov, excalidraw).
            Pass [] to disable the blacklist.
        include_hidden: If True, include .obsidian config directory (default False).

    Usage:
        list_vault("obsidian-skt")                      → all files (images excluded) at root
        list_vault("obsidian-skt", file_extensions=['.md'])  → notes only
        list_vault("obsidian-skt", "", True, file_extensions=['.md'])  → all notes recursively
        list_vault("obsidian-skt", exclude_extensions=[])  → everything including images

    Returns:
        directories: [{name, path, type}]
        files: [{name, path, type}]
        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

    if not str(target).startswith(str(vault)):
        return {"error": "Path escapes vault boundary.", "vault": vault_name}

    if not target.exists():
        return {"error": f"Path not found: {path or '/'}", "vault": vault_name}

    # Resolve exclude list: explicit arg > default image blacklist
    if exclude_extensions is None:
        exclude_set = DEFAULT_EXCLUDE_EXT
    else:
        exclude_set = {e.lower() for e in exclude_extensions}

    directories = []
    files = []

    if recursive:
        iterator = target.rglob("*")
    else:
        iterator = target.iterdir()

    for entry in sorted(iterator, key=lambda p: (not p.is_dir(), p.name.lower())):
        rel = entry.relative_to(vault)

        # Filter hidden directories/files unless include_hidden is set
        if _is_hidden_path(rel) and not include_hidden:
            continue

        info = _file_info_lean(entry, vault)
        if entry.is_dir():
            if include_directories:
                directories.append(info)
        else:
            if not include_files:
                continue
            if file_extensions:
                # Case-insensitive extension check
                ext_lower = entry.suffix.lower()
                if not any(ext_lower == fe.lower() for fe in file_extensions):
                    continue
            if entry.suffix.lower() in exclude_set:
                continue
            files.append(info)

    return {
        "vault": vault_name,
        "path": str(target.relative_to(vault)).replace("\\", "/") or "/",
        "directories": directories,
        "files": files,
        "total": len(directories) + len(files),
    }


@mcp.tool()
def get_vault_document(vault_name: str, doc_path: str) -> dict[str, Any]:
    """FILESYSTEM TOOL (not memory) — Read a markdown note from vault on disk. Use after list_vault to get content.

    Args:
        vault_name: Name of the vault.
        doc_path: Relative path to the document within the vault (e.g. 'folder/note.md').

    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}
    if not _is_markdown(target):
        return {"error": f"Not a markdown file: {doc_path}", "vault": vault_name}

    content = target.read_text(encoding="utf-8", errors="replace")
    stat = target.stat()
    lines = content.splitlines()

    return {
        "vault": vault_name,
        "path": doc_path.lstrip("/\\").replace("\\", "/"),
        "content": content,
        "line_count": len(lines),
        "word_count": len(content.split()),
        "char_count": len(content),
        "size_bytes": stat.st_size,
        "modified": _compact_date(stat.st_mtime),
    }


@mcp.tool()
def get_file_info(vault_name: str, doc_path: str) -> dict[str, Any]:
    """FILESYSTEM TOOL (not memory) — Get size, date, and line count for a single file. Use after list_vault to inspect a specific note.

    Args:
        vault_name: Name of the vault.
        doc_path: Relative path to the file within the vault (e.g. 'folder/note.md').

    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 get_recent_changes(vault_name: str, limit: int = 20) -> dict[str, Any]:
    """Get recently modified documents in a vault.

    Args:
        vault_name: Name of the vault.
        limit: Maximum number of results (default 20).

    Returns documents sorted by modification time, most recent first.
    """
    _sync_pre_read(vault_name)
    vault = _vault_path(vault_name)
    md_files = []

    for md_file in vault.rglob("*.md"):
        rel_path = md_file.relative_to(vault)
        if _is_hidden_path(rel_path):
            continue
        rel = str(rel_path)
        stat = md_file.stat()
        md_files.append({
            "path": rel.replace("\\", "/"),
            "modified": _compact_date(stat.st_mtime),
        })

    md_files.sort(key=lambda f: f["modified"], reverse=True)

    return {
        "vault": vault_name,
        "recent": md_files[:limit],
        "total_notes": len(md_files),
    }


@mcp.tool()
def get_backlinks(vault_name: str, doc_path: str) -> dict[str, Any]:
    """Find all documents that link to the given document via [[wikilinks]].

    Args:
        vault_name: Name of the vault.
        doc_path: Relative path to the target document.

    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("\\", "/")
    target_stem = target.stem  # filename without extension

    backlinks = []

    for md_file in vault.rglob("*.md"):
        rel_path = md_file.relative_to(vault)
        if _is_hidden_path(rel_path):
            continue
        if md_file.resolve() == target.resolve():
            continue  # skip self

        try:
            content = md_file.read_text(encoding="utf-8", errors="replace")
        except Exception:
            continue

        # Check for wikilinks matching either the full path or stem
        pattern = re.compile(r"\[\[([^\]|#]+?)(?:[|#][^\]]+)?\]\]")
        matching_links = []
        for m in pattern.finditer(content):
            link_target = m.group(1).strip()
            # 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()
                line_text = lines[line_no - 1] if line_no <= len(lines) else ""
                matching_links.append({"line": line_no, "snippet": line_text.strip()[:120]})

        if matching_links:
            stat = md_file.stat()
            backlinks.append({
                "file": str(rel_path).replace("\\", "/"),
                "links": matching_links,
                "link_count": len(matching_links),
                "modified": _compact_date(stat.st_mtime),
            })

    backlinks.sort(key=lambda b: b["link_count"], reverse=True)

    return {
        "vault": vault_name,
        "target": doc_path.lstrip("/\\").replace("\\", "/"),
        "backlinks": backlinks,
        "total_backlinks": len(backlinks),
    }


@mcp.tool()
def create_document(vault_name: str, doc_path: str, content: str) -> dict[str, Any]:
    """Create a new markdown document in a vault.

    Args:
        vault_name: Name of the vault.
        doc_path: Relative path for the new document (must end in .md).
        content: Markdown content for the document.

    Returns the created 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 target.exists():
        return {"error": f"Document already exists: {doc_path}", "vault": vault_name}

    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": clean_path,
        "action": "created",
        "size_bytes": target.stat().st_size,
    }


@mcp.tool()
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: COMPLETE new markdown content for the whole document.

    Returns the replaced 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}

    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("\\", "/"),
        "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.

    Use before create_document when building out folder structures, or for
    organizing notes into subdirectories.

    Args:
        vault_name: Name of the vault.
        dir_path: Relative path for the new directory (e.g. 'Session-12' or 'NPCs/Dragon-Cult').

    Returns the created directory path.
    """
    # Normalize: strip leading slash, handle backslashes
    clean = dir_path.lstrip("/\\").replace("\\", "/")
    target = _resolve_doc(vault_name, clean)

    if target.exists():
        if not target.is_dir():
            return {"error": f"Path exists but is not a directory: {dir_path}", "vault": vault_name}
        return {"vault": vault_name, "path": str(target.relative_to(_vault_path(vault_name))).replace("\\", "/"),
                "action": "exists", "note": "Directory already exists"}

    target.mkdir(parents=True, exist_ok=True)

    return {
        "vault": vault_name,
        "path": str(target.relative_to(_vault_path(vault_name))).replace("\\", "/"),
        "action": "created",
    }


@mcp.tool()
def move_document(vault_name: str, src_path: str, dest_path: str) -> dict[str, Any]:
    """FILESYSTEM TOOL (not memory) — Move or rename a document within a vault.

    Moves a document from src_path to dest_path. Both paths are relative to the
    vault root. Automatically creates parent directories at the destination if
    they don't exist.

    Args:
        vault_name: Name of the vault.
        src_path: Current relative path of the document (e.g. 'Notes/draft.md').
        dest_path: Target relative path (e.g. 'Archive/draft.md' or 'Notes/final.md').

    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}
    if not src.is_file():
        return {"error": f"Source is not a file: {src_path}", "vault": vault_name}

    dest = _resolve_doc(vault_name, dest_path)
    if dest.exists():
        return {"error": f"Destination already exists: {dest_path}", "vault": vault_name}

    dest.parent.mkdir(parents=True, exist_ok=True)
    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": clean_src,
        "new_path": str(dest.relative_to(vault)).replace("\\", "/"),
        "action": "moved",
    }


@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.
    """
    _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")
    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.
    """
    _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")
    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")
    else:
        # 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)
        new_content = new_fm_block + "\n\n" + body
        target.write_text(new_content, encoding="utf-8")

    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)

    return {"vault": vault_name, "path": clean_path, "key": key, "value": value, "action": action}


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():
    parser = argparse.ArgumentParser(description="Obsidian MCP Server")
    parser.add_argument(
        "--vaults-root",
        default=None,
        help="Root directory containing Obsidian vaults (e.g. D:\\Obsidian)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=8765,
        help="Port for StreamableHTTP transport (default: 8765)",
    )
    parser.add_argument(
        "--host",
        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)

    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__":
    main()