Source code for ralph.config.mcp_loader

"""Three-layer mcp.toml loader.

Merge order (lowest → highest priority):
  1. Bundled default  - ralph/policy/defaults/mcp.toml   (ships in wheel)
  2. User-global      - $XDG_CONFIG_HOME/ralph-workflow-mcp.toml
                        (default: ~/.config/ralph-workflow-mcp.toml)
  3. Project-local    - .agent/mcp.toml  (resolved via WorkspaceScope)

Unlike the main config loader, TOML parse errors here are fail-fast: any
malformed file triggers a typed exit rather than a silent empty-dict fallback.
"""

from __future__ import annotations

import tomllib
from os import getenv
from pathlib import Path
from typing import TYPE_CHECKING, cast

from loguru import logger
from pydantic import ValidationError

from ralph.config.loader import deep_merge as _deep_merge
from ralph.config.mcp_models import McpConfig

if TYPE_CHECKING:
    from ralph.workspace.scope import WorkspaceScope

_GLOBAL_MCP_FILENAME = "ralph-workflow-mcp.toml"
_LOCAL_MCP_FILENAME = "mcp.toml"
_TOML_DECODE_ERROR = cast("type[ValueError]", tomllib.TOMLDecodeError)




class McpConfigError(SystemExit):
    """Typed fail-fast exit for invalid MCP configuration."""

    def __init__(self, message: str, *, code: int = 1) -> None:
        super().__init__(code)
        self.message = message
        self.code = code

    def __str__(self) -> str:
        return self.message





def bundled_default_mcp_config_path() -> Path:
    """Return the path to the bundled default MCP configuration file."""
    return Path(__file__).parent.parent / "policy" / "defaults" / _LOCAL_MCP_FILENAME





def global_mcp_config_path() -> Path:
    """Return the user-level global MCP config path, respecting XDG_CONFIG_HOME."""
    xdg = getenv("XDG_CONFIG_HOME")
    if xdg:
        return Path(xdg) / _GLOBAL_MCP_FILENAME
    return Path.home() / ".config" / _GLOBAL_MCP_FILENAME





def local_mcp_config_path(workspace_scope: WorkspaceScope) -> Path:
    """Return the workspace-local MCP config path for the given workspace scope."""
    if hasattr(workspace_scope, "resolve_agent_file"):
        return workspace_scope.resolve_agent_file(_LOCAL_MCP_FILENAME)
    return workspace_scope.local_config_path.parent / _LOCAL_MCP_FILENAME



def _load_mcp_toml(path: Path) -> dict[str, object]:
    if not path.exists():
        logger.debug("MCP config not found, skipping: {}", path)
        return {}
    logger.debug("Loading MCP config from {}", path)
    with path.open("rb") as fh:
        try:
            data: dict[str, object] = tomllib.load(fh)
        except _TOML_DECODE_ERROR as exc:
            message = f"MCP config parse error at {path}: {exc}"
            logger.error(message)
            raise McpConfigError(message) from exc
    return data


def _validate_fallback_backends(config: McpConfig) -> None:
    for entry in config.web_search.fallback:
        if entry != "ddgs" and entry not in config.web_search.backends:
            message = (
                f"MCP config: fallback backend '{entry}' is not configured in"
                f" [web_search.backends]; add a [web_search.backends.{entry}] section"
                " or remove it from the fallback list"
            )
            logger.error(message)
            raise McpConfigError(message)


def _inject_mcp_server_names(merged: dict[str, object]) -> None:
    raw_servers = merged.get("mcp_servers")
    if not isinstance(raw_servers, dict):
        return
    for server_name, server_spec in raw_servers.items():
        if isinstance(server_spec, dict) and "name" not in server_spec:
            server_spec["name"] = server_name




def load_mcp_config(
    workspace_scope: WorkspaceScope | None = None,
    config_path: Path | None = None,
) -> McpConfig:
    """Build merged McpConfig from all layers.

    Args:
        workspace_scope: Provides the project-local .agent/ root. Not used when
            config_path is given.
        config_path: Explicit override for the project-local layer.

    Returns:
        Validated McpConfig.

    Raises:
        McpConfigError: On TOML parse error, schema validation failure, or
            unknown fallback backend reference.
    """
    bundled = _load_mcp_toml(bundled_default_mcp_config_path())
    global_data = _load_mcp_toml(global_mcp_config_path())

    if config_path is not None:
        local_data = _load_mcp_toml(config_path)
    elif workspace_scope is not None:
        local_data = _load_mcp_toml(local_mcp_config_path(workspace_scope))
    else:
        local_data = {}

    merged = _deep_merge(bundled, global_data)
    merged = _deep_merge(merged, local_data)
    _inject_mcp_server_names(merged)

    try:
        config = McpConfig.model_validate(merged)
    except ValidationError as exc:
        message = f"MCP config validation failed:\n{exc}"
        logger.error(message)
        raise McpConfigError(message) from exc

    logger.debug(
        "MCP config loaded: {} server(s), web_search.enabled={}",
        len(config.mcp_servers),
        config.web_search.enabled,
    )
    _validate_fallback_backends(config)
    return config