feat: implement MCP server with vault management

backend/src/mcp/__init__.py

@@ -1 +1,5 @@
 """MCP (Model Context Protocol) server implementation."""
+
+from .server import mcp
+
+__all__ = ["mcp"]

backend/src/mcp/server.py

@@ -0,0 +1,138 @@
+"""FastMCP server exposing vault and indexing tools."""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Dict, List, Optional
+
+from fastmcp import FastMCP
+from pydantic import Field
+
+from ..services import IndexerService, VaultNote, VaultService
+
+mcp = FastMCP(
+    "obsidian-docs-viewer",
+    instructions="Interact with a multi-tenant Obsidian-like documentation vault.",
+)
+
+vault_service = VaultService()
+indexer_service = IndexerService()
+
+
+def _current_user_id() -> str:
+    """Resolve the acting user ID (local mode defaults to local-dev)."""
+    return os.getenv("LOCAL_USER_ID", "local-dev")
+
+
+def _note_to_response(note: VaultNote) -> Dict[str, Any]:
+    return {
+        "path": note["path"],
+        "title": note["title"],
+        "metadata": dict(note.get("metadata") or {}),
+        "body": note.get("body", ""),
+    }
+
+
+@mcp.tool(name="list_notes", description="List notes in the vault (optionally scoped to a folder).")
+def list_notes(
+    folder: Optional[str] = Field(
+        default=None,
+        description="Optional folder path (relative) to filter results. Example: 'api'.",
+    ),
+) -> List[Dict[str, Any]]:
+    user_id = _current_user_id()
+    notes = vault_service.list_notes(user_id, folder=folder)
+    return [
+        {
+            "path": entry["path"],
+            "title": entry["title"],
+            "last_modified": entry["last_modified"].isoformat(),
+        }
+        for entry in notes
+    ]
+
+
+@mcp.tool(name="read_note", description="Read a Markdown note with metadata and body.")
+def read_note(
+    path: str = Field(..., description="Relative path to the note (must include .md)"),
+) -> Dict[str, Any]:
+    user_id = _current_user_id()
+    note = vault_service.read_note(user_id, path)
+    return _note_to_response(note)
+
+
+@mcp.tool(
+    name="write_note",
+    description="Create or update a note. Automatically updates frontmatter timestamps and search index.",
+)
+def write_note(
+    path: str = Field(..., description="Relative note path (includes .md)"),
+    body: str = Field(..., description="Markdown body content"),
+    title: Optional[str] = Field(
+        default=None,
+        description="Optional title override. Defaults to frontmatter title or first heading.",
+    ),
+    metadata: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="Optional frontmatter metadata dictionary (tags, project, etc.).",
+    ),
+) -> Dict[str, Any]:
+    user_id = _current_user_id()
+    note = vault_service.write_note(
+        user_id,
+        path,
+        title=title,
+        metadata=metadata,
+        body=body,
+    )
+    indexer_service.index_note(user_id, note)
+    return {"status": "ok", "path": path}
+
+
+@mcp.tool(name="delete_note", description="Delete a note and remove it from the index.")
+def delete_note(
+    path: str = Field(..., description="Relative note path (includes .md)"),
+) -> Dict[str, str]:
+    user_id = _current_user_id()
+    vault_service.delete_note(user_id, path)
+    indexer_service.delete_note_index(user_id, path)
+    return {"status": "ok"}
+
+
+@mcp.tool(
+    name="search_notes",
+    description="Full-text search with snippets and recency-aware scoring.",
+)
+def search_notes(
+    query: str = Field(..., description="Search query (minimum 1 character)"),
+    limit: int = Field(50, ge=1, le=100, description="Maximum number of results to return."),
+) -> List[Dict[str, Any]]:
+    user_id = _current_user_id()
+    results = indexer_service.search_notes(user_id, query, limit=limit)
+    return [
+        {
+            "path": row["path"],
+            "title": row["title"],
+            "snippet": row["snippet"],
+        }
+        for row in results
+    ]
+
+
+@mcp.tool(name="get_backlinks", description="List notes that reference the target note.")
+def get_backlinks(
+    path: str = Field(..., description="Target note path (includes .md)"),
+) -> List[Dict[str, Any]]:
+    user_id = _current_user_id()
+    backlinks = indexer_service.get_backlinks(user_id, path)
+    return backlinks
+
+
+@mcp.tool(name="get_tags", description="List tags and associated note counts.")
+def get_tags() -> List[Dict[str, Any]]:
+    user_id = _current_user_id()
+    return indexer_service.get_tags(user_id)
+
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

backend/src/services/__init__.py

@@ -3,7 +3,8 @@
 from .auth import AuthError, AuthService
 from .config import AppConfig, get_config, reload_config
 from .database import DatabaseService, init_database
-from .vault import VaultService, sanitize_path, validate_note_path
+from .indexer import IndexerService, normalize_slug, normalize_tag
+from .vault import VaultNote, VaultService, sanitize_path, validate_note_path
 
 __all__ = [
     "AppConfig",
@@ -14,6 +15,10 @@ __all__ = [
     "AuthService",
     "AuthError",
     "VaultService",
+    "VaultNote",
     "sanitize_path",
     "validate_note_path",
+    "IndexerService",
+    "normalize_slug",
+    "normalize_tag",
 ]

backend/src/services/indexer.py

@@ -0,0 +1,376 @@
+"""SQLite-backed indexing utilities for notes."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+import re
+import sqlite3
+from typing import Any, Dict, List, Sequence
+
+from .database import DatabaseService
+from .vault import VaultNote
+
+WIKILINK_PATTERN = re.compile(r"\[\[([^\]]+)\]\]")
+
+
+def _utcnow_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+def normalize_slug(text: str | None) -> str:
+    """Normalize text into a slug suitable for wikilink matching."""
+    if not text:
+        return ""
+    slug = text.lower()
+    slug = re.sub(r"[\s_]+", "-", slug)
+    slug = re.sub(r"[^a-z0-9-]", "", slug)
+    slug = re.sub(r"-+", "-", slug)
+    return slug.strip("-")
+
+
+def normalize_tag(tag: str | None) -> str:
+    if not isinstance(tag, str):
+        return ""
+    return tag.strip().lower()
+
+
+class IndexerService:
+    """Manage SQLite-backed metadata, tags, search index, and link graph."""
+
+    def __init__(self, db_service: DatabaseService | None = None) -> None:
+        self.db_service = db_service or DatabaseService()
+
+    def index_note(self, user_id: str, note: VaultNote) -> int:
+        """Insert or update index rows for a note."""
+        note_path = note["path"]
+        metadata = dict(note.get("metadata") or {})
+        title = note.get("title") or metadata.get("title") or Path(note_path).stem
+        body = note.get("body", "") or ""
+        size_bytes = int(note.get("size_bytes") or len(body.encode("utf-8")))
+        created = str(metadata.get("created") or _utcnow_iso())
+        updated = str(metadata.get("updated") or _utcnow_iso())
+
+        normalized_title_slug = normalize_slug(title)
+        normalized_path_slug = normalize_slug(Path(note_path).stem)
+        if not normalized_title_slug:
+            normalized_title_slug = normalized_path_slug
+
+        tags = self._prepare_tags(metadata.get("tags"))
+        wikilinks = self.extract_wikilinks(body)
+
+        conn = self.db_service.connect()
+        try:
+            with conn:
+                version = self.increment_version(conn, user_id, note_path)
+                self._delete_current_entries(conn, user_id, note_path)
+
+                conn.execute(
+                    """
+                    INSERT INTO note_metadata (
+                        user_id, note_path, version, title,
+                        created, updated, size_bytes,
+                        normalized_title_slug, normalized_path_slug
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    (
+                        user_id,
+                        note_path,
+                        version,
+                        title,
+                        created,
+                        updated,
+                        size_bytes,
+                        normalized_title_slug,
+                        normalized_path_slug,
+                    ),
+                )
+
+                conn.execute(
+                    """
+                    INSERT INTO note_fts (user_id, note_path, title, body)
+                    VALUES (?, ?, ?, ?)
+                    """,
+                    (user_id, note_path, title, body),
+                )
+
+                if tags:
+                    conn.executemany(
+                        """
+                        INSERT INTO note_tags (user_id, note_path, tag)
+                        VALUES (?, ?, ?)
+                        """,
+                        [(user_id, note_path, tag) for tag in tags],
+                    )
+
+                if wikilinks:
+                    resolved = self.resolve_wikilinks(conn, user_id, note_path, wikilinks)
+                    conn.executemany(
+                        """
+                        INSERT INTO note_links (user_id, source_path, target_path, link_text, is_resolved)
+                        VALUES (?, ?, ?, ?, ?)
+                        """,
+                        [
+                            (
+                                user_id,
+                                note_path,
+                                entry["target_path"],
+                                entry["link_text"],
+                                1 if entry["is_resolved"] else 0,
+                            )
+                            for entry in resolved
+                        ],
+                    )
+
+                self.update_index_health(conn, user_id)
+
+            return version
+        finally:
+            conn.close()
+
+    def delete_note_index(self, user_id: str, note_path: str) -> None:
+        """Remove all index data for a note and update backlinks."""
+        conn = self.db_service.connect()
+        try:
+            with conn:
+                self._delete_current_entries(conn, user_id, note_path)
+                conn.execute(
+                    """
+                    UPDATE note_links
+                    SET target_path = NULL, is_resolved = 0
+                    WHERE user_id = ? AND target_path = ?
+                    """,
+                    (user_id, note_path),
+                )
+                self.update_index_health(conn, user_id)
+        finally:
+            conn.close()
+
+    def extract_wikilinks(self, body: str) -> List[str]:
+        """Extract wikilink text from Markdown body."""
+        links = []
+        for match in WIKILINK_PATTERN.finditer(body or ""):
+            link_text = match.group(1).strip()
+            if link_text:
+                links.append(link_text)
+        # Preserve order but drop duplicates
+        seen: Dict[str, None] = {}
+        for link in links:
+            if link not in seen:
+                seen[link] = None
+        return list(seen.keys())
+
+    def resolve_wikilinks(
+        self,
+        conn: sqlite3.Connection,
+        user_id: str,
+        note_path: str,
+        link_texts: Sequence[str],
+    ) -> List[Dict[str, Any]]:
+        """Resolve wikilinks to target note paths using slug comparison."""
+        if not link_texts:
+            return []
+
+        results: List[Dict[str, Any]] = []
+        note_folder = Path(note_path).parent
+
+        for text in link_texts:
+            slug = normalize_slug(text)
+            if not slug:
+                results.append({"link_text": text, "target_path": None, "is_resolved": False})
+                continue
+
+            rows = conn.execute(
+                """
+                SELECT note_path
+                FROM note_metadata
+                WHERE user_id = ?
+                  AND (normalized_title_slug = ? OR normalized_path_slug = ?)
+                """,
+                (user_id, slug, slug),
+            ).fetchall()
+
+            if not rows:
+                results.append({"link_text": text, "target_path": None, "is_resolved": False})
+                continue
+
+            candidates = [row["note_path"] if isinstance(row, sqlite3.Row) else row[0] for row in rows]
+            target = sorted(
+                candidates,
+                key=lambda candidate: (Path(candidate).parent != note_folder, candidate),
+            )[0]
+
+            results.append({"link_text": text, "target_path": target, "is_resolved": True})
+
+        return results
+
+    def increment_version(
+        self, conn: sqlite3.Connection, user_id: str, note_path: str
+    ) -> int:
+        """Return the next version number for a note."""
+        row = conn.execute(
+            "SELECT version FROM note_metadata WHERE user_id = ? AND note_path = ?",
+            (user_id, note_path),
+        ).fetchone()
+        if row is None:
+            return 1
+        current_version = row["version"] if isinstance(row, sqlite3.Row) else row[0]
+        return int(current_version) + 1
+
+    def update_index_health(self, conn: sqlite3.Connection, user_id: str) -> None:
+        """Update per-user index health stats."""
+        row = conn.execute(
+            "SELECT COUNT(*) AS count FROM note_metadata WHERE user_id = ?",
+            (user_id,),
+        ).fetchone()
+        note_count = int(row["count"] if isinstance(row, sqlite3.Row) else row[0])
+        now_iso = _utcnow_iso()
+        conn.execute(
+            """
+            INSERT INTO index_health (user_id, note_count, last_incremental_update)
+            VALUES (?, ?, ?)
+            ON CONFLICT(user_id) DO UPDATE SET
+                note_count = excluded.note_count,
+                last_incremental_update = excluded.last_incremental_update
+            """,
+            (user_id, note_count, now_iso),
+        )
+
+    def search_notes(self, user_id: str, query: str, *, limit: int = 50) -> List[Dict[str, Any]]:
+        """Execute a full-text search with recency bonus scoring."""
+        if not query or not query.strip():
+            raise ValueError("Search query cannot be empty")
+
+        conn = self.db_service.connect()
+        try:
+            rows = conn.execute(
+                """
+                SELECT
+                    m.note_path,
+                    m.title,
+                    m.updated,
+                    snippet(note_fts, 3, '<mark>', '</mark>', '...', 32) AS snippet,
+                    bm25(note_fts, 3.0, 1.0) AS score
+                FROM note_fts
+                JOIN note_metadata m USING (user_id, note_path)
+                WHERE note_fts.user_id = ? AND note_fts MATCH ?
+                ORDER BY score DESC
+                LIMIT ?
+                """,
+                (user_id, query, limit),
+            ).fetchall()
+        finally:
+            conn.close()
+
+        now = datetime.now(timezone.utc)
+        seven_days = timedelta(days=7)
+        thirty_days = timedelta(days=30)
+
+        results: List[Dict[str, Any]] = []
+        for row in rows:
+            updated_raw = row["updated"] if isinstance(row, sqlite3.Row) else row[2]
+            snippet = row["snippet"] if isinstance(row, sqlite3.Row) else row[3]
+            base_score = float(row["score"] if isinstance(row, sqlite3.Row) else row[4])
+            try:
+                updated_dt = datetime.fromisoformat(str(updated_raw))
+            except ValueError:
+                updated_dt = now
+            delta = now - updated_dt
+            if delta <= seven_days:
+                bonus = 1.0
+            elif delta <= thirty_days:
+                bonus = 0.5
+            else:
+                bonus = 0.0
+
+            results.append(
+                {
+                    "path": row["note_path"] if isinstance(row, sqlite3.Row) else row[0],
+                    "title": row["title"] if isinstance(row, sqlite3.Row) else row[1],
+                    "snippet": snippet or "",
+                    "score": base_score + bonus,
+                    "updated": updated_raw,
+                }
+            )
+
+        return sorted(results, key=lambda item: item["score"], reverse=True)
+
+    def get_backlinks(self, user_id: str, target_path: str) -> List[Dict[str, Any]]:
+        """Return backlinks for a note."""
+        conn = self.db_service.connect()
+        try:
+            rows = conn.execute(
+                """
+                SELECT DISTINCT l.source_path, m.title
+                FROM note_links l
+                JOIN note_metadata m
+                  ON l.user_id = m.user_id AND l.source_path = m.note_path
+                WHERE l.user_id = ? AND l.target_path = ?
+                ORDER BY m.updated DESC
+                """,
+                (user_id, target_path),
+            ).fetchall()
+        finally:
+            conn.close()
+
+        return [
+            {
+                "path": row["source_path"] if isinstance(row, sqlite3.Row) else row[0],
+                "title": row["title"] if isinstance(row, sqlite3.Row) else row[1],
+            }
+            for row in rows
+        ]
+
+    def get_tags(self, user_id: str) -> List[Dict[str, Any]]:
+        """Return tag counts for a user."""
+        conn = self.db_service.connect()
+        try:
+            rows = conn.execute(
+                """
+                SELECT tag, COUNT(DISTINCT note_path) AS count
+                FROM note_tags
+                WHERE user_id = ?
+                GROUP BY tag
+                ORDER BY count DESC, tag ASC
+                """,
+                (user_id,),
+            ).fetchall()
+        finally:
+            conn.close()
+
+        return [
+            {"tag": row["tag"] if isinstance(row, sqlite3.Row) else row[0], "count": int(row["count"] if isinstance(row, sqlite3.Row) else row[1])}
+            for row in rows
+        ]
+
+    def _delete_current_entries(self, conn: sqlite3.Connection, user_id: str, note_path: str) -> None:
+        """Delete existing index rows for a note."""
+        conn.execute(
+            "DELETE FROM note_metadata WHERE user_id = ? AND note_path = ?",
+            (user_id, note_path),
+        )
+        conn.execute(
+            "DELETE FROM note_fts WHERE user_id = ? AND note_path = ?",
+            (user_id, note_path),
+        )
+        conn.execute(
+            "DELETE FROM note_tags WHERE user_id = ? AND note_path = ?",
+            (user_id, note_path),
+        )
+        conn.execute(
+            "DELETE FROM note_links WHERE user_id = ? AND source_path = ?",
+            (user_id, note_path),
+        )
+
+    def _prepare_tags(self, tags: Any) -> List[str]:
+        if not isinstance(tags, list):
+            return []
+        normalized: List[str] = []
+        for tag in tags:
+            cleaned = normalize_tag(tag)
+            if cleaned and cleaned not in normalized:
+                normalized.append(cleaned)
+        return normalized
+
+
+__all__ = ["IndexerService", "normalize_slug", "normalize_tag"]

backend/src/services/vault.py

@@ -2,12 +2,20 @@
 
 from __future__ import annotations
 
+from datetime import datetime, timezone
 from pathlib import Path
-from typing import Tuple
+import re
+from typing import Any, Dict, List, Tuple
+
+import frontmatter
 
 from .config import AppConfig, get_config
 
 INVALID_PATH_CHARS = {'<', '>', ':', '"', '|', '?', '*'}
+MAX_NOTE_BYTES = 1_048_576
+H1_PATTERN = re.compile(r"^\s*#\s+(.+)$", re.MULTILINE)
+
+VaultNote = Dict[str, Any]
 
 
 def validate_note_path(note_path: str) -> Tuple[bool, str]:
@@ -44,6 +52,42 @@ def sanitize_path(user_id: str, vault_root: Path, note_path: str) -> Path:
     return full_path
 
 
+def _utcnow_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+def _validate_frontmatter(metadata: Dict[str, Any]) -> Dict[str, Any]:
+    reserved = {"version"}
+    for key in metadata.keys():
+        if key in reserved:
+            raise ValueError(f"Field '{key}' is reserved and cannot be set in frontmatter")
+    tags = metadata.get("tags")
+    if tags is not None:
+        if not isinstance(tags, list):
+            raise ValueError("Field 'tags' must be an array")
+        if not all(isinstance(tag, str) for tag in tags):
+            raise ValueError("All tags must be strings")
+    return metadata
+
+
+def _validate_note_body(body: str) -> None:
+    body_bytes = body.encode("utf-8")
+    if len(body_bytes) > MAX_NOTE_BYTES:
+        raise ValueError("Note exceeds 1 MiB limit")
+
+
+def _derive_title(note_path: str, metadata: Dict[str, Any], body: str) -> str:
+    title = metadata.get("title")
+    if isinstance(title, str) and title.strip():
+        return title.strip()
+    match = H1_PATTERN.search(body or "")
+    if match:
+        return match.group(1).strip()
+    stem = Path(note_path).stem
+    title_from_filename = stem.replace("-", " ").replace("_", " ").strip()
+    return title_from_filename or stem
+
+
 class VaultService:
     """Service for managing vault directories and basic path validation."""
 
@@ -69,5 +113,123 @@ class VaultService:
             raise ValueError(message)
         return sanitize_path(user_id, self.vault_root, note_path)
 
+    def read_note(self, user_id: str, note_path: str) -> VaultNote:
+        """Read a Markdown note, returning metadata, body, and derived title."""
+        base = self.initialize_vault(user_id)
+        absolute_path = self.resolve_note_path(user_id, note_path)
+        if not absolute_path.exists():
+            raise FileNotFoundError(f"Note not found: {note_path}")
+        post = frontmatter.load(absolute_path)
+        metadata = dict(post.metadata or {})
+        body = post.content or ""
+        return self._build_note_payload(note_path, metadata, body, absolute_path)
+
+    def write_note(
+        self,
+        user_id: str,
+        note_path: str,
+        *,
+        title: str | None = None,
+        metadata: Dict[str, Any] | None = None,
+        body: str,
+    ) -> VaultNote:
+        """Create or update a note with validated metadata and content."""
+        absolute_path = self.resolve_note_path(user_id, note_path)
+        body = body or ""
+        _validate_note_body(body)
+
+        metadata_dict: Dict[str, Any] = dict(metadata or {})
+        _validate_frontmatter(metadata_dict)
+
+        existing_created: str | None = None
+        if absolute_path.exists():
+            try:
+                current = frontmatter.load(absolute_path)
+                current_created = current.metadata.get("created")
+                if isinstance(current_created, str):
+                    existing_created = current_created
+            except Exception:
+                existing_created = None
+
+        effective_title = title or metadata_dict.get("title")
+        if not effective_title:
+            effective_title = _derive_title(note_path, metadata_dict, body)
+        metadata_dict["title"] = effective_title
+
+        now_iso = _utcnow_iso()
+        metadata_dict.setdefault("created", existing_created or metadata_dict.get("created") or now_iso)
+        metadata_dict["updated"] = now_iso
+
+        absolute_path.parent.mkdir(parents=True, exist_ok=True)
+        post = frontmatter.Post(body, **metadata_dict)
+        absolute_path.write_text(frontmatter.dumps(post), encoding="utf-8")
+        return self._build_note_payload(note_path, metadata_dict, body, absolute_path)
+
+    def delete_note(self, user_id: str, note_path: str) -> None:
+        """Delete a note from the vault."""
+        absolute_path = self.resolve_note_path(user_id, note_path)
+        try:
+            absolute_path.unlink()
+        except FileNotFoundError as exc:
+            raise FileNotFoundError(f"Note not found: {note_path}") from exc
+
+    def list_notes(self, user_id: str, folder: str | None = None) -> List[Dict[str, Any]]:
+        """List notes (optionally scoped to a folder) with titles and timestamps."""
+        base = self.initialize_vault(user_id).resolve()
+
+        if folder:
+            cleaned = folder.strip().strip("/")
+            if "\\" in cleaned or ".." in cleaned:
+                raise ValueError("Folder path contains invalid characters")
+            folder_path = (base / cleaned).resolve() if cleaned else base
+            if not str(folder_path).startswith(str(base)):
+                raise ValueError("Folder path escapes vault root")
+            if not folder_path.exists():
+                return []
+            if folder_path.is_file():
+                files = [folder_path] if folder_path.suffix == ".md" else []
+            else:
+                files = list(folder_path.rglob("*.md"))
+        else:
+            files = list(base.rglob("*.md"))
+
+        results: List[Dict[str, Any]] = []
+        for file_path in files:
+            if not file_path.is_file():
+                continue
+            relative_path = file_path.relative_to(base).as_posix()
+            try:
+                post = frontmatter.load(file_path)
+                metadata = dict(post.metadata or {})
+                body = post.content or ""
+                title = _derive_title(relative_path, metadata, body)
+            except Exception:
+                title = Path(relative_path).stem
+            stat = file_path.stat()
+            results.append(
+                {
+                    "path": relative_path,
+                    "title": title,
+                    "last_modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc),
+                }
+            )
+
+        return sorted(results, key=lambda item: item["path"].lower())
+
+    def _build_note_payload(
+        self, note_path: str, metadata: Dict[str, Any], body: str, absolute_path: Path
+    ) -> VaultNote:
+        stat = absolute_path.stat()
+        title = _derive_title(note_path, metadata, body)
+        return {
+            "path": note_path,
+            "title": title,
+            "metadata": metadata,
+            "body": body,
+            "size_bytes": stat.st_size,
+            "modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc),
+            "absolute_path": absolute_path,
+        }
+
 
 __all__ = ["VaultService", "validate_note_path", "sanitize_path"]