#!/usr/bin/env python3

"""Obsidian Vault MCP Server - Full coverage of the Local REST API v3.5.0 (OpenAPI 3.0.2)."""

import os

import sys

import json

import logging

from typing import Optional

from urllib.parse import quote

import httpx

from mcp.server.fastmcp import FastMCP

# Configure logging to stderr

logging.basicConfig(

level=logging.INFO,

format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",

stream=sys.stderr,

)

logger = logging.getLogger("obsidian-server")

# Initialize MCP server

mcp = FastMCP("obsidian")

# Configuration from environment variables

OBSIDIAN_API_URL = os.environ.get(

"OBSIDIAN_API_URL", "https://host.docker.internal:27124"

)

OBSIDIAN_API_KEY = os.environ.get("OBSIDIAN_API_KEY", "")

HTTP_TIMEOUT = 30.0

VALID_PERIODS = ("daily", "weekly", "monthly", "quarterly", "yearly")

VALID_OPERATIONS = ("append", "prepend", "replace")

VALID_TARGET_TYPES = ("heading", "block", "frontmatter")

def parse_tls_verify(value: str) -> bool:

"""Return whether the HTTP client should verify the API certificate."""

return value.strip().lower() in {"true", "1", "yes", "on"}

TLS_VERIFY = parse_tls_verify(os.environ.get("OBSIDIAN_TLS_VERIFY", "false"))

def build_headers(

content_type: str = "application/json",

extra_headers: Optional[dict] = None,

) -> dict:

"""Build request headers with API key authentication."""

headers = {

"Authorization": f"Bearer {OBSIDIAN_API_KEY}",

"Content-Type": content_type,

}

headers.update(extra_headers or {})

return headers

def clean_path(filepath: str) -> str:

"""Strip whitespace and leading slashes from a vault-relative path."""

return filepath.strip().lstrip("/")

def build_patch_headers(

operation: str,

target_type: str,

target: str,

target_delimiter: str = "::",

trim_target_whitespace: str = "false",

create_if_missing: str = "false",

) -> dict:

"""Build the extra headers required by PATCH endpoints."""

headers = {

"Operation": operation,

"Target-Type": target_type,

"Target": quote(target, safe=""),

"Target-Delimiter": target_delimiter,

}

if trim_target_whitespace.lower() == "true":

headers["Trim-Target-Whitespace"] = "true"

if create_if_missing.lower() == "true":

headers["Create-Target-If-Missing"] = "true"

return headers

async def make_request(

method: str,

endpoint: str,

data: str = "",

content_type: str = "application/json",

extra_headers: Optional[dict] = None,

params: Optional[dict] = None,

) -> str:

"""Make an HTTP request to the Obsidian Local REST API and return the response body."""

url = f"{OBSIDIAN_API_URL}{endpoint}"

headers = build_headers(content_type, extra_headers)

try:

async with httpx.AsyncClient(timeout=HTTP_TIMEOUT, verify=TLS_VERIFY) as client:

if method == "GET":

response = await client.get(url, headers=headers, params=params or {})

elif method == "POST":

response = await client.post(

url, headers=headers, content=data, params=params or {}

)

elif method == "PUT":

response = await client.put(

url, headers=headers, content=data, params=params or {}

)

elif method == "PATCH":

response = await client.patch(

url, headers=headers, content=data, params=params or {}

)

elif method == "DELETE":

response = await client.delete(url, headers=headers, params=params or {})

else:

return f"Unsupported HTTP method: {method}"

response.raise_for_status()

return response.text if response.text else "Operation completed successfully"

except httpx.HTTPStatusError as exc:

status = exc.response.status_code

body = exc.response.text

logger.error("HTTP error %s: %s", status, body)

return f"API Error {status}: {body}"

except httpx.TimeoutException:

logger.error("Request timeout for %s", url)

return "Request timeout - check if Obsidian Local REST API is running"

except Exception as exc:

logger.error("Request error: %s", str(exc))

return f"Error: {str(exc)}"

# ── System Tools ─────────────────────────────────────────────────────────────

@mcp.tool()

async def get_server_status() -> str:

"""Get server status, version info, and authentication status."""

logger.info("Getting server status")

return await make_request("GET", "/")

@mcp.tool()

async def get_certificate() -> str:

"""Download the self-signed certificate used by the Local REST API."""

logger.info("Getting API certificate")

return await make_request("GET", "/obsidian-local-rest-api.crt")

@mcp.tool()

async def get_openapi_spec() -> str:

"""Get the OpenAPI YAML spec describing the Local REST API capabilities."""

logger.info("Getting OpenAPI spec")

return await make_request("GET", "/openapi.yaml")

# ── Vault File Tools ─────────────────────────────────────────────────────────

@mcp.tool()

async def read_note(filepath: str = "", include_metadata: str = "false") -> str:

"""Read a note from the vault. Set include_metadata to 'true' for tags, frontmatter, and file stats."""

if not filepath.strip():

return "Error: filepath is required"

cleaned = clean_path(filepath)

logger.info("Reading note: %s", cleaned)

extra = {}

if include_metadata.lower() == "true":

extra["Accept"] = "application/vnd.olrapi.note+json"

return await make_request("GET", f"/vault/{cleaned}", extra_headers=extra)

@mcp.tool()

async def get_document_map(filepath: str = "") -> str:

"""Get the structure of a note: headings, block references, and frontmatter field names."""

if not filepath.strip():

return "Error: filepath is required"

cleaned = clean_path(filepath)

logger.info("Getting document map: %s", cleaned)

extra = {"Accept": "application/vnd.olrapi.document-map+json"}

return await make_request("GET", f"/vault/{cleaned}", extra_headers=extra)

@mcp.tool()

async def create_or_replace_note(filepath: str = "", content: str = "") -> str:

"""Create a new note or replace the content of an existing one."""

if not filepath.strip():

return "Error: filepath is required"

cleaned = clean_path(filepath)

logger.info("Creating/replacing note: %s", cleaned)

return await make_request("PUT", f"/vault/{cleaned}", content, "text/markdown")

@mcp.tool()

async def append_to_note(filepath: str = "", content: str = "") -> str:

"""Append content to the end of an existing note. Creates the file if it does not exist."""

if not filepath.strip():

return "Error: filepath is required"

cleaned = clean_path(filepath)

logger.info("Appending to note: %s", cleaned)

return await make_request("POST", f"/vault/{cleaned}", content, "text/markdown")

@mcp.tool()

async def patch_note(

filepath: str = "",

content: str = "",

operation: str = "append",

target_type: str = "heading",

target: str = "",

target_delimiter: str = "::",

trim_target_whitespace: str = "false",

create_if_missing: str = "false",

content_type: str = "",

) -> str:

"""Patch a note relative to a heading, block reference, or frontmatter field. Operations: append, prepend, replace. Use content_type 'application/json' for table rows or frontmatter values."""

if not filepath.strip():

return "Error: filepath is required"

if not target.strip():

return "Error: target is required"

if operation not in VALID_OPERATIONS:

return f"Error: operation must be one of {VALID_OPERATIONS}"

if target_type not in VALID_TARGET_TYPES:

return f"Error: target_type must be one of {VALID_TARGET_TYPES}"

cleaned = clean_path(filepath)

logger.info("Patching note %s: %s at %s=%s", cleaned, operation, target_type, target)

ct = content_type if content_type else (

"application/json" if target_type == "frontmatter" else "text/markdown"

)

extra = build_patch_headers(

operation, target_type, target, target_delimiter,

trim_target_whitespace, create_if_missing,

)

return await make_request("PATCH", f"/vault/{cleaned}", content, ct, extra_headers=extra)

@mcp.tool()

async def delete_note(filepath: str = "") -> str:

"""Delete a note from the vault."""

if not filepath.strip():

return "Error: filepath is required"

cleaned = clean_path(filepath)

logger.info("Deleting note: %s", cleaned)

return await make_request("DELETE", f"/vault/{cleaned}")

# ── Vault Directory Tools ────────────────────────────────────────────────────

@mcp.tool()

async def list_vault_root() -> str:

"""List all files and folders in the vault root directory."""

logger.info("Listing vault root")

return await make_request("GET", "/vault/")

@mcp.tool()

async def list_directory(folder: str = "") -> str:

"""List files and folders inside a specific vault directory."""

if not folder.strip():

return "Error: folder is required (use list_vault_root for the root)"

cleaned = clean_path(folder).rstrip("/")

logger.info("Listing directory: %s", cleaned)

return await make_request("GET", f"/vault/{cleaned}/")

# ── Active File Tools ────────────────────────────────────────────────────────

@mcp.tool()

async def get_active_file(include_metadata: str = "false") -> str:

"""Get the currently active file in Obsidian. Set include_metadata to 'true' for tags, frontmatter, and stats."""

logger.info("Getting active file")

extra = {}

if include_metadata.lower() == "true":

extra["Accept"] = "application/vnd.olrapi.note+json"

return await make_request("GET", "/active/", extra_headers=extra)

@mcp.tool()

async def get_active_file_map() -> str:

"""Get the document map of the active file: headings, block references, and frontmatter fields."""

logger.info("Getting active file document map")

extra = {"Accept": "application/vnd.olrapi.document-map+json"}

return await make_request("GET", "/active/", extra_headers=extra)

@mcp.tool()

async def replace_active_file(content: str = "") -> str:

"""Replace the entire content of the currently active file."""

if not content.strip():

return "Error: content is required"

logger.info("Replacing active file content")

return await make_request("PUT", "/active/", content, "text/markdown")

@mcp.tool()

async def append_to_active_file(content: str = "") -> str:

"""Append content to the end of the currently active file."""

if not content.strip():

return "Error: content is required"

logger.info("Appending to active file")

return await make_request("POST", "/active/", content, "text/markdown")

@mcp.tool()

async def patch_active_file(

content: str = "",

operation: str = "append",

target_type: str = "heading",

target: str = "",

target_delimiter: str = "::",

trim_target_whitespace: str = "false",

create_if_missing: str = "false",

content_type: str = "",

) -> str:

"""Patch the active file relative to a heading, block reference, or frontmatter field. Operations: append, prepend, replace."""

if not target.strip():

return "Error: target is required"

if operation not in VALID_OPERATIONS:

return f"Error: operation must be one of {VALID_OPERATIONS}"

if target_type not in VALID_TARGET_TYPES:

return f"Error: target_type must be one of {VALID_TARGET_TYPES}"

logger.info("Patching active file: %s at %s=%s", operation, target_type, target)

ct = content_type if content_type else (

"application/json" if target_type == "frontmatter" else "text/markdown"

)

extra = build_patch_headers(

operation, target_type, target, target_delimiter,

trim_target_whitespace, create_if_missing,

)

return await make_request("PATCH", "/active/", content, ct, extra_headers=extra)

@mcp.tool()

async def delete_active_file() -> str:

"""Delete the currently active file in Obsidian."""

logger.info("Deleting active file")

return await make_request("DELETE", "/active/")

# ── Search Tools ─────────────────────────────────────────────────────────────

@mcp.tool()

async def search_simple(query: str = "", context_length: str = "100") -> str:

"""Simple text search across the vault. Returns matching filenames with surrounding context."""

if not query.strip():

return "Error: query is required"

logger.info("Simple search: %s", query)

params = {"query": query.strip(), "contextLength": context_length}

return await make_request("POST", "/search/simple/", params=params)

@mcp.tool()

async def search_dataview(dql_query: str = "") -> str:

"""Run a Dataview DQL TABLE query. Requires the Dataview plugin installed in Obsidian."""

if not dql_query.strip():

return "Error: dql_query is required"

logger.info("Dataview query: %s", dql_query)

return await make_request(

"POST", "/search/", dql_query.strip(),

"application/vnd.olrapi.dataview.dql+txt",

)

@mcp.tool()

async def search_jsonlogic(query_json: str = "") -> str:

"""Run a JsonLogic query to search notes by tags, frontmatter, path, or stats. Pass as a JSON string."""

if not query_json.strip():

return "Error: query_json is required"

logger.info("JsonLogic query")

return await make_request(

"POST", "/search/", query_json.strip(),

"application/vnd.olrapi.jsonlogic+json",

)

# ── Commands Tools ───────────────────────────────────────────────────────────

@mcp.tool()

async def list_commands() -> str:

"""List all available Obsidian commands including those from installed plugins."""

logger.info("Listing commands")

return await make_request("GET", "/commands/")

@mcp.tool()

async def execute_command(command_id: str = "") -> str:

"""Execute an Obsidian command by its ID (e.g. 'daily-notes:open-today', 'graph:open')."""

if not command_id.strip():

return "Error: command_id is required"

cleaned = clean_path(command_id)

logger.info("Executing command: %s", cleaned)

return await make_request("POST", f"/commands/{cleaned}/")

# ── Periodic Notes Tools (current period) ────────────────────────────────────

@mcp.tool()

async def get_current_periodic_note(

period: str = "daily", include_metadata: str = "false"

) -> str:

"""Get the current periodic note (daily, weekly, monthly, quarterly, yearly)."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

logger.info("Getting current %s note", period)

extra = {}

if include_metadata.lower() == "true":

extra["Accept"] = "application/vnd.olrapi.note+json"

return await make_request("GET", f"/periodic/{period}/", extra_headers=extra)

@mcp.tool()

async def get_current_periodic_note_map(period: str = "daily") -> str:

"""Get the document map of the current periodic note: headings, block references, frontmatter fields."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

logger.info("Getting current %s note map", period)

extra = {"Accept": "application/vnd.olrapi.document-map+json"}

return await make_request("GET", f"/periodic/{period}/", extra_headers=extra)

@mcp.tool()

async def create_current_periodic_note(

period: str = "daily", content: str = ""

) -> str:

"""Create or replace the current periodic note."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

logger.info("Creating current %s note", period)

return await make_request("PUT", f"/periodic/{period}/", content, "text/markdown")

@mcp.tool()

async def append_to_current_periodic_note(

period: str = "daily", content: str = ""

) -> str:

"""Append content to the current periodic note. Creates the note if it does not exist."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

logger.info("Appending to current %s note", period)

return await make_request("POST", f"/periodic/{period}/", content, "text/markdown")

@mcp.tool()

async def patch_current_periodic_note(

period: str = "daily",

content: str = "",

operation: str = "append",

target_type: str = "heading",

target: str = "",

target_delimiter: str = "::",

trim_target_whitespace: str = "false",

create_if_missing: str = "false",

content_type: str = "",

) -> str:

"""Patch the current periodic note relative to a heading, block reference, or frontmatter field."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

if not target.strip():

return "Error: target is required"

if operation not in VALID_OPERATIONS:

return f"Error: operation must be one of {VALID_OPERATIONS}"

if target_type not in VALID_TARGET_TYPES:

return f"Error: target_type must be one of {VALID_TARGET_TYPES}"

logger.info("Patching current %s note: %s at %s=%s", period, operation, target_type, target)

ct = content_type if content_type else (

"application/json" if target_type == "frontmatter" else "text/markdown"

)

extra = build_patch_headers(

operation, target_type, target, target_delimiter,

trim_target_whitespace, create_if_missing,

)

return await make_request("PATCH", f"/periodic/{period}/", content, ct, extra_headers=extra)

@mcp.tool()

async def delete_current_periodic_note(period: str = "daily") -> str:

"""Delete the current periodic note for the specified period."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

logger.info("Deleting current %s note", period)

return await make_request("DELETE", f"/periodic/{period}/")

# ── Periodic Notes Tools (dated) ─────────────────────────────────────────────

@mcp.tool()

async def get_dated_periodic_note(

period: str = "daily",

year: str = "",

month: str = "",

day: str = "",

include_metadata: str = "false",

) -> str:

"""Get a periodic note for a specific date. All of year, month, day are required."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

if not year or not month or not day:

return "Error: year, month, and day are all required"

logger.info("Getting %s note for %s-%s-%s", period, year, month, day)

extra = {}

if include_metadata.lower() == "true":

extra["Accept"] = "application/vnd.olrapi.note+json"

return await make_request(

"GET", f"/periodic/{period}/{year}/{month}/{day}/", extra_headers=extra,

)

@mcp.tool()

async def get_dated_periodic_note_map(

period: str = "daily", year: str = "", month: str = "", day: str = ""

) -> str:

"""Get the document map of a dated periodic note: headings, block references, frontmatter fields."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

if not year or not month or not day:

return "Error: year, month, and day are all required"

logger.info("Getting %s note map for %s-%s-%s", period, year, month, day)

extra = {"Accept": "application/vnd.olrapi.document-map+json"}

return await make_request(

"GET", f"/periodic/{period}/{year}/{month}/{day}/", extra_headers=extra,

)

@mcp.tool()

async def create_dated_periodic_note(

period: str = "daily",

year: str = "",

month: str = "",

day: str = "",

content: str = "",

) -> str:

"""Create or replace a periodic note for a specific date."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

if not year or not month or not day:

return "Error: year, month, and day are all required"

logger.info("Creating %s note for %s-%s-%s", period, year, month, day)

return await make_request(

"PUT", f"/periodic/{period}/{year}/{month}/{day}/", content, "text/markdown",

)

@mcp.tool()

async def append_to_dated_periodic_note(

period: str = "daily",

year: str = "",

month: str = "",

day: str = "",

content: str = "",

) -> str:

"""Append content to a dated periodic note. Creates the note if it does not exist."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

if not year or not month or not day:

return "Error: year, month, and day are all required"

logger.info("Appending to %s note for %s-%s-%s", period, year, month, day)

return await make_request(

"POST", f"/periodic/{period}/{year}/{month}/{day}/", content, "text/markdown",

)

@mcp.tool()

async def patch_dated_periodic_note(

period: str = "daily",

year: str = "",

month: str = "",

day: str = "",

content: str = "",

operation: str = "append",

target_type: str = "heading",

target: str = "",

target_delimiter: str = "::",

trim_target_whitespace: str = "false",

create_if_missing: str = "false",

content_type: str = "",

) -> str:

"""Patch a dated periodic note relative to a heading, block reference, or frontmatter field."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

if not year or not month or not day:

return "Error: year, month, and day are all required"

if not target.strip():

return "Error: target is required"

if operation not in VALID_OPERATIONS:

return f"Error: operation must be one of {VALID_OPERATIONS}"

if target_type not in VALID_TARGET_TYPES:

return f"Error: target_type must be one of {VALID_TARGET_TYPES}"

logger.info(

"Patching %s note for %s-%s-%s: %s at %s=%s",

period, year, month, day, operation, target_type, target,

)

ct = content_type if content_type else (

"application/json" if target_type == "frontmatter" else "text/markdown"

)

extra = build_patch_headers(

operation, target_type, target, target_delimiter,

trim_target_whitespace, create_if_missing,

)

return await make_request(

"PATCH", f"/periodic/{period}/{year}/{month}/{day}/", content, ct, extra_headers=extra,

)

@mcp.tool()

async def delete_dated_periodic_note(

period: str = "daily", year: str = "", month: str = "", day: str = ""

) -> str:

"""Delete a periodic note for a specific date."""

if period not in VALID_PERIODS:

return f"Error: period must be one of {VALID_PERIODS}"

if not year or not month or not day:

return "Error: year, month, and day are all required"

logger.info("Deleting %s note for %s-%s-%s", period, year, month, day)

return await make_request(

"DELETE", f"/periodic/{period}/{year}/{month}/{day}/",

)

# ── Tags Tool ────────────────────────────────────────────────────────────────

@mcp.tool()

async def get_tags() -> str:

"""List all tags in the vault with usage counts. Tags are returned without the '#' prefix."""

logger.info("Getting tags")

return await make_request("GET", "/tags/")

# ── Open / Navigation Tool ──────────────────────────────────────────────────

@mcp.tool()

async def open_note(filepath: str = "", new_leaf: str = "false") -> str:

"""Open a note in the Obsidian UI. Set new_leaf to 'true' to open in a new pane. Creates the file if it does not exist."""

if not filepath.strip():

return "Error: filepath is required"

cleaned = clean_path(filepath)

logger.info("Opening note: %s", cleaned)

params = {}

if new_leaf.lower() == "true":

params["newLeaf"] = "true"

return await make_request("POST", f"/open/{cleaned}", params=params)

# ── Server Startup ───────────────────────────────────────────────────────────

if __name__ == "__main__":

logger.info("Starting Obsidian Vault MCP server...")

if not OBSIDIAN_API_KEY:

logger.warning("OBSIDIAN_API_KEY not set - authentication will fail")

logger.info("Configured API URL: %s", OBSIDIAN_API_URL)

try:

mcp.run(transport="stdio")

except Exception as exc:

logger.error("Server error: %s", exc, exc_info=True)

sys.exit(1)