From 37df28e1590734377b50f015963af65375523f1c Mon Sep 17 00:00:00 2001 From: helm Date: Sat, 4 Jul 2026 10:10:30 -0400 Subject: [PATCH] Initial commit: Obsidian MCP Server StreamableHTTP MCP server exposing Obsidian vaults as tools. - 8 tools: list_vault, get_vaults, get_vault_document, search_vault, get_recent_changes, get_backlinks, create_document, edit_document - FastMCP framework with path traversal prevention - Cross-OS transport (WSL2->Windows via HTTP) --- .gitignore | 14 ++ README.md | 147 +++++++++++++ config.example.json | 3 + requirements.txt | 1 + server.py | 500 ++++++++++++++++++++++++++++++++++++++++++++ start.bat | 26 +++ 6 files changed, 691 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 config.example.json create mode 100644 requirements.txt create mode 100644 server.py create mode 100644 start.bat diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3385edd --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +__pycache__/ +*.py[cod] +*$py.class +*.so +.env +.venv/ +env/ +venv/ +*.egg-info/ +dist/ +build/ +.DS_Store +.vscode/ +.idea/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..9b2dba2 --- /dev/null +++ b/README.md @@ -0,0 +1,147 @@ +# Obsidian MCP Server + +Exposes Obsidian vaults as MCP (Model Context Protocol) tools. Runs on Windows, reachable from WSL2, Hermes Agent, or any MCP-compatible client. + +## Tools Provided + +| Tool | Description | +|------|-------------| +| `get_vaults` | List all vaults in the vaults root directory | +| `get_vault_directories` | Browse directory structure within a vault | +| `get_vault_document` | Read a markdown note by path | +| `search_vault` | Full-text search across all notes in a vault | +| `get_recent_changes` | Recently modified notes, newest first | +| `get_backlinks` | Find all notes linking to a given note via [[wikilinks]] | +| `create_document` | Create a new markdown note | +| `edit_document` | Overwrite an existing markdown note | + +## Setup (Windows) + +### 1. Prerequisites + +- Python 3.10+ installed on Windows ([python.org](https://python.org)) +- Recommended: `uv` for package management (`pip install uv`) + +### 2. Install + +```cmd +cd C:\path\to\obsidian-mcp-server +pip install -r requirements.txt +``` + +Or with uv: +```cmd +uv pip install -r requirements.txt +``` + +### 3. Configure + +Copy and edit the config file: +```cmd +copy config.example.json config.json +``` + +Edit `config.json` — set `vaults_root` to the folder containing your Obsidian vaults: +```json +{ + "vaults_root": "D:\\Obsidian" +} +``` + +Alternatively, set the environment variable: +```cmd +set OBSIDIAN_VAULTS_ROOT=D:\Obsidian +``` + +Or pass it on the command line: +```cmd +python server.py --vaults-root "D:\Obsidian" +``` + +Priority: CLI arg > env var > config.json + +### 4. Run + +```cmd +python server.py +``` + +Or use the launcher script (auto-restarts on crash): +```cmd +start.bat +``` + +Output: +``` +Obsidian MCP Server + Vaults root: D:\Obsidian + Listening: http://0.0.0.0:8765/mcp + Vaults found: 2 +``` + +### 5. Windows Firewall + +If connecting from WSL2 or another machine, allow Python through Windows Firewall on port 8765: + +```powershell +New-NetFirewallRule -DisplayName "Obsidian MCP" -Direction Inbound -Protocol TCP -LocalPort 8765 -Action Allow +``` + +## Connecting from Hermes Agent (WSL2) + +### Find Windows Host IP from WSL2 + +```bash +# The gateway IP from WSL2 IS the Windows host +ip route | grep default | awk '{print $3}' +# Usually 192.168.1.x or 172.x.x.x +``` + +Typical output: `172.30.112.1` or your LAN IP like `192.168.1.100`. + +### Add to Hermes config + +Edit `~/.hermes/config.yaml`: + +```yaml +mcp_servers: + obsidian: + url: "http://192.168.1.100:8765/mcp" + timeout: 60 + connect_timeout: 30 +``` + +Replace `192.168.1.100` with your actual Windows IP. + +Then reload MCP servers: +``` +/reload-mcp +``` + +Tools will appear as: +- `mcp_obsidian_get_vaults` +- `mcp_obsidian_get_vault_directories` +- `mcp_obsidian_get_vault_document` +- `mcp_obsidian_search_vault` +- `mcp_obsidian_get_recent_changes` +- `mcp_obsidian_get_backlinks` +- `mcp_obsidian_create_document` +- `mcp_obsidian_edit_document` + +## Security + +- Binds to `0.0.0.0` by default for LAN access. On a trusted home network behind NAT, this is fine. +- No authentication — relies on network isolation. Don't expose to the internet. +- Path traversal is prevented — all vault paths are validated to stay within the vault root. +- Hidden directories (`.obsidian`, `.trash`, `.git`) are excluded from listings and search. + +## Example Queries via Hermes + +``` +> List my vaults +> Search the SKT vault for "Klauth" +> Get recent changes in the SKT vault, last 10 +> Read the note "Session-Notes/Session-12" from the SKT vault +> Find all notes linking to "Adonis" in the SKT vault +> Create a new note "Ideas/dragon-lore.md" in my D&D vault +``` \ No newline at end of file diff --git a/config.example.json b/config.example.json new file mode 100644 index 0000000..2e0f969 --- /dev/null +++ b/config.example.json @@ -0,0 +1,3 @@ +{ + "vaults_root": "D:\\Obsidian" +} \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9074148 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +fastmcp>=2.0.0 \ No newline at end of file diff --git a/server.py b/server.py new file mode 100644 index 0000000..c668d6e --- /dev/null +++ b/server.py @@ -0,0 +1,500 @@ +""" +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 +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +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"} + + +# ── Helpers ─────────────────────────────────────────────────────────────── + +def _vault_path(vault_name: str) -> Path: + """Get the absolute path to a vault directory, with safety check.""" + 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}") + 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.""" + 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}") + return target + + +def _file_info(p: Path, relative_to: Path) -> dict[str, Any]: + """Build a file/directory info dict.""" + stat = p.stat() + return { + "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 0, + "modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), + } + + +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]] + 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]: + """List all Obsidian vaults found in the configured vaults root directory. + + 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, +) -> dict[str, Any]: + """List the contents of a vault directory — both files and subdirectories. + + This is THE primary tool for browsing a vault. Use it to discover what + documents exist before calling get_vault_document to read them. + + Args: + vault_name: Name of the vault (subdirectory name under vaults root) + path: Relative path within the vault. Empty string = vault root. + 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: Optional filter — only return files with these extensions + (e.g. ['.md', '.canvas']). Case-insensitive. + + Returns: + directories: list of subdirectory info dicts + files: list of file info dicts (name, path, type, size, modified) + total: combined count + """ + vault = _vault_path(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} + + 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())): + # Skip hidden/obsidian metadata unless explicitly inside .obsidian + if entry.name.startswith(".") and entry.name not in (".obsidian",): + continue + if entry.name in OBSIDIAN_HIDDEN and not str(entry.resolve()).startswith(str(vault / ".obsidian")): + continue + + info = _file_info(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 + 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]: + """Read a markdown document from a vault. + + 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. + """ + 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": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), + } + + +@mcp.tool() +def search_vault(vault_name: str, query: str, max_results: int = 50) -> dict[str, Any]: + """Full-text search across all markdown files in a vault. + + 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"): + # Skip hidden/trash + rel = str(md_file.relative_to(vault)) + if any(part.startswith(".") for part in md_file.relative_to(vault).parts): + continue + if ".trash" in rel or ".obsidian" in rel: + continue + + files_searched += 1 + matches = _search_file(md_file, query) + if matches: + results.append({ + "file": rel.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]: + """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. + """ + vault = _vault_path(vault_name) + md_files = [] + + for md_file in vault.rglob("*.md"): + rel = str(md_file.relative_to(vault)) + if any(part.startswith(".") for part in md_file.relative_to(vault).parts): + continue + if ".trash" in rel or ".obsidian" in rel: + continue + stat = md_file.stat() + md_files.append({ + "path": rel.replace("\\", "/"), + "size_bytes": stat.st_size, + "modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), + }) + + 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. + """ + 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 = str(md_file.relative_to(vault)).replace("\\", "/") + if any(part.startswith(".") for part in md_file.relative_to(vault).parts): + continue + if ".trash" in rel or ".obsidian" in rel: + 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() + # 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: + # 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": rel, + "links": matching_links, + "link_count": len(matching_links), + "modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), + }) + + 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. + """ + 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") + + return { + "vault": vault_name, + "path": doc_path.lstrip("/\\").replace("\\", "/"), + "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. + + Args: + vault_name: Name of the vault. + doc_path: Relative path to the document. + content: New markdown content (replaces entire file). + + Returns the edited 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(): + return {"error": f"Document does not exist: {doc_path}", "vault": vault_name} + + target.write_text(content, encoding="utf-8") + + return { + "vault": vault_name, + "path": doc_path.lstrip("/\\").replace("\\", "/"), + "action": "edited", + "size_bytes": target.stat().st_size, + } + + +# ── 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)", + ) + 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 __name__ == "__main__": + main() \ No newline at end of file diff --git a/start.bat b/start.bat new file mode 100644 index 0000000..887bbfd --- /dev/null +++ b/start.bat @@ -0,0 +1,26 @@ +@echo off +REM Obsidian MCP Server Launcher +REM Edit VAULTS_ROOT below to match your setup, or set OBSIDIAN_VAULTS_ROOT env var. + +setlocal + +REM ── Adjust this to your Obsidian vaults folder ── +set VAULTS_ROOT=D:\Obsidian + +REM ── Restart if it crashes (basic resilience) ── +:loop +echo. +echo ============================================ +echo Obsidian MCP Server +echo Vaults: %VAULTS_ROOT% +echo Port: 8765 +echo Ctrl+C to stop +echo ============================================ +echo. + +python server.py --vaults-root "%VAULTS_ROOT%" --port 8765 + +echo. +echo Server stopped. Restarting in 5 seconds... +timeout /t 5 /nobreak >nul +goto loop \ No newline at end of file