helm
31a288933f
Ulysses truncates MCP tool descriptions to 120 chars in prompt text. Moved '(not memory)' to first line of all four tool docstrings so the disambiguation survives truncation. Kept Args/Returns sections intact.
497 lines
No EOL
17 KiB
Python
497 lines
No EOL
17 KiB
Python
"""
|
|
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]:
|
|
"""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,
|
|
) -> dict[str, Any]:
|
|
"""FILESYSTEM TOOL (not memory) — List files and folders in vault on disk. See structure before reading docs.
|
|
|
|
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]:
|
|
"""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.
|
|
"""
|
|
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]:
|
|
"""FILESYSTEM TOOL (not memory) — Search markdown files on disk. Find notes about characters, locations, concepts.
|
|
|
|
Args:
|
|
vault_name: Name of the vault to search.
|
|
query: Search query (case-insensitive substring match).
|
|
max_results: Maximum number of match results to return (default 50).
|
|
|
|
Returns matches grouped by file with line numbers and context snippets.
|
|
"""
|
|
vault = _vault_path(vault_name)
|
|
results = []
|
|
files_searched = 0
|
|
|
|
for md_file in vault.rglob("*.md"):
|
|
# 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() |