helm/obsidian-mcp-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial commit: Obsidian MCP Server

Browse source 
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)
This commit is contained in:
helm 2026-07-04 10:10:30 -04:00
commit 37df28e159
6 changed files with 691 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
.gitignore vendored Normal file
View file

@ -0,0 +1,14 @@
__pycache__/
*.py[cod]
*$py.class
*.so
.env
.venv/
env/
venv/
*.egg-info/
dist/
build/
.DS_Store
.vscode/
.idea/

147
README.md Normal file
View file

@ -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
```

3
config.example.json Normal file
View file

@ -0,0 +1,3 @@
{
"vaults_root": "D:\\Obsidian"
}

1
requirements.txt Normal file
View file

@ -0,0 +1 @@
fastmcp>=2.0.0

500
server.py Normal file
View file

@ -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()

26
start.bat Normal file
View file

@ -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