ajarbot/obsidian_mcp.py

"""Obsidian MCP Server Integration.

Manages the external obsidian-mcp-server process and provides
health checking, fallback routing, and configuration loading.
"""

import os
import time
import threading
from pathlib import Path
from typing import Any, Dict, Optional

import yaml
import httpx

# Default config path
_CONFIG_FILE = Path("config/obsidian_mcp.yaml")

# Cached state
_obsidian_healthy: bool = False
_last_health_check: float = 0.0
_health_lock = threading.Lock()
_config_cache: Optional[Dict] = None


def load_obsidian_config() -> Dict[str, Any]:
    """Load Obsidian MCP configuration with env var overrides."""
    global _config_cache

    if _config_cache is not None:
        return _config_cache

    config = {}
    if _CONFIG_FILE.exists():
        with open(_CONFIG_FILE, encoding="utf-8") as f:
            config = yaml.safe_load(f) or {}

    obsidian = config.get("obsidian_mcp", {})

    # Apply environment variable overrides
    env_overrides = {
        "OBSIDIAN_API_KEY": ("connection", "api_key"),
        "OBSIDIAN_BASE_URL": ("connection", "base_url"),
        "OBSIDIAN_MCP_ENABLED": None,  # Special: top-level "enabled"
        "OBSIDIAN_ROUTING_STRATEGY": ("routing", "strategy"),
        "OBSIDIAN_VAULT_PATH": ("vault", "path"),
    }

    for env_var, path in env_overrides.items():
        value = os.getenv(env_var)
        if value is None:
            continue

        if path is None:
            # Top-level key
            obsidian["enabled"] = value.lower() in ("true", "1", "yes")
        else:
            section, key = path
            obsidian.setdefault(section, {})[key] = value

    _config_cache = obsidian
    return obsidian


def is_obsidian_enabled() -> bool:
    """Check if Obsidian MCP integration is enabled in config."""
    config = load_obsidian_config()
    return config.get("enabled", False)


def check_obsidian_health(force: bool = False) -> bool:
    """Check if Obsidian REST API is reachable.

    Uses cached result unless force=True or cache has expired.
    Thread-safe.
    """
    global _obsidian_healthy, _last_health_check

    config = load_obsidian_config()
    check_interval = config.get("routing", {}).get("health_check_interval", 60)
    timeout = config.get("routing", {}).get("api_timeout", 10)

    with _health_lock:
        now = time.time()
        if not force and (now - _last_health_check) < check_interval:
            return _obsidian_healthy

        base_url = config.get("connection", {}).get(
            "base_url", "http://127.0.0.1:27123"
        )
        api_key = config.get("connection", {}).get("api_key", "")

        try:
            # Obsidian Local REST API health endpoint
            response = httpx.get(
                f"{base_url}/",
                headers={"Authorization": f"Bearer {api_key}"},
                timeout=timeout,
                verify=config.get("connection", {}).get("verify_ssl", False),
            )
            _obsidian_healthy = response.status_code == 200
        except Exception:
            _obsidian_healthy = False

        _last_health_check = now
        return _obsidian_healthy


def get_obsidian_server_config() -> Dict[str, Any]:
    """Build the MCP server configuration for Agent SDK registration.

    Returns the config dict suitable for ClaudeAgentOptions.mcp_servers.
    The obsidian-mcp-server runs as a stdio subprocess.
    """
    config = load_obsidian_config()
    connection = config.get("connection", {})
    vault = config.get("vault", {})
    cache = config.get("cache", {})
    logging = config.get("logging", {})

    env = {
        "OBSIDIAN_API_KEY": connection.get("api_key", ""),
        "OBSIDIAN_BASE_URL": connection.get(
            "base_url", "http://127.0.0.1:27123"
        ),
        "OBSIDIAN_VERIFY_SSL": str(
            connection.get("verify_ssl", False)
        ).lower(),
        "OBSIDIAN_VAULT_PATH": str(vault.get("path", "")),
        "OBSIDIAN_ENABLE_CACHE": str(
            cache.get("enabled", True)
        ).lower(),
        "OBSIDIAN_CACHE_REFRESH_INTERVAL_MIN": str(
            cache.get("refresh_interval_min", 10)
        ),
        "MCP_LOG_LEVEL": logging.get("level", "info"),
    }

    return {
        "command": "npx",
        "args": ["obsidian-mcp-server"],
        "env": env,
    }


def get_routing_strategy() -> str:
    """Get the configured tool routing strategy."""
    config = load_obsidian_config()
    return config.get("routing", {}).get("strategy", "obsidian_preferred")


def should_fallback_to_custom() -> bool:
    """Check if fallback to custom tools is enabled."""
    config = load_obsidian_config()
    return config.get("routing", {}).get("fallback_to_custom", True)


# List of all Obsidian MCP tool names
OBSIDIAN_TOOLS = [
    "obsidian_read_note",
    "obsidian_update_note",
    "obsidian_search_replace",
    "obsidian_global_search",
    "obsidian_list_notes",
    "obsidian_manage_frontmatter",
    "obsidian_manage_tags",
    "obsidian_delete_note",
]