"""MCP Mount Manager for external MCP server integration.

This module provides functionality for mounting external MCP servers
(e.g., GitHub, Slack, Notion) and discovering their tools.

Based on: https://www.anthropic.com/engineering/code-execution-with-mcp
"""

import asyncio
import json
import logging
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import urlparse

from nexus.bricks.mcp.models import MCPMount, MCPToolConfig, MCPToolDefinition
from nexus.config import SSRFConfig
from nexus.lib.events import emit_audit_event
from nexus.lib.security.ssrf_transport import make_pinned_client_factory
from nexus.lib.security.url_validator import SSRFBlocked, validate_outbound_url

if TYPE_CHECKING:
    from nexus.bricks.approvals.policy_gate import PolicyGate
    from nexus.contracts.types import OperationContext
    from nexus.core.nexus_fs import NexusFS


logger = logging.getLogger(__name__)


def _host_port_from_url(url: str) -> str | None:
    """Return ``host:port`` for an http/https URL, or None if unparseable.

    Used as the ``subject`` for ``EGRESS_HOST`` approval requests so
    operators can decide once per host (not per URL path). Defaults to
    443 for https and 80 for http when the URL omits an explicit port.
    """
    try:
        parsed = urlparse(url)
    except Exception:
        return None
    host = parsed.hostname
    if not host:
        return None
    if parsed.port:
        port = parsed.port
    elif parsed.scheme == "https":
        port = 443
    elif parsed.scheme == "http":
        port = 80
    else:
        return None
    return f"{host}:{port}"


class MCPMountError(Exception):
    """Raised when MCP mount operations fail."""

    is_expected = True

    def __init__(self, message: str, path: str | None = None):
        self.message = message
        self.path = path
        super().__init__(message)


class MCPMountManager:
    """Manager for mounting and interacting with external MCP servers.

    Features:
    - Mount external MCP servers (stdio, http, sse transports)
    - Discover tools from mounted servers via list_tools()
    - Store tool definitions in /skills/system/mcp-tools/{mount_name}/
    - Execute tools on mounted servers
    - Manage authentication (OAuth, API keys)

    Storage Structure (per-folder):
        /skills/system/mcp-tools/
        ├── github/
        │   ├── mount.json              # Connection info for this mount
        │   ├── README.md                # Human-readable docs
        │   ├── search_repositories.json
        │   └── create_issue.json
        └── slack/
            ├── mount.json
            ├── README.md
            └── send_message.json

    Example:
        >>> from nexus import connect
        >>> from nexus.bricks.mcp.mount import MCPMountManager
        >>>
        >>> nx = connect()
        >>> manager = MCPMountManager(nx)
        >>>
        >>> # Mount GitHub MCP server
        >>> await manager.mount(MCPMount(
        ...     name="github",
        ...     description="GitHub API integration",
        ...     transport="stdio",
        ...     command="npx",
        ...     args=["-y", "@modelcontextprotocol/server-github"],
        ...     env={"GITHUB_PERSONAL_ACCESS_TOKEN": "..."}
        ... ))
        >>>
        >>> # Sync tools (discover available tools)
        >>> await manager.sync_tools("github")
        >>>
        >>> # List mounted servers
        >>> mounts = manager.list_mounts()
    """

    # Tier priority for MCP mounts (higher = checked first, wins on name conflict)
    # MCP mounts have 3 levels: user > zone > system (no agent level)
    TIER_PRIORITY = {
        "user": 3,
        "zone": 2,
        "system": 1,
    }

    @staticmethod
    def get_mcp_tier_paths(context: "OperationContext | None" = None) -> dict[str, str]:
        """Get context-aware tier paths for MCP tool discovery.

        Structure:
            /skills/system/mcp-tools/               - System-wide MCP tools (priority 1)
            /skills/zones/{zone_id}/mcp-tools/  - Zone shared MCP tools (priority 2)
            /skills/users/{user_id}/mcp-tools/      - User personal MCP tools (priority 3)

        Args:
            context: Operation context with user_id, zone_id

        Returns:
            Dict mapping tier name to mcp-tools path (only tiers available for this context)
        """
        paths = {"system": "/skills/system/mcp-tools/"}

        if context:
            if context.zone_id:
                paths["zone"] = f"/skills/zones/{context.zone_id}/mcp-tools/"

            if context.user_id:
                paths["user"] = f"/skills/users/{context.user_id}/mcp-tools/"

        return paths

    # Legacy: Base path for MCP tools (system level only)
    # Use get_mcp_tier_paths(context) for context-aware paths
    MCP_TOOLS_PATH = "/skills/system/mcp-tools/"

    # Mount configuration filename (per-folder)
    MOUNT_CONFIG_FILENAME = "mount.json"

    def __init__(
        self,
        filesystem: "NexusFS | None" = None,
        *,
        ssrf_config: "SSRFConfig | None" = None,
        policy_gate: "PolicyGate | None" = None,
        zone_id: str | None = None,
    ) -> None:
        """Initialize MCP mount manager.

        Args:
            filesystem: Optional filesystem instance (defaults to local FS)
            ssrf_config: Optional SSRFConfig override. When provided, the
                validator for SSE/HTTP mounts uses these operator settings
                (``allow_private`` / ``extra_deny_cidrs``). When None, a
                conservative default ``SSRFConfig()`` is used.
            policy_gate: Optional PolicyGate for approval-queue routing
                (Issue #3790, Task 18). When provided, an SSRF-blocked egress
                attempt for an unlisted host is offered to the gate first;
                if an operator approves the host the connect proceeds, else
                the original ``SSRFBlocked`` is re-raised. When None, the
                manager preserves the existing fail-closed behavior.
            zone_id: Optional zone identifier for approval-queue
                attribution (Issue #3790, F4). When set, SSRF-blocked
                egress requests are filed against this zone; when ``None``
                the manager fails closed instead of charging the request
                to ROOT_ZONE_ID. Operators wire this from the daemon's
                primary zone (or per-mount config when multi-zone is
                supported in a future revision); see
                :meth:`set_zone` for post-construction attachment.
        """
        self._filesystem = filesystem
        self._ssrf_config_override = ssrf_config
        self._policy_gate = policy_gate
        self._zone_id = zone_id

        # Active mount configurations (keyed by name, may include tier info)
        self._mounts: dict[str, MCPMount] = {}

        # Tier index: tier -> list of mount names (for context-aware lookup)
        self._tier_index: dict[str, list[str]] = {}

        # Active MCP client connections
        self._clients: dict[str, Any] = {}

        # Mount configs loaded lazily on first async access (syscalls are async)
        self._mounts_loaded = False

    def set_zone(self, zone_id: str | None) -> None:
        """Attach (or detach) the daemon zone after construction.

        F4 (Issue #3790): the approvals lifespan determines the daemon's
        primary zone after MCP services are constructed, so callers may
        attach the zone post-hoc. ``None`` triggers fail-closed routing
        in :meth:`_ssrf_blocked_via_gate` (the gate is bypassed entirely
        and the original ``SSRFBlocked`` is re-raised).
        """
        self._zone_id = zone_id

    async def _load_mounts_config(self) -> None:
        """Load mount configurations from per-folder mount.json files.

        Each mount has its own folder with mount.json:
        /skills/system/mcp-tools/{mount_name}/mount.json
        """
        if self._mounts_loaded:
            return
        self._mounts_loaded = True
        try:
            await self._load_mounts_from_folders()
        except Exception as e:
            logger.warning("Failed to load mount configurations: %s", e)

    async def _load_mounts_from_folders(self) -> bool:
        """Load mounts from per-folder mount.json files.

        Returns:
            True if any mounts were loaded
        """
        loaded_any = False

        try:
            if self._filesystem:
                # Check if base path exists
                if not self._filesystem.access(self.MCP_TOOLS_PATH):
                    return False

                # List directories in MCP_TOOLS_PATH
                items = await asyncio.to_thread(self._filesystem.sys_readdir, self.MCP_TOOLS_PATH)
                for item in items:
                    # Skip files at root level
                    item_path = f"{self.MCP_TOOLS_PATH}{item}"
                    mount_json_path = f"{item_path}/mount.json"

                    try:
                        if self._filesystem.access(mount_json_path):
                            raw_content = await asyncio.to_thread(
                                self._filesystem.sys_read, mount_json_path
                            )
                            content_str = (
                                raw_content.decode("utf-8")
                                if isinstance(raw_content, bytes)
                                else str(raw_content)
                            )
                            mount_data = json.loads(content_str)
                            mount = MCPMount.from_dict(mount_data)
                            mount.mounted = False
                            self._mounts[mount.name] = mount
                            loaded_any = True
                            logger.debug("Loaded mount config from %s", mount_json_path)
                    except Exception as e:
                        logger.warning("Failed to load mount from %s: %s", mount_json_path, e)
            else:
                # Local filesystem
                base_path = Path(self.MCP_TOOLS_PATH.lstrip("/"))
                if not base_path.exists():
                    return False

                for mount_dir in base_path.iterdir():
                    if mount_dir.is_dir():
                        mount_json_file = mount_dir / "mount.json"
                        if mount_json_file.exists():
                            try:
                                mount_data = json.loads(mount_json_file.read_text())
                                mount = MCPMount.from_dict(mount_data)
                                mount.mounted = False
                                self._mounts[mount.name] = mount
                                loaded_any = True
                                logger.debug("Loaded mount config from %s", mount_json_file)
                            except Exception as e:
                                logger.warning(
                                    "Failed to load mount from %s: %s", mount_json_file, e
                                )

        except Exception as e:
            logger.warning("Error scanning for mount configs: %s", e)

        return loaded_any

    async def _save_mounts_config(self) -> None:
        """Save mount configurations to per-folder mount.json files.

        Each mount is saved to its own folder:
        /skills/system/mcp-tools/{mount_name}/mount.json
        """
        for mount in self._mounts.values():
            await self._save_mount_config(mount)

    async def _save_mount_config(self, mount: MCPMount) -> None:
        """Save a single mount's configuration to its folder.

        Args:
            mount: Mount configuration to save
        """
        try:
            mount_json_path = f"{self.MCP_TOOLS_PATH}{mount.name}/{self.MOUNT_CONFIG_FILENAME}"
            content = json.dumps(mount.to_dict(), indent=2)

            if self._filesystem:
                # Ensure mount directory exists
                mount_dir = f"{self.MCP_TOOLS_PATH}{mount.name}/"
                try:
                    self._filesystem.mkdir(mount_dir, parents=True)
                except FileExistsError:
                    pass
                except OSError as e:
                    logger.warning("Failed to create directory %s: %s", mount_dir, e)

                self._filesystem.write(mount_json_path, content.encode("utf-8"))
            else:
                mount_path = Path(mount_json_path.lstrip("/"))
                mount_path.parent.mkdir(parents=True, exist_ok=True)
                mount_path.write_text(content)

            logger.debug("Saved mount config: %s", mount.name)
        except Exception as e:
            logger.error("Failed to save mount config for %s: %s", mount.name, e)

    async def mount(self, mount_config: MCPMount) -> bool:
        """Mount an external MCP server.

        Steps:
        1. Validate configuration
        2. Connect to MCP server
        3. Store configuration
        4. Mark as mounted

        Args:
            mount_config: Mount configuration

        Returns:
            True if mount successful

        Raises:
            MCPMountError: If mount fails
        """
        # Validate configuration
        if not mount_config.name:
            raise MCPMountError("Mount name is required")

        if mount_config.transport not in ("stdio", "http", "sse", "klavis_rest"):
            raise MCPMountError(f"Unsupported transport: {mount_config.transport}")

        if mount_config.transport == "stdio" and not mount_config.command:
            raise MCPMountError("Command is required for stdio transport")

        if mount_config.transport in ("http", "sse", "klavis_rest") and not mount_config.url:
            raise MCPMountError("URL is required for http/sse/klavis_rest transport")

        # Set tools path
        mount_config.tools_path = f"{self.MCP_TOOLS_PATH}{mount_config.name}/"

        # Try to connect to the MCP server and sync tools
        try:
            # Update mount state
            mount_config.mounted = True
            mount_config.mounted_at = datetime.now(UTC)

            # Store configuration
            self._mounts[mount_config.name] = mount_config
            await self._save_mounts_config()

            # Sync tools from the server
            await self.sync_tools(mount_config.name)

            logger.info("Mounted MCP server: %s", mount_config.name)
            return True

        except Exception as e:
            raise MCPMountError(f"Failed to mount {mount_config.name}: {e}") from e

    async def _create_client(self, mount_config: MCPMount) -> Any:
        """Create an MCP client connection.

        Args:
            mount_config: Mount configuration

        Returns:
            MCP client instance
        """
        if mount_config.transport == "stdio":
            return await self._create_stdio_client(mount_config)
        elif mount_config.transport in ("http", "sse"):
            return await self._create_sse_client(mount_config)
        else:
            raise MCPMountError(f"Unsupported transport: {mount_config.transport}")

    async def _create_stdio_client(self, mount_config: MCPMount) -> Any:
        """Create a stdio MCP client.

        Args:
            mount_config: Mount configuration

        Returns:
            MCP client session
        """
        try:
            from mcp import ClientSession, StdioServerParameters
            from mcp.client.stdio import stdio_client
        except ImportError as e:
            raise MCPMountError(
                "MCP client library not installed. Install with: pip install mcp"
            ) from e

        server_params = StdioServerParameters(
            command=mount_config.command or "",
            args=mount_config.args,
            env=mount_config.env or None,
        )

        # Create client session
        async with (
            stdio_client(server_params) as (read, write),
            ClientSession(read, write) as session,
        ):
            # Initialize the session
            await session.initialize()
            return session

    async def _create_sse_client(self, mount_config: MCPMount) -> Any:
        """Create an SSE/HTTP MCP client.

        Args:
            mount_config: Mount configuration

        Returns:
            MCP client session
        """
        try:
            from mcp import ClientSession
            from mcp.client.sse import sse_client
        except ImportError as e:
            raise MCPMountError(
                "MCP client library not installed. Install with: pip install mcp"
            ) from e

        if not mount_config.url:
            raise MCPMountError("URL is required for SSE/HTTP transport")

        ssrf_cfg = self._ssrf_config()
        try:
            validated = validate_outbound_url(
                mount_config.url,
                allow_private=ssrf_cfg.allow_private,
                extra_deny_cidrs=ssrf_cfg.extra_deny_cidrs,
            )
        except SSRFBlocked as exc:
            self._emit_ssrf_audit(mount_config, exc)
            logger.warning("SSRF blocked for MCP mount %r: %s", mount_config.name, exc)
            if not await self._ssrf_blocked_via_gate(
                mount_config, mount_config.url, "mcp_mount_connect"
            ):
                raise
            # Operator approved the host via the approvals queue. Re-validate
            # with private ranges relaxed so RFC 1918 / ULA hosts become
            # reachable; metadata, loopback, and the operator-configured
            # ``extra_deny_cidrs`` remain enforced. Operator approval covers
            # "this host" only — it must not silently disable the operator's
            # own deny list (e.g. internal service mesh CIDRs).
            try:
                validated = validate_outbound_url(
                    mount_config.url,
                    allow_private=True,
                    extra_deny_cidrs=ssrf_cfg.extra_deny_cidrs,
                )
            except SSRFBlocked:
                logger.warning(
                    "approvals-approved mount %r still failed SSRF re-validation; "
                    "honoring the original block",
                    mount_config.name,
                )
                raise exc from None

        # Start with custom headers from mount config
        headers: dict[str, str] = dict(mount_config.headers) if mount_config.headers else {}

        # Add authentication headers if needed (these override custom headers)
        if mount_config.auth_type == "api_key" and mount_config.auth_config:
            api_key = mount_config.auth_config.get("api_key")
            header_name = mount_config.auth_config.get("header_name", "Authorization")
            if api_key:
                headers[header_name] = f"Bearer {api_key}"

        # Cast to Any: mcp's ``McpHttpClientFactory`` is a Protocol and the
        # structural match against our Callable type is lost through mypy's
        # strict arg-type check (defaults vs positional).
        pinned_factory = cast(Any, make_pinned_client_factory(validated))

        # Create SSE client session — underlying httpx.AsyncClient is built
        # from the pinned factory so TCP connects go to the validated IPs
        # and redirects are disabled (prevents DNS-rebinding / redirect-
        # based SSRF escapes).
        async with (
            sse_client(
                mount_config.url,
                headers=headers,
                httpx_client_factory=pinned_factory,
            ) as (read, write),
            ClientSession(read, write) as session,
        ):
            await session.initialize()
            return session

    def _ssrf_config(self) -> SSRFConfig:
        """Return the effective SSRFConfig.

        Uses the override passed into __init__ when available so operator
        settings in ``NexusConfig.security.ssrf`` (``allow_private`` /
        ``extra_deny_cidrs``) take effect at SSE/HTTP mount validation
        time. Falls back to a conservative default ``SSRFConfig()`` when
        the manager was constructed without an override.
        """
        if self._ssrf_config_override is not None:
            return self._ssrf_config_override
        return SSRFConfig()

    async def _ssrf_blocked_via_gate(
        self,
        mount_config: MCPMount,
        url: str,
        operation: str,
    ) -> bool:
        """Consult the PolicyGate when SSRF would block this URL.

        Issue #3790, Task 18: route unlisted-host egress through the
        approval queue. Returns True iff an operator approved the host
        within the gate's timeout, in which case the caller may proceed
        as if SSRF had passed. Returns False on missing gate, denial,
        timeout, or any unexpected gate error (graceful degradation —
        the original ``SSRFBlocked`` is then re-raised by the caller).

        F4 (#3790): the zone is the daemon's configured zone (set via
        ``__init__`` or :meth:`set_zone`). When the manager has no zone
        bound, the gate is bypassed and the original ``SSRFBlocked`` is
        re-raised — this is fail-closed behavior. We do NOT silently
        charge approvals to ROOT_ZONE_ID, because that would let any
        operator with root-zone ``approvals:decide`` approve egress for
        every other zone's traffic.
        """
        gate = self._policy_gate
        if gate is None:
            return False

        host_port = _host_port_from_url(url)
        if host_port is None:
            # Cannot derive a meaningful subject — keep fail-closed.
            return False

        # F4 (#3790): require a bound zone. Without one we cannot scope
        # the approval to a single operator-decide capability set, so
        # routing the request through the queue would be a privilege
        # escalation. Fail closed; the caller will re-raise SSRFBlocked.
        if not self._zone_id:
            logger.warning(
                "MCPMountManager has no zone bound (mount=%r); "
                "skipping approval-queue routing and honoring SSRF block",
                mount_config.name,
            )
            return False

        # Lazy imports keep ``mcp`` brick free of an eager top-level
        # cross-brick import (boundary checker enforces this; see
        # `.pre-commit-hooks/check_brick_imports.py`).
        from nexus.bricks.approvals.models import ApprovalKind, Decision

        zone_id = self._zone_id
        # Mount-time admin operations have no agent token. Synthesize a
        # stable token_id from the mount name so operators can correlate
        # repeated requests for the same mount in the queue UI.
        token_id = f"mcp_mount:{mount_config.name}" if mount_config.name else "mcp_mount"
        try:
            decision = await gate.check(
                kind=ApprovalKind.EGRESS_HOST,
                subject=host_port,
                zone_id=zone_id,
                token_id=token_id,
                session_id=None,
                agent_id=mount_config.name or None,
                reason=operation,
                metadata={
                    "url": url,
                    "mount_name": mount_config.name,
                    "operation": operation,
                },
            )
        except Exception:
            logger.warning(
                "approvals gate raised for mount %r (%s); falling back to deny",
                mount_config.name,
                operation,
                exc_info=True,
            )
            return False
        return decision is Decision.APPROVED

    def _emit_ssrf_audit(self, mount_config: MCPMount, exc: SSRFBlocked) -> None:
        """Emit a security.ssrf_blocked event; never raises."""
        try:
            emit_audit_event(
                "security.ssrf_blocked",
                {
                    "url": exc.url,
                    "reason": exc.reason,
                    "ip": exc.ip,
                    "cidr": exc.cidr,
                    "integration": "mcp",
                    "mount_name": mount_config.name,
                },
            )
        except Exception as audit_err:  # pragma: no cover — audit must never raise
            logger.warning("Failed to emit SSRF audit event: %s", audit_err)

    async def unmount(self, name: str) -> bool:
        """Unmount an MCP server.

        Args:
            name: Mount name

        Returns:
            True if unmount successful

        Raises:
            MCPMountError: If unmount fails
        """
        if name not in self._mounts:
            raise MCPMountError(f"Mount not found: {name}")

        # Close client connection if active
        if name in self._clients:
            try:
                client = self._clients[name]
                if hasattr(client, "close"):
                    await client.close()
            except Exception as e:
                logger.warning("Error closing client for %s: %s", name, e)
            finally:
                del self._clients[name]

        # Update mount state
        mount = self._mounts[name]
        mount.mounted = False

        await self._save_mounts_config()

        logger.info("Unmounted MCP server: %s", name)
        return True

    async def sync_tools(self, name: str) -> int:
        """Sync tools from a mounted MCP server.

        Discovers available tools via list_tools() and stores
        tool definitions in the skills filesystem.

        Args:
            name: Mount name

        Returns:
            Number of tools discovered

        Raises:
            MCPMountError: If sync fails
        """
        if name not in self._mounts:
            raise MCPMountError(f"Mount not found: {name}")

        mount = self._mounts[name]

        # Get tools from MCP server
        try:
            tools = await self._list_tools_from_server(mount)
        except Exception as e:
            raise MCPMountError(f"Failed to list tools from {name}: {e}") from e

        # Create tool definitions and store in skills filesystem
        tool_count = 0
        tool_names = []
        tool_defs = []

        for tool in tools:
            try:
                tool_def = self._create_tool_definition(tool, mount)
                await self._store_tool_definition(tool_def, mount)
                tool_count += 1
                tool_names.append(tool_def.name)
                tool_defs.append(tool_def)
            except Exception as e:
                logger.warning("Failed to store tool %s: %s", tool.get("name"), e)

        # Store single README.md for the mount
        if tool_defs:
            await self._store_mount_readme_md(mount, tool_defs)

        # Update mount state
        mount.last_sync = datetime.now(UTC)
        mount.tool_count = tool_count
        mount.tools = tool_names
        await self._save_mounts_config()

        logger.info("Synced %d tools from %s", tool_count, name)
        return tool_count

    async def _list_tools_from_server(self, mount: MCPMount) -> list[dict[str, Any]]:
        """List tools from an MCP server.

        Args:
            mount: Mount configuration

        Returns:
            List of tool definitions from the server
        """
        # If we have an active client, use it
        if mount.name in self._clients:
            client = self._clients[mount.name]
            if hasattr(client, "list_tools"):
                result = await client.list_tools()
                return [
                    {
                        "name": tool.name,
                        "description": tool.description or "",
                        "inputSchema": tool.inputSchema if hasattr(tool, "inputSchema") else {},
                    }
                    for tool in result.tools
                ]

        # Otherwise, spawn a temporary process to list tools
        if mount.transport == "stdio":
            return await self._list_tools_stdio(mount)
        elif mount.transport in ("sse", "http"):
            return await self._list_tools_sse(mount)
        elif mount.transport == "klavis_rest":
            return await self._list_tools_klavis(mount)

        return []

    async def _list_tools_stdio(self, mount: MCPMount) -> list[dict[str, Any]]:
        """List tools using stdio transport.

        Args:
            mount: Mount configuration

        Returns:
            List of tool definitions
        """
        try:
            from mcp import ClientSession, StdioServerParameters
            from mcp.client.stdio import stdio_client
        except ImportError as e:
            raise MCPMountError(
                "MCP client library not installed. Install with: pip install mcp"
            ) from e

        server_params = StdioServerParameters(
            command=mount.command or "",
            args=mount.args,
            env=mount.env or None,
        )

        tools = []

        try:
            async with (
                stdio_client(server_params) as (read, write),
                ClientSession(read, write) as session,
            ):
                await session.initialize()
                result = await session.list_tools()
                tools = [
                    {
                        "name": tool.name,
                        "description": tool.description or "",
                        "inputSchema": (tool.inputSchema if hasattr(tool, "inputSchema") else {}),
                    }
                    for tool in result.tools
                ]
        except Exception as e:
            logger.error("Failed to list tools via stdio: %s", e)
            raise

        return tools

    async def _list_tools_sse(self, mount: MCPMount) -> list[dict[str, Any]]:
        """List tools using SSE/HTTP transport.

        Args:
            mount: Mount configuration

        Returns:
            List of tool definitions
        """
        try:
            from mcp import ClientSession
            from mcp.client.sse import sse_client
        except ImportError as e:
            raise MCPMountError(
                "MCP client library not installed. Install with: pip install mcp"
            ) from e

        if not mount.url:
            raise MCPMountError("URL is required for SSE/HTTP transport")

        ssrf_cfg = self._ssrf_config()
        try:
            validated = validate_outbound_url(
                mount.url,
                allow_private=ssrf_cfg.allow_private,
                extra_deny_cidrs=ssrf_cfg.extra_deny_cidrs,
            )
        except SSRFBlocked as exc:
            self._emit_ssrf_audit(mount, exc)
            logger.warning("SSRF blocked for MCP mount %r (list_tools): %s", mount.name, exc)
            if not await self._ssrf_blocked_via_gate(mount, mount.url, "mcp_mount_list_tools"):
                raise
            # Preserve operator-configured ``extra_deny_cidrs`` on
            # approval re-validation — see ``_create_sse_client`` for
            # rationale.
            try:
                validated = validate_outbound_url(
                    mount.url,
                    allow_private=True,
                    extra_deny_cidrs=ssrf_cfg.extra_deny_cidrs,
                )
            except SSRFBlocked:
                logger.warning(
                    "approvals-approved mount %r still failed SSRF re-validation "
                    "(list_tools); honoring the original block",
                    mount.name,
                )
                raise exc from None

        # Start with custom headers from mount config
        headers: dict[str, str] = dict(mount.headers) if mount.headers else {}

        # Add authentication headers if needed
        if mount.auth_type == "api_key" and mount.auth_config:
            api_key = mount.auth_config.get("api_key")
            header_name = mount.auth_config.get("header_name", "Authorization")
            if api_key:
                headers[header_name] = f"Bearer {api_key}"

        # Cast to Any: see _create_sse_client for rationale.
        pinned_factory = cast(Any, make_pinned_client_factory(validated))

        tools = []

        try:
            async with (
                sse_client(
                    mount.url,
                    headers=headers,
                    httpx_client_factory=pinned_factory,
                ) as (read, write),
                ClientSession(read, write) as session,
            ):
                await session.initialize()
                result = await session.list_tools()
                tools = [
                    {
                        "name": tool.name,
                        "description": tool.description or "",
                        "inputSchema": (tool.inputSchema if hasattr(tool, "inputSchema") else {}),
                    }
                    for tool in result.tools
                ]
        except Exception as e:
            logger.error("Failed to list tools via SSE: %s", e)
            raise

        return tools

    async def _list_tools_klavis(self, mount: MCPMount) -> list[dict[str, Any]]:
        """List tools using Klavis REST API.

        Klavis uses a REST API instead of direct MCP SSE connection.
        Tools are listed via POST /mcp-server/list-tools.

        Args:
            mount: Mount configuration with klavis_rest transport

        Returns:
            List of tool definitions
        """
        try:
            import httpx
        except ImportError as e:
            raise MCPMountError(
                "httpx library not installed. Install with: pip install httpx"
            ) from e

        if not mount.url:
            raise MCPMountError("URL (Strata server URL) is required for Klavis transport")

        # Get API key from mount config or environment
        import os

        api_key = mount.klavis_api_key or os.getenv("KLAVIS_API_KEY")
        if not api_key:
            raise MCPMountError(
                "Klavis API key not configured. Set klavis_api_key in mount config "
                "or KLAVIS_API_KEY environment variable."
            )

        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

        tools = []

        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    "https://api.klavis.ai/mcp-server/list-tools",
                    json={"serverUrl": mount.url},
                    headers=headers,
                )
                response.raise_for_status()
                data = response.json()

                if not data.get("success"):
                    raise MCPMountError(f"Klavis API error: {data.get('error')}")

                tools = [
                    {
                        "name": tool.get("name", ""),
                        "description": tool.get("description", ""),
                        "inputSchema": tool.get("inputSchema", {}),
                    }
                    for tool in data.get("tools", [])
                ]

        except httpx.HTTPStatusError as e:
            logger.error("Klavis API HTTP error: %s", e)
            raise MCPMountError(f"Klavis API error: {e}") from e
        except Exception as e:
            logger.error("Failed to list tools via Klavis: %s", e)
            raise

        return tools

    async def call_tool_klavis(
        self,
        mount: MCPMount,
        tool_name: str,
        tool_args: dict[str, Any],
    ) -> dict[str, Any]:
        """Call a tool via Klavis REST API.

        This is the unified method for calling tools on Klavis-hosted MCP servers.

        Args:
            mount: Mount configuration with klavis_rest transport
            tool_name: Name of the tool to call
            tool_args: Arguments to pass to the tool

        Returns:
            Tool execution result

        Raises:
            MCPMountError: If tool call fails
        """
        try:
            import httpx
        except ImportError as e:
            raise MCPMountError(
                "httpx library not installed. Install with: pip install httpx"
            ) from e

        if not mount.url:
            raise MCPMountError("URL (Strata server URL) is required for Klavis transport")

        # Get API key from mount config or environment
        import os

        api_key = mount.klavis_api_key or os.getenv("KLAVIS_API_KEY")
        if not api_key:
            raise MCPMountError("Klavis API key not configured")

        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

        try:
            async with httpx.AsyncClient(timeout=60.0) as client:
                response = await client.post(
                    "https://api.klavis.ai/mcp-server/call-tool",
                    json={
                        "serverUrl": mount.url,
                        "toolName": tool_name,
                        "toolArgs": tool_args,
                        "connectionType": mount.klavis_connection_type or "StreamableHttp",
                    },
                    headers=headers,
                )
                response.raise_for_status()
                data = response.json()

                if not data.get("success"):
                    raise MCPMountError(f"Klavis tool call failed: {data.get('error')}")

                call_result: dict[str, Any] = data.get("result", {})
                return call_result

        except httpx.HTTPStatusError as e:
            logger.error("Klavis API HTTP error: %s", e)
            raise MCPMountError(f"Klavis API error: {e}") from e
        except Exception as e:
            logger.error("Failed to call tool via Klavis: %s", e)
            raise MCPMountError(f"Klavis tool call failed: {e}") from e

    def _create_tool_definition(
        self, tool_data: dict[str, Any], mount: MCPMount
    ) -> MCPToolDefinition:
        """Create a tool definition from MCP tool data.

        Args:
            tool_data: Tool data from MCP server
            mount: Mount configuration

        Returns:
            MCPToolDefinition instance
        """
        name = tool_data.get("name", "")
        description = tool_data.get("description", "")
        input_schema = tool_data.get("inputSchema", {})

        # Create MCP config
        mcp_config = MCPToolConfig(
            endpoint=f"mcp://{mount.name}/{name}",
            input_schema=input_schema,
            requires_mount=True,
            mount_name=mount.name,
            when_to_use=description,
        )

        now = datetime.now(UTC)

        return MCPToolDefinition(
            name=name,
            description=description,
            version="1.0.0",
            skill_type="mcp_tool",
            mcp_config=mcp_config,
            created_at=now,
            modified_at=now,
        )

    async def _store_tool_definition(self, tool_def: MCPToolDefinition, mount: MCPMount) -> str:
        """Store a tool definition in the skills filesystem.

        Stores tool.json directly in the mount's tools directory.

        Args:
            tool_def: Tool definition
            mount: Mount configuration

        Returns:
            Path to stored tool
        """
        # Store tool.json directly: /skills/system/mcp-tools/github/search_repositories.json
        tool_json_path = f"{mount.tools_path}{tool_def.name}.json"

        # Create tool.json
        tool_json = json.dumps(tool_def.to_dict(), indent=2)

        if self._filesystem:
            # Ensure directory exists
            if mount.tools_path:
                try:
                    self._filesystem.mkdir(mount.tools_path, parents=True)
                except FileExistsError:
                    pass
                except OSError as e:
                    logger.warning("Failed to create directory %s: %s", mount.tools_path, e)

            # Write tool.json
            self._filesystem.write(tool_json_path, tool_json.encode("utf-8"))
        else:
            # Local filesystem
            if mount.tools_path:
                tools_dir = Path(mount.tools_path.lstrip("/"))
                tools_dir.mkdir(parents=True, exist_ok=True)
                (tools_dir / f"{tool_def.name}.json").write_text(tool_json)

        logger.debug("Stored tool definition: %s", tool_def.name)
        return tool_json_path

    async def _store_mount_readme_md(
        self, mount: MCPMount, tool_defs: list[MCPToolDefinition]
    ) -> str:
        """Store a single README.md for the mount describing all tools.

        Args:
            mount: Mount configuration
            tool_defs: List of tool definitions

        Returns:
            Path to README.md
        """
        readme_md_path = f"{mount.tools_path}README.md"
        readme_md = self._generate_mount_readme_md(mount, tool_defs)

        if self._filesystem:
            if mount.tools_path:
                try:
                    self._filesystem.mkdir(mount.tools_path, parents=True)
                except FileExistsError:
                    pass
                except OSError as e:
                    logger.warning("Failed to create directory %s: %s", mount.tools_path, e)
            self._filesystem.write(readme_md_path, readme_md.encode("utf-8"))
        else:
            if mount.tools_path:
                tools_dir = Path(mount.tools_path.lstrip("/"))
                tools_dir.mkdir(parents=True, exist_ok=True)
                (tools_dir / "README.md").write_text(readme_md)

        logger.debug("Stored mount README.md: %s", mount.name)
        return readme_md_path

    def _generate_mount_readme_md(self, mount: MCPMount, tool_defs: list[MCPToolDefinition]) -> str:
        """Generate README.md content for a mount with all its tools.

        Args:
            mount: Mount configuration
            tool_defs: List of tool definitions

        Returns:
            README.md content
        """
        import yaml

        # Build frontmatter
        frontmatter: dict[str, Any] = {
            "name": mount.name,
            "description": mount.description,
            "version": "1.0.0",
            "skill_type": "mcp_tools",
            "tool_count": len(tool_defs),
            "transport": mount.transport,
        }

        if mount.command:
            frontmatter["command"] = mount.command
        if mount.url:
            frontmatter["url"] = mount.url

        frontmatter_yaml = yaml.dump(frontmatter, default_flow_style=False, sort_keys=False)

        # Build markdown content
        content_parts = [
            f"# {mount.name}",
            "",
            mount.description,
            "",
            f"**Tools:** {len(tool_defs)}",
            f"**Transport:** {mount.transport}",
            "",
            "## Tools",
            "",
        ]

        # Add each tool
        for tool_def in sorted(tool_defs, key=lambda t: t.name):
            content_parts.append(f"### {tool_def.name}")
            content_parts.append("")
            content_parts.append(tool_def.description)
            content_parts.append("")

            # Add input parameters summary
            if tool_def.mcp_config and tool_def.mcp_config.input_schema:
                schema = tool_def.mcp_config.input_schema
                props = schema.get("properties", {})
                required = schema.get("required", [])

                if props:
                    content_parts.append("**Parameters:**")
                    for param_name, param_info in props.items():
                        param_type = param_info.get("type", "any")
                        param_desc = param_info.get("description", "")
                        req_mark = " *(required)*" if param_name in required else ""
                        content_parts.append(
                            f"- `{param_name}` ({param_type}){req_mark}: {param_desc}"
                        )
                    content_parts.append("")

        content = "\n".join(content_parts)

        return f"---\n{frontmatter_yaml}---\n\n{content}"

    async def execute_tool(self, mount_name: str, tool_name: str, args: dict[str, Any]) -> Any:
        """Execute a tool from a mounted MCP server.

        Automatically routes to the correct transport:
        - stdio/sse/http: Use MCP client
        - klavis_rest: Use Klavis REST API

        Args:
            mount_name: Mount name
            tool_name: Tool name
            args: Tool arguments

        Returns:
            Tool execution result

        Raises:
            MCPMountError: If execution fails
        """
        if mount_name not in self._mounts:
            raise MCPMountError(f"Mount not found: {mount_name}")

        mount = self._mounts[mount_name]

        # For Klavis REST transport, use the Klavis API directly
        if mount.transport == "klavis_rest":
            return await self.call_tool_klavis(mount, tool_name, args)

        # For standard MCP transports, require active client
        if not mount.mounted:
            raise MCPMountError(f"Mount {mount_name} is not active. Call mount() first.")

        if mount_name not in self._clients:
            raise MCPMountError(f"No active client for {mount_name}")

        client = self._clients[mount_name]

        try:
            result = await client.call_tool(tool_name, args)
            return result
        except Exception as e:
            raise MCPMountError(f"Tool execution failed: {e}") from e

    async def discover_mounts(self, context: "OperationContext | None" = None) -> int:
        """Discover mounts from context-aware tier paths.

        Scans all available tiers for the given context and loads mount configurations.
        Uses tier priority: user (3) > zone (2) > system (1).
        When same mount name exists at multiple tiers, higher priority wins.

        Args:
            context: Operation context with user_id, zone_id

        Returns:
            Number of mounts discovered
        """
        tier_paths = self.get_mcp_tier_paths(context)
        discovered_count = 0

        # Clear existing index (but keep _mounts for active connections)
        self._tier_index = {}

        # Track which names we've seen at which priority
        seen_names: dict[str, int] = {}  # name -> priority

        # Discover from each tier (in priority order - higher first)
        for tier in sorted(
            tier_paths.keys(), key=lambda t: self.TIER_PRIORITY.get(t, 0), reverse=True
        ):
            tier_path = tier_paths[tier]
            count = await self._discover_mounts_from_tier(tier, tier_path, seen_names)
            discovered_count += count

        logger.info("Discovered %d mounts from %d tiers", discovered_count, len(tier_paths))
        return discovered_count

    async def _discover_mounts_from_tier(
        self, tier: str, tier_path: str, seen_names: dict[str, int]
    ) -> int:
        """Discover mounts from a single tier path.

        Args:
            tier: Tier name (user, zone, system)
            tier_path: Path to mcp-tools directory for this tier
            seen_names: Dict of mount names already seen with their priorities

        Returns:
            Number of mounts discovered from this tier
        """
        count = 0
        tier_priority = self.TIER_PRIORITY.get(tier, 0)

        if tier not in self._tier_index:
            self._tier_index[tier] = []

        try:
            if self._filesystem:
                # Check if path exists
                if not self._filesystem.access(tier_path):
                    logger.debug("MCP tier path does not exist: %s", tier_path)
                    return 0

                # List directories in tier_path
                items = await asyncio.to_thread(self._filesystem.sys_readdir, tier_path)
                for item in items:
                    item_path = f"{tier_path}{item}"
                    mount_json_path = f"{item_path}/mount.json"

                    try:
                        if self._filesystem.access(mount_json_path):
                            raw_content = await asyncio.to_thread(
                                self._filesystem.sys_read, mount_json_path
                            )
                            content_str = (
                                raw_content.decode("utf-8")
                                if isinstance(raw_content, bytes)
                                else str(raw_content)
                            )
                            mount_data = json.loads(content_str)
                            mount = MCPMount.from_dict(mount_data)
                            mount.mounted = False

                            # Check if we've seen this name at a higher priority
                            if mount.name in seen_names:
                                existing_priority = seen_names[mount.name]
                                if tier_priority <= existing_priority:
                                    logger.debug(
                                        "Skipping mount '%s' from %s "
                                        "(already loaded from higher priority tier)",
                                        mount.name,
                                        tier,
                                    )
                                    continue

                            # Store mount with tier info
                            mount.tier = tier
                            self._mounts[mount.name] = mount
                            self._tier_index[tier].append(mount.name)
                            seen_names[mount.name] = tier_priority
                            count += 1
                            logger.debug(
                                "Discovered mount '%s' from %s: %s",
                                mount.name,
                                tier,
                                mount_json_path,
                            )

                    except Exception as e:
                        logger.warning("Failed to load mount from %s: %s", mount_json_path, e)

            else:
                # Local filesystem
                base_path = Path(tier_path.lstrip("/"))
                if not base_path.exists():
                    logger.debug("MCP tier path does not exist: %s", tier_path)
                    return 0

                for mount_dir in base_path.iterdir():
                    if mount_dir.is_dir():
                        mount_json_file = mount_dir / "mount.json"
                        if mount_json_file.exists():
                            try:
                                mount_data = json.loads(mount_json_file.read_text())
                                mount = MCPMount.from_dict(mount_data)
                                mount.mounted = False

                                # Check priority
                                if mount.name in seen_names:
                                    existing_priority = seen_names[mount.name]
                                    if tier_priority <= existing_priority:
                                        logger.debug(
                                            "Skipping mount '%s' from %s "
                                            "(already loaded from higher priority tier)",
                                            mount.name,
                                            tier,
                                        )
                                        continue

                                mount.tier = tier
                                self._mounts[mount.name] = mount
                                self._tier_index[tier].append(mount.name)
                                seen_names[mount.name] = tier_priority
                                count += 1
                                logger.debug(
                                    "Discovered mount '%s' from %s: %s",
                                    mount.name,
                                    tier,
                                    mount_json_file,
                                )

                            except Exception as e:
                                logger.warning(
                                    "Failed to load mount from %s: %s", mount_json_file, e
                                )

        except Exception as e:
            logger.warning("Error scanning MCP tier %s at %s: %s", tier, tier_path, e)

        return count

    async def list_mounts(
        self,
        include_unmounted: bool = True,
        tier: str | None = None,
        context: "OperationContext | None" = None,
    ) -> list[MCPMount]:
        """List mount configurations.

        If context is provided, discovers mounts from context-aware paths first.
        Optionally filter by tier.

        Args:
            include_unmounted: Include unmounted configurations
            tier: Optional tier filter (user, zone, system)
            context: Optional operation context for discovery

        Returns:
            List of MCPMount configurations
        """
        # If context provided, re-discover mounts for that context
        if context:
            await self.discover_mounts(context)

        # Filter by tier if specified
        if tier:
            mount_names = self._tier_index.get(tier, [])
            mounts = [self._mounts[name] for name in mount_names if name in self._mounts]
        else:
            mounts = list(self._mounts.values())

        # Filter by mounted status
        if not include_unmounted:
            mounts = [m for m in mounts if m.mounted]

        return mounts

    async def get_mount(
        self,
        name: str,
        context: "OperationContext | None" = None,
    ) -> MCPMount | None:
        """Get mount configuration by name.

        If context is provided, discovers mounts from context-aware paths first.
        Returns the highest priority mount matching the name.

        Args:
            name: Mount name
            context: Optional operation context for discovery

        Returns:
            MCPMount or None if not found
        """
        # If context provided, re-discover mounts for that context
        if context:
            await self.discover_mounts(context)

        return self._mounts.get(name)

    async def remove_mount(self, name: str) -> bool:
        """Remove a mount configuration.

        Args:
            name: Mount name

        Returns:
            True if removed
        """
        if name in self._mounts:
            # Ensure unmounted
            if name in self._clients:
                # Can't remove active mount
                return False

            del self._mounts[name]
            await self._save_mounts_config()
            return True

        return False

    async def add_mount_config(self, mount_config: MCPMount) -> None:
        """Add a mount configuration without connecting.

        Useful for pre-configuring mounts that will be connected later.

        Args:
            mount_config: Mount configuration
        """
        # Set tools path
        mount_config.tools_path = f"{self.MCP_TOOLS_PATH}{mount_config.name}/"
        mount_config.mounted = False

        self._mounts[mount_config.name] = mount_config
        await self._save_mounts_config()

        logger.info("Added mount configuration: %s", mount_config.name)
