"""MCP (Model Context Protocol) tools loader.

This module provides async functions to load and manage MCP servers using
`langchain-mcp-adapters`, supporting Claude Desktop style JSON configs.
It also supports automatic discovery of `.mcp.json` files from user-level
and project-level locations.
"""

from __future__ import annotations

import asyncio
import copy
import fnmatch
import json
import logging
import re
import shutil
from contextlib import AsyncExitStack
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast, overload

if TYPE_CHECKING:
    from collections.abc import Mapping, Sequence

    from langchain_core.tools import BaseTool
    from langchain_mcp_adapters.client import Connection
    from mcp import ClientSession

    from deepagents_code.project_utils import ProjectContext

logger = logging.getLogger(__name__)

# Maintainer note: `deepagents-talon` imports `MCPConfigError`,
# `MCPServerInfo`, and `get_mcp_tools` from this module, and its tests construct
# `MCPToolInfo`. Keep those symbols' names, signatures, and return/dataclass
# shapes stable unless `deepagents-talon` is migrated in the same change.


@dataclass(frozen=True, slots=True)
class MCPToolInfo:
    """Metadata for a single MCP tool."""

    name: str
    """Tool name (may include server name prefix)."""

    description: str
    """Human-readable description of what the tool does."""

    input_schema: dict[str, Any] | None = None
    """Raw MCP `inputSchema` dict (JSON Schema), or `None` when unavailable.

    Supplied directly from `mcp_tool.inputSchema` at tool-load time. The viewer
    reads `properties` and `required` from this dict for parameter display;
    `None` is rendered as "no parameters".
    """


MCPServerStatus = Literal[
    "ok",
    "unauthenticated",
    "awaiting_reconnect",
    "error",
    "disabled",
]
"""Load states a configured MCP server can end up in.

`ok` means the server loaded successfully and has an authoritative tool list.

`unauthenticated` means the server requires OAuth login before tools can load.

`error` means the server failed to load after a connection or configuration
failure.

`disabled` is set when the user has turned the server off via the TUI
(`/mcp` -> F2). No connection is attempted and no tools are loaded, but
the entry is still surfaced in the viewer so the user can re-enable it.

`awaiting_reconnect` is a transient UI-only state used after OAuth login
has succeeded but before the LangGraph server has restarted and loaded
the newly available MCP tools.
"""


@dataclass(frozen=True, slots=True)
class MCPServerInfo:
    """Metadata for a configured MCP server and its tools."""

    name: str
    """Server name from the MCP configuration."""

    transport: str
    """Transport identifier — `stdio`, `sse`, `http`, the synthetic
    `config` value used for entries surfacing a bad config file, or
    `unknown` for a disabled server whose original config could not be
    classified."""

    tools: tuple[MCPToolInfo, ...] = ()
    """Tools exposed by this server (empty when `status != "ok"`)."""

    status: MCPServerStatus = "ok"
    """Load status.

    One of `ok`, `unauthenticated`, `awaiting_reconnect`, `error`, or
    `disabled`.
    """

    error: str | None = None
    """Human-readable reason when `status != "ok"`."""

    def __post_init__(self) -> None:
        """Enforce the status/error/tools consistency invariant.

        Raises:
            ValueError: If any of: `status='ok'` with a non-`None` error;
                non-`ok` status without an error message; non-`ok` status
                carrying tools.
        """
        if self.status == "ok":
            if self.error is not None:
                msg = (
                    f"MCPServerInfo {self.name!r}: status='ok' cannot carry "
                    f"an error (got {self.error!r})"
                )
                raise ValueError(msg)
        else:
            if self.error is None:
                msg = (
                    f"MCPServerInfo {self.name!r}: status={self.status!r} "
                    "requires an error message"
                )
                raise ValueError(msg)
            if self.tools:
                msg = (
                    f"MCPServerInfo {self.name!r}: status={self.status!r} "
                    "cannot carry tools"
                )
                raise ValueError(msg)

    def needs_attention(self) -> bool:
        """Return whether this server is blocked on user login."""
        return self.status == "unauthenticated"


_SUPPORTED_REMOTE_TYPES = {"sse", "http"}
"""Supported transport types for remote MCP servers (SSE and HTTP)."""

_TRANSPORT_ALIASES = {"streamable_http": "http", "streamable-http": "http"}
"""Aliases that normalize to canonical transport names.

The MCP spec and `langchain_mcp_adapters` use `streamable_http` for what the
app calls `http`. Accept both so users copy-pasting from upstream docs don't
hit a validation error.
"""


_SERVER_NAME_RE = re.compile(r"^[A-Za-z0-9_-]+$")
"""Server names become token-file basenames and must remain path-safe."""


class MCPConfigError(ValueError):
    """An MCP configuration file is malformed or structurally invalid.

    Subclasses `ValueError` so existing `except ValueError` handlers
    keep working; new code can catch this specifically to render a
    user-actionable message (typically with a file path and hint).
    """


def _is_transient_session_error(exc: BaseException) -> bool:
    """Return `True` when `exc` signals the MCP session transport is dead.

    The anyio import is guarded so an anyio rename or removal surfaces as
    an `ImportError` at module import rather than silent mis-classification
    at runtime. Standard-library socket/pipe/EOF errors are covered as a
    fallback regardless of anyio's presence.
    """
    try:
        import anyio
    except ImportError:  # pragma: no cover - anyio is a transitive MCP dep
        anyio_excs: tuple[type[BaseException], ...] = ()
    else:
        anyio_excs = (
            anyio.ClosedResourceError,
            anyio.BrokenResourceError,
            anyio.EndOfStream,
        )
    return isinstance(
        exc,
        (
            *anyio_excs,
            BrokenPipeError,
            ConnectionAbortedError,
            ConnectionResetError,
            EOFError,
            asyncio.IncompleteReadError,
        ),
    )


@dataclass(frozen=True, slots=True)
class _MCPSessionEntry:
    """Cached MCP session and its close stack."""

    session: ClientSession
    exit_stack: AsyncExitStack


def _connection_signature(value: Any) -> Any:  # noqa: ANN401
    """Return a stable comparison signature for MCP connection configs."""
    from mcp.client.auth import OAuthClientProvider

    if isinstance(value, dict):
        return tuple(
            sorted((key, _connection_signature(item)) for key, item in value.items()),
        )
    if isinstance(value, list | tuple):
        return tuple(_connection_signature(item) for item in value)
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, OAuthClientProvider):
        context = value.context
        storage_path = getattr(getattr(context, "storage", None), "path", None)
        return (
            "oauth",
            _connection_signature(context.server_url),
            _connection_signature(
                context.client_metadata.model_dump(mode="json", exclude_none=True),
            ),
            _connection_signature(storage_path),
            _connection_signature(context.timeout),
            _connection_signature(context.client_metadata_url),
            _connection_signature(context.auth_server_url),
            _connection_signature(context.protocol_version),
        )

    model_dump = getattr(value, "model_dump", None)
    if callable(model_dump):
        return _connection_signature(model_dump(mode="json", exclude_none=True))
    return value


def _connections_signature(
    connections: dict[str, Connection],
) -> tuple[tuple[str, Any], ...]:
    """Return a stable signature for a full MCP connections mapping."""
    return tuple(
        sorted(
            (name, _connection_signature(connection))
            for name, connection in connections.items()
        ),
    )


class MCPSessionManager:
    """Lazy, per-server cache of persistent MCP sessions.

    Discovery always happens through throwaway sessions. Live sessions are
    only created on the first real tool call inside the runtime event loop
    so sessions stay bound to the loop that owns their subprocess/transport
    handles, and so stdio servers are not restarted on every invocation.
    """

    def __init__(self, *, connections: dict[str, Connection] | None = None) -> None:
        """Initialize the session manager.

        Args:
            connections: Optional initial server connection configs.
        """
        self._connections: dict[str, Connection] = dict(connections or {})
        self._entries: dict[str, _MCPSessionEntry] = {}
        self._locks: dict[str, asyncio.Lock] = {}
        self._closed = False

    def configure(self, connections: dict[str, Connection]) -> None:
        """Set or validate the connection configs used by this manager.

        When no sessions exist yet, `connections` overwrites the stored
        configs unconditionally. Once any session has been created, the
        new `connections` must produce the same signature as the stored
        ones — otherwise this raises to prevent rebinding live sessions
        to different transports or auth providers.

        Args:
            connections: Connection configs keyed by server name.

        Raises:
            RuntimeError: If the manager is closed or reconfigured
                incompatibly after sessions already exist.
        """
        if self._closed:
            msg = "Cannot configure a closed MCP session manager"
            raise RuntimeError(msg)

        if not self._entries:
            self._connections = dict(connections)
            return

        if _connections_signature(self._connections) != _connections_signature(
            connections,
        ):
            msg = "Cannot reconfigure MCP session manager after sessions are active"
            raise RuntimeError(msg)
        self._connections = dict(connections)

    async def get_session(self, server_name: str) -> ClientSession:
        """Return a cached session for `server_name`, creating it lazily."""
        entry = self._entries.get(server_name)
        if entry is not None:
            return entry.session

        lock = self._get_lock(server_name)
        async with lock:
            entry = self._entries.get(server_name)
            if entry is not None:
                return entry.session

            entry = await self._create_entry(server_name)
            self._entries[server_name] = entry
            return entry.session

    async def invalidate(
        self,
        server_name: str,
        *,
        expected_session: ClientSession | None = None,
    ) -> None:
        """Evict and close a cached session if it still matches `expected_session`.

        Args:
            server_name: MCP server name.
            expected_session: Optional identity check for race-safe eviction.
        """
        lock = self._get_lock(server_name)
        async with lock:
            entry = self._entries.get(server_name)
            if entry is None:
                return
            if expected_session is not None and entry.session is not expected_session:
                return
            self._entries.pop(server_name, None)
            exit_stack = entry.exit_stack

        await exit_stack.aclose()

    async def cleanup(self) -> None:
        """Close all cached sessions concurrently and reject future creation.

        Each server's `exit_stack.aclose()` runs with a 5 second timeout so
        one slow stdio server cannot stall shutdown. Per-server failures
        are logged — teardown is best-effort — but `CancelledError` is
        re-raised so the enclosing `asyncio.gather` still cancels peers.
        """
        if self._closed and not self._entries:
            return

        self._closed = True
        names = list(self._entries)

        async def _close(server_name: str) -> None:
            try:
                await asyncio.wait_for(self.invalidate(server_name), timeout=5.0)
            except TimeoutError:
                logger.warning(
                    "MCP session cleanup for %r timed out after 5s",
                    server_name,
                )
            except (KeyboardInterrupt, SystemExit, asyncio.CancelledError):
                raise
            except Exception:
                logger.warning(
                    "MCP session cleanup for %r failed",
                    server_name,
                    exc_info=True,
                )

        await asyncio.gather(*[_close(name) for name in names])

    def _get_lock(self, server_name: str) -> asyncio.Lock:
        """Return the per-server creation/eviction lock."""
        lock = self._locks.get(server_name)
        if lock is None:
            lock = asyncio.Lock()
            self._locks[server_name] = lock
        return lock

    async def _create_entry(self, server_name: str) -> _MCPSessionEntry:
        """Create and initialize a new cached session entry.

        Args:
            server_name: MCP server name.

        Returns:
            A cached session entry containing the live session and close stack.

        Raises:
            RuntimeError: If the manager has already been cleaned up.
            ValueError: If `server_name` is not configured in the manager.
        """
        if self._closed:
            msg = "Cannot create an MCP session after cleanup"
            raise RuntimeError(msg)

        try:
            connection = self._connections[server_name]
        except KeyError as exc:
            msg = (
                f"Couldn't find an MCP server named '{server_name}', "
                f"expected one of {sorted(self._connections)}"
            )
            raise ValueError(msg) from exc

        from langchain_mcp_adapters.sessions import create_session

        exit_stack = AsyncExitStack()
        try:
            session = await exit_stack.enter_async_context(create_session(connection))
            await session.initialize()
        except Exception:
            await exit_stack.aclose()
            raise

        return _MCPSessionEntry(session=session, exit_stack=exit_stack)


def _resolve_server_type(server_config: Mapping[str, Any]) -> str:
    """Determine the transport type for a server config.

    Accepts `type` or `transport` interchangeably. When neither is set, a
    `url` field implies a remote server (defaulting to `http`) and the
    absence of `url` implies stdio. This matches Claude Code's `.mcp.json`
    convention where remote entries are commonly written as `{"url": "..."}`
    alone.

    Args:
        server_config: Server configuration dictionary.

    Returns:
        Transport type string (`stdio`, `sse`, or `http`).
    """
    transport = server_config.get("type") or server_config.get("transport")
    if transport is not None:
        return _TRANSPORT_ALIASES.get(transport, transport)
    if "url" in server_config:
        return "http"
    return "stdio"


def _validate_server_config(server_name: str, server_config: dict[str, Any]) -> None:
    """Validate a single server configuration.

    Performs only shape checks — `${VAR}` header interpolation is deferred
    to activation time so one unset env var only fails its own server
    rather than hiding every other MCP entry in the same file.

    Args:
        server_name: Name of the server.
        server_config: Server configuration dictionary.

    Raises:
        TypeError: If config fields have wrong types.
        ValueError: If required fields are missing or server type is unsupported.
    """
    if not _SERVER_NAME_RE.fullmatch(server_name):
        error_msg = (
            f"Invalid server name {server_name!r}: server names must contain "
            "only alphanumerics, hyphens, and underscores."
        )
        raise ValueError(error_msg)

    if not isinstance(server_config, dict):
        error_msg = f"Server '{server_name}' config must be a dictionary"
        raise TypeError(error_msg)

    server_type = _resolve_server_type(server_config)

    if server_type in _SUPPORTED_REMOTE_TYPES:
        if "url" not in server_config:
            error_msg = (
                f"Server '{server_name}' with type '{server_type}' "
                "missing required 'url' field"
            )
            raise ValueError(error_msg)

        if "command" in server_config:
            error_msg = (
                f"Server '{server_name}' has type '{server_type}' (remote) "
                "but also declares a 'command' field. Remove 'command' or "
                'set `"type": "stdio"`.'
            )
            raise ValueError(error_msg)

        headers = server_config.get("headers")
        if headers is not None and not isinstance(headers, dict):
            error_msg = f"Server '{server_name}' 'headers' must be a dictionary"
            raise TypeError(error_msg)

        if isinstance(headers, dict):
            for name, value in headers.items():
                if not isinstance(value, str):
                    error_msg = (
                        f"Server '{server_name}' header {name!r} must be "
                        f"a string, got {type(value).__name__}"
                    )
                    raise TypeError(error_msg)
    elif server_type == "stdio":
        if "command" not in server_config:
            error_msg = f"Server '{server_name}' missing required 'command' field"
            raise ValueError(error_msg)

        if "url" in server_config:
            error_msg = (
                f"Server '{server_name}' has type 'stdio' but also declares "
                "a 'url' field. Remove 'url' or set "
                '`"type": "http"` (or `"sse"`) for a remote server.'
            )
            raise ValueError(error_msg)

        if "args" in server_config and not isinstance(server_config["args"], list):
            error_msg = f"Server '{server_name}' 'args' must be a list"
            raise TypeError(error_msg)

        if "env" in server_config and not isinstance(server_config["env"], dict):
            error_msg = f"Server '{server_name}' 'env' must be a dictionary"
            raise TypeError(error_msg)
    else:
        error_msg = (
            f"Server '{server_name}' has unsupported transport type '{server_type}'. "
            "Supported types: stdio, sse, http"
        )
        raise ValueError(error_msg)

    auth = server_config.get("auth")
    if auth is not None:
        if auth != "oauth":
            msg = (
                f"Server '{server_name}' has unsupported auth value "
                f"{auth!r}. Only 'oauth' is supported."
            )
            raise ValueError(msg)
        if server_type == "stdio":
            msg = (
                f"Server '{server_name}' uses stdio transport; "
                "'auth: oauth' is only valid for http/sse transports."
            )
            raise ValueError(msg)
        header_names = {name.lower() for name in (server_config.get("headers") or {})}
        if "authorization" in header_names:
            msg = (
                f"Server '{server_name}' cannot combine 'auth: oauth' "
                "with an 'Authorization' header."
            )
            raise ValueError(msg)

    _validate_tool_filter_fields(server_name, server_config)


def _validate_tool_filter_fields(
    server_name: str,
    server_config: dict[str, Any],
) -> None:
    """Validate optional `allowedTools` / `disabledTools` fields.

    Both fields, when present, must be non-empty lists of strings. Setting
    both on the same server is rejected to keep the filter semantics
    unambiguous. An empty list is rejected because it would silently strip
    every tool from the server (`allowedTools`) or be a no-op
    (`disabledTools`) — both are almost certainly user errors; omit the field
    instead.

    Args:
        server_name: Name of the server (for error messages).
        server_config: Server configuration dictionary.

    Raises:
        TypeError: If a field is not a list of strings.
        ValueError: If both fields are set, or either field is empty.
    """
    has_allowed = "allowedTools" in server_config
    has_disabled = "disabledTools" in server_config
    if has_allowed and has_disabled:
        error_msg = (
            f"Server '{server_name}' cannot set both 'allowedTools' and"
            " 'disabledTools' — pick one."
        )
        raise ValueError(error_msg)

    for field_name in ("allowedTools", "disabledTools"):
        if field_name not in server_config:
            continue
        value = server_config[field_name]
        if not isinstance(value, list) or not all(
            isinstance(item, str) for item in value
        ):
            error_msg = (
                f"Server '{server_name}' '{field_name}' must be a list of strings"
            )
            raise TypeError(error_msg)
        if not value:
            error_msg = (
                f"Server '{server_name}' '{field_name}' must be non-empty;"
                " omit the field to disable filtering."
            )
            raise ValueError(error_msg)


def _looks_like_comment(doc: str, lineno: int) -> bool:
    """Return `True` if the offending line *begins* with `//` or `/*`.

    Only the failing line is checked, and only its leading characters (after
    stripping indentation). A `url` value such as `"url": "https://..."`
    begins with a quote, not `//`, so a URL scheme inside a quoted string
    never triggers a false comment hint.

    Args:
        doc: Full source text that failed to parse.
        lineno: 1-based line number of the error; out-of-range values
            return `False`.

    Returns:
        `True` when the stripped failing line starts with `//` or `/*`.
    """
    lines = doc.splitlines()
    if lineno < 1 or lineno > len(lines):
        return False
    stripped = lines[lineno - 1].lstrip()
    return stripped.startswith(("//", "/*"))


def _json_error_hint(exc: json.JSONDecodeError) -> str | None:
    """Return an actionable hint for a common JSON mistake, or `None`.

    Checks are ordered most-specific-first (trailing comma, then comment,
    then generic decoder-message keywords) so a more precise hint wins when
    several could apply.

    Args:
        exc: The decode error to classify.

    Returns:
        A hint string for a recognized mistake, or `None` when no specific
            guidance applies.
    """
    msg = exc.msg.lower()
    if "trailing comma" in msg:
        return (
            "Hint: JSON does not allow trailing commas. Remove the comma "
            "before the closing '}' or ']'."
        )
    if _looks_like_comment(exc.doc, exc.lineno):
        return "Hint: JSON does not allow comments (// or /* */). Remove them."
    if "expecting property name" in msg:
        return (
            "Hint: check for trailing commas, a missing key, or an unquoted "
            "property name near this position."
        )
    if "expecting value" in msg:
        return (
            "Hint: check for a missing value, an extra comma, or unquoted text "
            "near this position."
        )
    if "delimiter" in msg:
        return (
            "Hint: check for a missing comma, ':', or closing bracket near "
            "this position."
        )
    return None


def _trailing_comma_pos(doc: str, pos: int) -> int | None:
    """Return the comma position for decoder errors at a trailing comma."""
    if pos < 0 or pos >= len(doc) or doc[pos] not in "}]":
        return None
    idx = pos - 1
    while idx >= 0 and doc[idx].isspace():
        idx -= 1
    if idx >= 0 and doc[idx] == ",":
        return idx
    return None


def _json_error_snippet(
    doc: str, lineno: int, colno: int, *, pos: int | None = None
) -> str | None:
    """Build a caret snippet pointing at a JSON error location.

    Args:
        doc: Full source text that failed to parse.
        lineno: 1-based line number of the error.
        colno: 1-based column number of the error.
        pos: 0-based absolute error offset, if available.

    Returns:
        A two-line `<source line>` + caret string, or `None` when the line
        is out of range or blank.
    """
    if pos is not None:
        trailing_pos = _trailing_comma_pos(doc, pos)
        if trailing_pos is not None:
            lineno = doc.count("\n", 0, trailing_pos) + 1
            line_start = doc.rfind("\n", 0, trailing_pos) + 1
            colno = trailing_pos - line_start + 1
    lines = doc.splitlines()
    if lineno < 1 or lineno > len(lines):
        return None
    source = lines[lineno - 1].rstrip()
    if not source:
        return None
    caret_col = max(0, min(colno - 1, len(source)))
    return f"    {source}\n    {' ' * caret_col}^"


def load_mcp_config(config_path: str) -> dict[str, Any]:
    """Load and validate MCP configuration from a JSON file.

    Supports multiple server types:

    - stdio: Process-based servers with `command`, `args`, `env` fields (default)
    - sse: Server-Sent Events servers with `type: "sse"`, `url`, and optional `headers`
    - http: HTTP-based servers with `type: "http"`, `url`, and optional `headers`

    Any server type may also set an optional tool filter:

    - `allowedTools`: list of tool names or patterns to keep (all others dropped)
    - `disabledTools`: list of tool names or patterns to drop (all others kept)

    Entries are either literal tool names or `fnmatch`-style glob patterns
    (entries containing `*`, `?`, or `[`). Each entry is matched against both
    the bare MCP tool name and the server-prefixed form
    (`f"{server_name}_{tool}"`), so either `read_*` or `fs_read_*` works.
    Setting both fields on a single server is an error.

    Args:
        config_path: Path to the MCP JSON configuration file.

    Returns:
        Parsed configuration dictionary.

    Raises:
        FileNotFoundError: If config file doesn't exist.
        json.JSONDecodeError: If config file contains invalid JSON.
        TypeError: If config fields have wrong types.
        ValueError: If config is missing required fields.
        RuntimeError: If header env-var interpolation references an unset var.
    """  # noqa: DOC502 - `_validate_server_config()` raises `RuntimeError` indirectly
    path = Path(config_path)

    if not path.exists():
        error_msg = f"MCP config file not found: {config_path}"
        raise FileNotFoundError(error_msg)

    try:
        with path.open(encoding="utf-8") as file_obj:
            config = json.load(file_obj)
    except json.JSONDecodeError as exc:
        # Build a layered message: core reason, an actionable hint for common
        # mistakes, then a caret snippet last so the auto-appended
        # "line X column Y" suffix reads as the location of the caret.
        parts = [f"Invalid JSON in MCP config file: {exc.msg}"]
        hint = _json_error_hint(exc)
        if hint is not None:
            parts.append(hint)
        snippet = _json_error_snippet(exc.doc, exc.lineno, exc.colno, pos=exc.pos)
        if snippet is not None:
            parts.append(snippet)
        error_msg = "\n".join(parts)
        raise json.JSONDecodeError(error_msg, exc.doc, exc.pos) from exc

    if "mcpServers" not in config:
        error_msg = (
            "MCP config must contain 'mcpServers' field. "
            'Expected format: {"mcpServers": {"server-name": {...}}}'
        )
        raise ValueError(error_msg)

    if not isinstance(config["mcpServers"], dict):
        error_msg = "'mcpServers' field must be a dictionary"
        raise TypeError(error_msg)

    if not config["mcpServers"]:
        error_msg = "'mcpServers' field is empty - no servers configured"
        raise ValueError(error_msg)

    for server_name, server_config in config["mcpServers"].items():
        _validate_server_config(server_name, server_config)

    return config


def _resolve_project_config_base(project_context: ProjectContext | None) -> Path:
    """Resolve the base directory for project-level MCP configuration lookup.

    Args:
        project_context: Explicit project path context, if available.

    Returns:
        Project root when one exists, otherwise the user working directory.
    """
    if project_context is not None:
        return project_context.project_root or project_context.user_cwd

    from deepagents_code.project_utils import find_project_root

    return find_project_root() or Path.cwd()


MCP_CONFIG_DISCOVERY_PATHS: tuple[tuple[str, str], ...] = (
    ("~/.deepagents/.mcp.json", "user-level"),
    ("<project-root>/.deepagents/.mcp.json", "project subdir"),
    ("<project-root>/.mcp.json", "project root"),
)
"""Display strings for the auto-discovered MCP config paths.

Ordered from lowest to highest precedence. Each entry is `(path, label)`
suitable for rendering in help screens and error messages. The runtime
discovery in `discover_mcp_configs` builds the same paths from
`Path.home()` and `_resolve_project_config_base()`.
"""


def discover_mcp_configs(
    *,
    project_context: ProjectContext | None = None,
) -> list[Path]:
    """Find MCP config files from standard locations.

    Checks the paths listed in `MCP_CONFIG_DISCOVERY_PATHS`, lowest to
    highest precedence.

    Args:
        project_context: Explicit project path context, if available.

    Returns:
        Existing config file paths, ordered from lowest to highest precedence.
    """
    user_dir = Path.home() / ".deepagents"
    project_root = _resolve_project_config_base(project_context)

    candidates = [
        user_dir / ".mcp.json",
        project_root / ".deepagents" / ".mcp.json",
        project_root / ".mcp.json",
    ]

    found: list[Path] = []
    for path in candidates:
        try:
            if path.is_file():
                found.append(path)
        except OSError:
            logger.warning("Could not check MCP config %s", path, exc_info=True)
    return found


def classify_discovered_configs(
    config_paths: list[Path],
) -> tuple[list[Path], list[Path]]:
    """Split discovered config paths into user-level and project-level configs.

    Args:
        config_paths: Candidate config paths from discovery.

    Returns:
        Tuple of `(user_configs, project_configs)`.
    """
    user_dir = Path.home() / ".deepagents"
    user: list[Path] = []
    project: list[Path] = []
    for path in config_paths:
        try:
            if path.resolve().is_relative_to(user_dir.resolve()):
                user.append(path)
            else:
                project.append(path)
        except (OSError, ValueError):
            project.append(path)
    return user, project


def extract_stdio_server_commands(
    config: dict[str, Any],
) -> list[tuple[str, str, list[str]]]:
    """Extract stdio server entries from a parsed MCP config.

    Args:
        config: Parsed MCP config dictionary.

    Returns:
        List of `(server_name, command, args)` tuples for stdio servers.
    """
    results: list[tuple[str, str, list[str]]] = []
    servers = config.get("mcpServers", {})
    if not isinstance(servers, dict):
        return results
    for name, server in servers.items():
        if not isinstance(server, dict):
            continue
        if _resolve_server_type(server) == "stdio":
            results.append((name, server.get("command", ""), server.get("args", [])))
    return results


def extract_project_server_summaries(
    config: dict[str, Any],
) -> list[tuple[str, str, str]]:
    """Return `(name, kind, summary)` for every server in a project config.

    Used by the trust prompt and the untrusted-config skip warning so that
    both stdio servers (which spawn local commands) and remote servers
    (which can SSRF or exfiltrate environment variables via interpolated
    headers when an attacker controls `.mcp.json`) are gated identically.

    Args:
        config: Parsed MCP config dictionary.

    Returns:
        List of `(server_name, kind, summary)`. `kind` is `"stdio"`,
            `"http"`, `"sse"`, or `"unknown"`. `summary` is `"<command> <args>"`
            for stdio entries and the URL for remote entries.
    """
    results: list[tuple[str, str, str]] = []
    servers = config.get("mcpServers", {})
    if not isinstance(servers, dict):
        return results
    for name, server in servers.items():
        if not isinstance(server, dict):
            continue
        kind = _resolve_server_type(server)
        if kind == "stdio":
            args = server.get("args") or []
            summary = f"{server.get('command', '')} {' '.join(args)}".strip()
        elif kind in _SUPPORTED_REMOTE_TYPES:
            summary = str(server.get("url", ""))
        else:
            summary = ""
        results.append((name, kind, summary))
    return results


def merge_mcp_configs(configs: list[dict[str, Any]]) -> dict[str, Any]:
    """Merge multiple MCP config dicts by server name.

    Args:
        configs: Config dictionaries in ascending precedence order.

    Returns:
        A single config dict with later server definitions overriding earlier ones.
    """
    merged: dict[str, Any] = {}
    for config in configs:
        servers = config.get("mcpServers")
        if isinstance(servers, dict):
            merged.update(servers)
    return {"mcpServers": merged}


def load_mcp_config_lenient(config_path: Path) -> dict[str, Any] | None:
    """Load an MCP config file, returning `None` on any error.

    Args:
        config_path: Config path to load.

    Returns:
        The parsed config, or `None` if loading or validation fails.
    """
    config, _ = load_mcp_config_with_error(config_path)
    return config


def load_mcp_config_with_error(
    config_path: Path,
) -> tuple[dict[str, Any] | None, str | None]:
    """Load an MCP config file, returning `(config, error)`.

    Missing files yield `(None, None)` — not an error. Malformed files
    yield `(None, error_text)` so callers can surface the reason to users.

    Args:
        config_path: Config path to load.

    Returns:
        `(parsed_config, None)` on success, `(None, None)` when the file
        doesn't exist, or `(None, error_message)` on load/validate failure.
    """
    try:
        return load_mcp_config(str(config_path)), None
    except FileNotFoundError:
        return None, None
    except OSError as exc:
        logger.warning("Skipping unreadable MCP config %s: %s", config_path, exc)
        return None, f"Unreadable: {exc}"
    except (json.JSONDecodeError, ValueError, TypeError, RuntimeError) as exc:
        logger.warning("Skipping invalid MCP config %s: %s", config_path, exc)
        return None, str(exc)


def _check_stdio_server(server_name: str, server_config: dict[str, Any]) -> None:
    """Verify that a stdio server's command exists on PATH.

    Args:
        server_name: Server name for error messages.
        server_config: Validated server config.

    Raises:
        RuntimeError: If the command is missing or not found on PATH.
    """
    command = server_config.get("command")
    if command is None:
        msg = f"MCP server '{server_name}': missing 'command' in config."
        raise RuntimeError(msg)
    if shutil.which(command) is None:
        msg = (
            f"MCP server '{server_name}': command '{command}' not found on PATH. "
            "Install it or check your MCP config."
        )
        raise RuntimeError(msg)


async def _check_remote_server(server_name: str, server_config: dict[str, Any]) -> None:
    """Check network connectivity to a remote MCP server URL.

    Args:
        server_name: Server name for error messages.
        server_config: Validated remote server config.

    Raises:
        RuntimeError: If the URL is missing, unreachable, or returns 5xx.
    """
    import httpx

    url = server_config.get("url")
    if url is None:
        msg = f"MCP server '{server_name}': missing 'url' in config."
        raise RuntimeError(msg)
    try:
        async with httpx.AsyncClient(timeout=2.0) as client:
            response = await client.head(url)
    except (httpx.HTTPError, httpx.InvalidURL, OSError) as exc:
        msg = (
            f"MCP server '{server_name}': URL '{url}' is unreachable: {exc}. "
            "Check that the URL is correct and the server is running."
        )
        raise RuntimeError(msg) from exc
    if response.status_code >= 500:  # noqa: PLR2004  # HTTP server-error band
        msg = (
            f"MCP server '{server_name}': {url} returned HTTP "
            f"{response.status_code}. Server may be down; retry later."
        )
        raise RuntimeError(msg)


async def _discover_tools(session: ClientSession) -> list[Any]:
    """Enumerate MCP tools from `session`, paginating until exhausted.

    Args:
        session: Initialized MCP client session.

    Returns:
        Discovered MCP tool definitions.

    Raises:
        RuntimeError: If pagination never terminates within the hard safety bound.
    """
    cursor: str | None = None
    tools: list[Any] = []
    for _ in range(1000):
        page = await session.list_tools(cursor=cursor)
        if page.tools:
            tools.extend(page.tools)
        if not page.nextCursor:
            return tools
        cursor = page.nextCursor
    msg = (
        "Reached max of 1000 iterations while listing MCP tools; "
        "server may be returning a non-terminating cursor."
    )
    raise RuntimeError(msg)


def _normalize_mcp_arguments(
    arguments: dict[str, Any],
    input_schema: Any,  # noqa: ANN401  # raw JSON Schema dict from the MCP tool
) -> dict[str, Any]:
    """Drop empty-string values for optional MCP tool params.

    Some MCP servers (e.g. Slack's `slack_search_public_and_private`) validate
    optional ID-typed params with `value is not a channel ID` when the model
    fills them in with `""` instead of omitting them. JSON-Schema-derived
    Pydantic models happily accept `""` for `Optional[str]`, so the request
    reaches the server and gets rejected with a generic `ToolException`.

    Treat `""` for non-required string fields as "omitted" so the MCP server
    sees the same payload it would have for a field the model genuinely
    skipped. Required fields are passed through unchanged so the server's
    own missing-field error path still runs when applicable.

    Only `""` is normalized; `None` is left to the caller / server. Schemas
    that declare `["string", "null"]` will see `""` dropped but `None`
    forwarded — callers that want symmetric "no value" handling should
    omit the kwarg explicitly.

    Dropped keys are logged at debug so unexpected MCP behavior is
    diagnosable when a tool semantically distinguishes `""` from omitted.

    Args:
        arguments: Keyword arguments collected by LangChain's tool runner.
        input_schema: The MCP tool's `inputSchema` (raw JSON Schema dict).

    Returns:
        A new dict suitable for `session.call_tool`.
    """
    if not isinstance(input_schema, dict):
        return arguments
    required = set(input_schema.get("required") or ())
    properties = input_schema.get("properties") or {}
    cleaned: dict[str, Any] = {}
    for key, value in arguments.items():
        if value != "" or key in required:  # noqa: PLC1901  # distinguishing "" from other falsy types (0, False, []) is the point
            cleaned[key] = value
            continue
        prop = properties.get(key)
        prop_type = prop.get("type") if isinstance(prop, dict) else None
        is_string_typed = prop_type == "string" or (
            isinstance(prop_type, list) and "string" in prop_type
        )
        # Three drop conditions converge here:
        #   - explicit string type (the original Slack-style failure mode);
        #   - missing `type` (oneOf/anyOf/$ref or untyped — treat as ambiguous
        #     and conservatively drop, since the server will reject `""` for
        #     any ID-shaped slot anyway);
        #   - key absent from `properties` entirely (model invented a field).
        # Anything with an explicit non-string `type` is kept — `""` can't be
        # a valid integer/bool/array so it was the model's mistake to send,
        # and the server's own validation gives a clearer error than ours.
        if isinstance(prop, dict) and not is_string_typed and prop_type is not None:
            cleaned[key] = value
    if cleaned.keys() != arguments.keys():
        dropped = sorted(set(arguments) - set(cleaned))
        logger.debug("MCP arg normalize: dropped empty-string keys %s", dropped)
    return cleaned


def _build_cached_mcp_tool(
    *,
    mcp_tool: Any,  # noqa: ANN401
    server_name: str,
    session_manager: MCPSessionManager,
    tool_name_prefix: bool,
) -> BaseTool:
    """Build a `StructuredTool` backed by the cached session manager.

    Args:
        mcp_tool: MCP tool metadata object.
        server_name: Owning MCP server name.
        session_manager: Runtime session cache used for tool calls.
        tool_name_prefix: Whether to prefix the LangChain tool name with the
            server name.

    Returns:
        A LangChain `BaseTool` wrapper around the MCP tool.
    """
    from langchain_core.tools import StructuredTool, ToolException
    from langchain_mcp_adapters.tools import (
        _convert_call_tool_result,  # noqa: PLC2701
        _handle_mcp_tool_error,  # noqa: PLC2701
    )

    original_tool_name = mcp_tool.name
    lc_tool_name = (
        f"{server_name}_{original_tool_name}"
        if tool_name_prefix and server_name
        else original_tool_name
    )

    meta = getattr(mcp_tool, "meta", None)
    base_meta = (
        mcp_tool.annotations.model_dump() if mcp_tool.annotations is not None else {}
    )
    wrapped_meta = {"_meta": meta} if meta is not None else {}
    metadata = {**base_meta, **wrapped_meta} or None

    def _handle_cached_mcp_tool_error(error: ToolException) -> Any:  # noqa: ANN401
        try:
            return _handle_mcp_tool_error(error)
        except ToolException:
            logger.warning(
                "MCP tool %r failed with recoverable ToolException: %s",
                lc_tool_name,
                error,
                exc_info=True,
            )
            return str(error) or f"{lc_tool_name} failed with no error detail"

    async def coroutine(
        # `runtime` is injected by LangChain's tool-calling plumbing.
        # MCP tools don't use it but the kwarg must still be accepted.
        runtime: Any = None,  # noqa: ANN401, ARG001
        **arguments: Any,
    ) -> Any:  # noqa: ANN401
        from deepagents_code.mcp_auth import find_reauth_required

        arguments = _normalize_mcp_arguments(arguments, mcp_tool.inputSchema)

        session = await session_manager.get_session(server_name)
        try:
            result = await session.call_tool(original_tool_name, arguments)
        # Re-raise control-flow/shutdown signals (CancelledError,
        # KeyboardInterrupt, SystemExit) and ToolException unchanged. Wrapping a
        # ToolException here would bury its actionable message (e.g. an MCP
        # `isError` instruction like "use the X tool instead") under a generic
        # retry wrapper; re-raising preserves it for the tool-local error
        # handler and the model.
        except (asyncio.CancelledError, KeyboardInterrupt, SystemExit, ToolException):
            raise
        except Exception as exc:
            reauth = find_reauth_required(exc)
            if reauth is not None:
                await session_manager.invalidate(
                    server_name,
                    expected_session=session,
                )
                raise ToolException(str(reauth)) from exc
            if not _is_transient_session_error(exc):
                msg = (
                    f"MCP tool {lc_tool_name!r} failed on server "
                    f"{server_name!r}: {type(exc).__name__}: {exc}"
                )
                raise ToolException(msg) from exc
            logger.info(
                "MCP session for %r appears dead (%s: %s); "
                "invalidating and retrying once",
                server_name,
                type(exc).__name__,
                exc,
            )
            await session_manager.invalidate(
                server_name,
                expected_session=session,
            )

            retry_session = await session_manager.get_session(server_name)
            try:
                result = await retry_session.call_tool(original_tool_name, arguments)
            except (
                asyncio.CancelledError,
                KeyboardInterrupt,
                SystemExit,
                ToolException,
            ):
                raise
            except Exception as retry_exc:  # noqa: BLE001 - wrapped into ToolException below so the agent sees it
                try:
                    retry_reauth = find_reauth_required(retry_exc)
                    if retry_reauth is not None:
                        raise ToolException(str(retry_reauth)) from retry_exc
                    msg = (
                        f"MCP tool {lc_tool_name!r} failed after one retry on "
                        f"server {server_name!r}: {type(retry_exc).__name__}: "
                        f"{retry_exc}"
                    )
                    raise ToolException(msg) from retry_exc
                finally:
                    # Invalidate the retry session last; log cleanup failure
                    # so resource leaks are observable.
                    try:
                        await session_manager.invalidate(
                            server_name,
                            expected_session=retry_session,
                        )
                    except Exception:
                        logger.warning(
                            "Failed to invalidate retry session for %r after "
                            "tool failure",
                            server_name,
                            exc_info=True,
                        )

        # On an MCP `isError=True` result the adapter's `_convert_call_tool_result`
        # raises, and the `handle_tool_error` callback registered below converts
        # the MCP content blocks into a `ToolMessage(status="error")`. Other
        # expected `ToolException`s raised by this wrapper are formatted by that
        # same tool-local handler.
        return _convert_call_tool_result(result)

    return StructuredTool(
        name=lc_tool_name,
        description=mcp_tool.description or "",
        args_schema=mcp_tool.inputSchema,
        coroutine=coroutine,
        response_format="content_and_artifact",
        metadata=metadata,
        handle_tool_error=cast("Any", _handle_cached_mcp_tool_error),
    )


_GLOB_METACHARS = frozenset("*?[")


def _entry_matches_tool(entry: str, tool_name: str, prefix: str) -> bool:
    """Return True if a single filter entry matches a tool name.

    An entry containing `*`, `?`, or `[` is treated as an `fnmatch`-style glob;
    otherwise it is matched literally. Each entry is tried against both the
    bare MCP tool name and the server-prefixed form (`f"{prefix}{tool}"`), so
    users can write either `read_*` or `fs_read_*`.

    Args:
        entry: Filter list entry from `allowedTools` / `disabledTools`.
        tool_name: Adapter-supplied tool name (already server-prefixed).
        prefix: Server prefix (`f"{server_name}_"`).

    Returns:
        True if the entry matches this tool under either match mode.
    """
    is_glob = any(ch in _GLOB_METACHARS for ch in entry)
    if is_glob:
        if fnmatch.fnmatchcase(tool_name, entry):
            return True
        if tool_name.startswith(prefix):
            return fnmatch.fnmatchcase(tool_name[len(prefix) :], entry)
        return False
    if tool_name == entry:
        return True
    return tool_name.startswith(prefix) and tool_name[len(prefix) :] == entry


@overload
def _apply_tool_filter(
    tools: list[BaseTool],
    server_name: str,
    server_config: dict[str, Any],
) -> list[BaseTool]: ...


@overload
def _apply_tool_filter(
    tools: Sequence[BaseTool],
    server_name: str,
    server_config: dict[str, Any],
) -> Sequence[BaseTool]: ...


def _apply_tool_filter(
    tools: Sequence[BaseTool],
    server_name: str,
    server_config: dict[str, Any],
) -> Sequence[BaseTool]:
    """Filter a server's loaded tools by its `allowedTools` / `disabledTools`.

    Entries may be literal tool names or `fnmatch`-style glob patterns
    (entries containing `*`, `?`, or `[`). Each entry is tried against both
    the bare MCP tool name and the server-prefixed name produced by
    `tool_name_prefix=True` (`f"{server_name}_{tool}"`). Entries that match
    no loaded tool are logged but not an error — the underlying MCP server
    may expose different tools across versions, so a stale entry should not
    fail startup. The same warning is emitted symmetrically for both fields
    so a typo in `disabledTools` is visible (otherwise a tool the user
    intended to disable would silently remain enabled).

    Args:
        tools: Tools returned by `load_mcp_tools` for a single server.
        server_name: Server name used by the adapter to build the prefix.
        server_config: Server config dict (read for filter fields).

    Returns:
        Filtered tool list preserving input order.
    """
    allowed: list[str] | None = server_config.get("allowedTools")
    disabled: list[str] | None = server_config.get("disabledTools")
    entries: list[str] | None = allowed if allowed is not None else disabled
    if entries is None:
        return tools

    prefix = f"{server_name}_"
    field_name = "allowedTools" if allowed is not None else "disabledTools"

    def _any_entry_matches(tool_name: str, entry_list: list[str]) -> bool:
        return any(_entry_matches_tool(e, tool_name, prefix) for e in entry_list)

    missing = [
        e
        for e in entries
        if not any(_entry_matches_tool(e, t.name, prefix) for t in tools)
    ]
    if missing:
        logger.warning(
            "MCP server '%s' %s entries matched no tools: %s",
            server_name,
            field_name,
            ", ".join(missing),
        )

    if allowed is not None:
        return [t for t in tools if _any_entry_matches(t.name, entries)]
    return [t for t in tools if not _any_entry_matches(t.name, entries)]


async def _load_tools_from_config(
    config: dict[str, Any],
    *,
    stateless: bool = False,
    session_manager: MCPSessionManager | None = None,
) -> tuple[list[BaseTool], MCPSessionManager | None, list[MCPServerInfo]]:
    """Build MCP connections from a validated config and load tools.

    Discovery always opens throwaway sessions to capture tool metadata only.
    Runtime tools either:

    - bind to a caller-managed `session_manager` (server mode),
    - bind to a new local `session_manager` returned to the caller, or
    - stay fully stateless and open a fresh session per tool call.

    Per-server config/auth/setup failures are captured in the returned
    `server_infos` list rather than propagated — one bad server never
    hides the others.

    Args:
        config: Validated MCP configuration dict with `mcpServers` key.
        stateless: When `True`, tools avoid returning an owned session manager.
        session_manager: Optional externally owned runtime session manager.

    Returns:
        Tuple of `(tools_list, session_manager, server_infos)`.

    Raises:
        RuntimeError: If `session_manager` is reconfigured incompatibly with
            sessions already active on it.
    """  # noqa: DOC501, DOC502 - `RuntimeError` surfaces via `MCPSessionManager.configure`; `KeyboardInterrupt` / `SystemExit` / `CancelledError` are re-raised pass-throughs
    from langchain_mcp_adapters.sessions import (
        SSEConnection,
        StdioConnection,
        StreamableHttpConnection,
        create_session,
    )
    from langchain_mcp_adapters.tools import convert_mcp_tool_to_langchain_tool

    skipped: dict[str, tuple[MCPServerStatus, str]] = {}

    for server_name, server_config in config["mcpServers"].items():
        server_type = _resolve_server_type(server_config)
        try:
            if server_type in _SUPPORTED_REMOTE_TYPES:
                await _check_remote_server(server_name, server_config)
            elif server_type == "stdio":
                _check_stdio_server(server_name, server_config)
        except RuntimeError as exc:
            logger.warning(
                "MCP server '%s' skipped: pre-flight failed: %s",
                server_name,
                exc,
            )
            skipped[server_name] = ("error", str(exc))

    connections: dict[str, Connection] = {}
    for server_name, server_config in config["mcpServers"].items():
        if server_name in skipped:
            continue
        server_type = _resolve_server_type(server_config)
        try:
            if server_type in _SUPPORTED_REMOTE_TYPES:
                if server_type == "http":
                    conn: Connection = StreamableHttpConnection(
                        transport="streamable_http",
                        url=server_config["url"],
                    )
                else:
                    conn = SSEConnection(
                        transport="sse",
                        url=server_config["url"],
                    )

                if "headers" in server_config:
                    from deepagents_code.mcp_auth import resolve_headers

                    conn["headers"] = resolve_headers(
                        server_config["headers"],
                        server_name=server_name,
                    )

                if server_config.get("auth") == "oauth":
                    from deepagents_code.mcp_auth import (
                        FileTokenStorage,
                        build_oauth_provider,
                    )

                    storage = FileTokenStorage(
                        server_name,
                        server_url=server_config["url"],
                    )
                    if await storage.get_tokens() is None:
                        auth_msg = (
                            f"MCP server {server_name!r} needs re-authentication."
                        )
                        logger.warning(
                            "MCP server '%s' skipped: not authenticated.",
                            server_name,
                        )
                        skipped[server_name] = ("unauthenticated", auth_msg)
                        continue
                    conn["auth"] = build_oauth_provider(
                        server_name=server_name,
                        server_url=server_config["url"],
                        storage=storage,
                        interactive=False,
                    )

                connections[server_name] = conn
            else:
                connections[server_name] = StdioConnection(
                    command=server_config["command"],
                    args=server_config.get("args", []),
                    env=server_config.get("env") or None,
                    transport="stdio",
                )
        except (OSError, RuntimeError, TypeError, ValueError) as exc:
            logger.warning(
                "MCP server '%s' skipped: config/setup failed: %s",
                server_name,
                exc,
            )
            skipped[server_name] = ("error", str(exc))

    runtime_manager: MCPSessionManager | None = session_manager
    if runtime_manager is not None:
        runtime_manager.configure(connections)
    elif not stateless:
        runtime_manager = MCPSessionManager(connections=connections)

    all_tools: list[BaseTool] = []
    server_infos: list[MCPServerInfo] = []

    for server_name, server_config in config["mcpServers"].items():
        transport = _resolve_server_type(server_config)
        if server_name in skipped:
            status, error = skipped[server_name]
            server_infos.append(
                MCPServerInfo(
                    name=server_name,
                    transport=transport,
                    status=status,
                    error=error,
                ),
            )
            continue

        try:
            async with create_session(connections[server_name]) as discover_session:
                await discover_session.initialize()
                mcp_tools = await _discover_tools(discover_session)
        except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
            raise
        except Exception as exc:
            from deepagents_code.mcp_auth import find_reauth_required

            reauth = find_reauth_required(exc)
            status: MCPServerStatus
            if reauth is not None:
                # Tokens existed (we checked above) but the OAuth provider
                # fell back to interactive reauth — the refresh attempt
                # failed. Flag unauthenticated so the user is prompted to
                # re-login, and keep the original exception in the log so
                # debugging a real provider outage is possible.
                status = "unauthenticated"
                error = (
                    f"{reauth} "
                    "(token refresh failed; the original error is in debug logs)"
                )
            else:
                status = "error"
                error = str(exc)
            logger.warning(
                "MCP server '%s' skipped: tool discovery failed",
                server_name,
                exc_info=True,
            )
            server_infos.append(
                MCPServerInfo(
                    name=server_name,
                    transport=transport,
                    status=status,
                    error=error,
                ),
            )
            continue

        if runtime_manager is None:
            server_tools = [
                convert_mcp_tool_to_langchain_tool(
                    None,
                    mcp_tool,
                    connection=connections[server_name],
                    server_name=server_name,
                    tool_name_prefix=True,
                )
                for mcp_tool in mcp_tools
            ]
        else:
            server_tools = [
                _build_cached_mcp_tool(
                    mcp_tool=mcp_tool,
                    server_name=server_name,
                    session_manager=runtime_manager,
                    tool_name_prefix=True,
                )
                for mcp_tool in mcp_tools
            ]

        server_tools = _apply_tool_filter(server_tools, server_name, server_config)
        all_tools.extend(server_tools)

        # Pair each tool's input_schema by its LangChain (server-prefixed)
        # name — the same form `server_tools` carries — so the lookup needs
        # no string surgery and stays correct if `tool_name_prefix` ever
        # changes. Deep-copy the raw dict because `MCPToolInfo` is `frozen`
        # but Python's `frozen=True` does not freeze nested mutables; a
        # shared reference would let one holder mutate every other's view.
        schemas: dict[str, dict[str, Any] | None] = {}
        for mcp_tool in mcp_tools:
            tool_name = getattr(mcp_tool, "name", "")
            try:
                raw_schema = getattr(mcp_tool, "inputSchema", None)
                schema_copy = (
                    copy.deepcopy(raw_schema) if raw_schema is not None else None
                )
            except (AttributeError, TypeError, RecursionError) as exc:
                logger.warning(
                    "MCP tool %r on server %r: inputSchema access raised %s: %s; "
                    "rendering with no parameters",
                    tool_name,
                    server_name,
                    exc.__class__.__name__,
                    exc,
                )
                schema_copy = None
            lc_name = f"{server_name}_{tool_name}"
            schemas[lc_name] = schema_copy

        tool_infos: list[MCPToolInfo] = []
        for tool in server_tools:
            schema = schemas.get(tool.name)
            if schema is None and schemas:
                logger.debug(
                    "MCP tool %r on server %r: no schema matched in lookup "
                    "(available keys: %s); rendering with no parameters",
                    tool.name,
                    server_name,
                    list(schemas.keys())[:5],
                )
            tool_infos.append(
                MCPToolInfo(
                    name=tool.name,
                    description=tool.description or "",
                    input_schema=schema,
                ),
            )
        server_infos.append(
            MCPServerInfo(
                name=server_name,
                transport=transport,
                tools=tuple(tool_infos),
            ),
        )

    all_tools.sort(key=lambda tool: tool.name)
    return all_tools, None if stateless else runtime_manager, server_infos


async def get_mcp_tools(
    config_path: str,
) -> tuple[list[BaseTool], MCPSessionManager | None, list[MCPServerInfo]]:
    """Load MCP tools from a configuration file.

    Args:
        config_path: Path to an MCP config file.

    Returns:
        Tuple of `(tools_list, runtime_session_manager, server_infos)`.

    Raises:
        FileNotFoundError: If `config_path` doesn't exist.
        json.JSONDecodeError: If the config file contains invalid JSON.
        TypeError: If config fields have wrong types.
        ValueError: If the config is missing required fields.
    """  # noqa: DOC502 - surfaced via `load_mcp_config`
    config = load_mcp_config(config_path)
    return await _load_tools_from_config(config)


async def resolve_and_load_mcp_tools(
    *,
    explicit_config_path: str | None = None,
    no_mcp: bool = False,
    trust_project_mcp: bool | None = None,
    project_context: ProjectContext | None = None,
    stateless: bool = False,
    session_manager: MCPSessionManager | None = None,
) -> tuple[list[BaseTool], MCPSessionManager | None, list[MCPServerInfo]]:
    """Resolve MCP config and load tools.

    Auto-discovers configs from standard locations and merges them. When
    `explicit_config_path` is provided it is added as the highest-precedence
    source and errors in that file are fatal.

    Args:
        explicit_config_path: Extra config file to layer on top of
            auto-discovered configs.
        no_mcp: If `True`, disable all MCP loading.
        trust_project_mcp: Controls project-level stdio server trust.

            - `True`: always trust project configs, including stdio servers.
            - `False`: drop stdio entries from project configs.
            - `None`: consult the persistent trust store — trusted configs
              load fully, untrusted project stdio servers are dropped.
        project_context: Explicit project path context for config discovery
            and trust resolution.
        stateless: When `True`, do not return an owned runtime session manager.
        session_manager: Optional externally owned runtime session manager.

    Returns:
        Tuple of `(tools_list, session_manager, server_infos)`.

    Raises:
        FileNotFoundError: If `explicit_config_path` was provided and points
            at a missing file.
        json.JSONDecodeError: If `explicit_config_path` contains invalid
            JSON.
        TypeError: If `explicit_config_path` contents have wrong field
            types.
        ValueError: If `explicit_config_path` is missing required fields
            or declares an unsupported transport.
        RuntimeError: If the merged MCP config is malformed, or header
            env-var interpolation in `explicit_config_path` references an
            unset variable.
    """  # noqa: DOC502 - FileNotFoundError / JSONDecodeError / TypeError / ValueError surface via `load_mcp_config`
    if no_mcp:
        return [], None, []

    config_load_errors: list[tuple[Path, str]] = []

    try:
        config_paths = discover_mcp_configs(project_context=project_context)
    except (OSError, RuntimeError) as exc:
        logger.warning("MCP config auto-discovery failed", exc_info=True)
        config_paths = []
        config_load_errors.append((Path("<discovery>"), str(exc)))

    user_configs, project_configs = classify_discovered_configs(config_paths)
    configs: list[dict[str, Any]] = []

    for path in user_configs:
        config, error = load_mcp_config_with_error(path)
        if error is not None:
            config_load_errors.append((path, error))
        if config is not None:
            configs.append(config)

    project_trusted: bool | None = None
    for path in project_configs:
        config, error = load_mcp_config_with_error(path)
        if error is not None:
            config_load_errors.append((path, error))
        if config is None:
            continue

        project_servers = extract_project_server_summaries(config)
        if not project_servers:
            configs.append(config)
            continue

        if trust_project_mcp is True:
            configs.append(config)
            continue

        if trust_project_mcp is None and project_trusted is None:
            from deepagents_code.mcp_trust import (
                compute_config_fingerprint,
                is_project_mcp_trusted,
            )

            project_root = str(_resolve_project_config_base(project_context).resolve())
            fingerprint = compute_config_fingerprint(project_configs)
            project_trusted = is_project_mcp_trusted(project_root, fingerprint)

        if project_trusted is True:
            configs.append(config)
            continue

        # Untrusted project config: drop ALL servers (stdio + remote). Remote
        # entries from an attacker-controlled .mcp.json can SSRF localhost or
        # cloud metadata endpoints during the preflight HEAD probe, and any
        # `${VAR}` references in their `headers` would exfiltrate the value
        # to the attacker URL during the discovery handshake.
        skipped = [
            f"{name} [{kind}]: {summary}" for name, kind, summary in project_servers
        ]
        if trust_project_mcp is False:
            logger.warning(
                "Skipped untrusted project MCP servers: %s",
                "; ".join(skipped),
            )
        else:
            logger.warning(
                "Skipped untrusted project MCP servers "
                "(config changed or not yet approved): %s",
                "; ".join(skipped),
            )

    if explicit_config_path:
        config_path = (
            str(project_context.resolve_user_path(explicit_config_path))
            if project_context is not None
            else explicit_config_path
        )
        configs.append(load_mcp_config(config_path))

    def _bad_config_infos() -> list[MCPServerInfo]:
        return [
            MCPServerInfo(
                name=f"<config:{path.name}>",
                transport="config",
                status="error",
                error=f"{path}: {error}",
            )
            for path, error in config_load_errors
        ]

    if not configs:
        return [], None, _bad_config_infos()

    merged = merge_mcp_configs(configs)
    if not merged.get("mcpServers"):
        return [], None, _bad_config_infos()

    from deepagents_code.mcp_disabled import get_disabled_servers

    disabled_names = get_disabled_servers()
    disabled_infos: list[MCPServerInfo] = []
    if disabled_names:
        active: dict[str, Any] = {}
        for server_name, server_config in merged["mcpServers"].items():
            if server_name in disabled_names:
                disabled_infos.append(
                    MCPServerInfo(
                        name=server_name,
                        transport=_resolve_server_type(server_config)
                        if isinstance(server_config, dict)
                        else "unknown",
                        status="disabled",
                        error="Disabled by user (`/mcp` F2 to re-enable).",
                    ),
                )
            else:
                active[server_name] = server_config
        merged = {"mcpServers": active}

    if not merged.get("mcpServers"):
        return [], None, disabled_infos + _bad_config_infos()

    try:
        for server_name, server_config in merged["mcpServers"].items():
            _validate_server_config(server_name, server_config)
    except (TypeError, ValueError, RuntimeError) as exc:
        msg = f"Invalid MCP server configuration: {exc}"
        raise RuntimeError(msg) from exc

    tools, manager, server_infos = await _load_tools_from_config(
        merged,
        stateless=stateless,
        session_manager=session_manager,
    )
    server_infos.extend(disabled_infos)
    server_infos.extend(_bad_config_infos())
    return tools, manager, server_infos
