butler/mcp/client_stdio.py

Rama main · réplica de lectura del repositorio original.

"""Stdio MCP client transport."""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any, cast

from butler.mcp.config import validate_stdio_command
from butler.mcp.types import McpServerConfig

logger = logging.getLogger(__name__)


_PROTECTED_ENV_KEYS = frozenset({
    "PATH", "HOME", "USER", "SHELL", "LOGNAME", "LANG",
    "LD_PRELOAD", "LD_LIBRARY_PATH", "PYTHONPATH",
})


def _merge_stdio_path(host_path: str) -> str:
    """Ensure stdio MCP subprocesses can find ``sh`` and allowlisted commands.

    Gateway systemd may load ``.env`` with ``PATH=...:$PATH`` where ``$PATH`` is
    literal and ``/bin`` is missing — ``npx`` then fails with ``spawn sh ENOENT``.
    """
    required = (
        "/usr/local/sbin",
        "/usr/local/bin",
        "/usr/sbin",
        "/usr/bin",
        "/sbin",
        "/bin",
    )
    parts: list[str] = []
    seen: set[str] = set()
    for segment in str(host_path or "").split(":"):
        s = segment.strip()
        if not s or s == "$PATH":
            continue
        if s not in seen:
            parts.append(s)
            seen.add(s)
    for req in required:
        if req not in seen:
            parts.append(req)
            seen.add(req)
    nvm = [p for p in parts if "/.nvm/versions/node/" in p]
    if nvm:
        other = [p for p in parts if p not in nvm]
        return ":".join(nvm + other)
    return ":".join(parts)


def _build_stdio_env(config: McpServerConfig) -> dict[str, str]:
    import os

    from butler.tools.path_safety import safe_subprocess_env

    base = safe_subprocess_env()
    host_path = os.environ.get("PATH", "").strip()
    base["PATH"] = _merge_stdio_path(host_path) if host_path else base["PATH"]
    for key, value in (config.env or {}).items():
        if value is not None:
            k = str(key)
            if k.upper() in _PROTECTED_ENV_KEYS:
                logger.warning("MCP env override blocked for protected key: %s", k)
                continue
            base[k] = str(value)
    return cast(dict[str, str], base)


def _resolve_cwd(config: McpServerConfig, workspace: Path | None) -> str | None:
    if config.cwd:
        return str(Path(config.cwd).expanduser())
    if workspace is not None and workspace.is_dir():
        return str(workspace)
    return None


async def connect_stdio(
    config: McpServerConfig,
    *,
    workspace: Path | None = None,
) -> tuple[Any, list[Any], Any]:
    """Return (session, tools, cleanup_async_callable)."""
    err = validate_stdio_command(config)
    if err:
        raise ValueError(err)

    from mcp import ClientSession
    from mcp.client.stdio import StdioServerParameters, stdio_client

    params = StdioServerParameters(
        command=config.command,
        args=list(config.args),
        env=_build_stdio_env(config),
        cwd=_resolve_cwd(config, workspace),
    )
    transport = stdio_client(params)
    read_stream, write_stream = await transport.__aenter__()
    session = ClientSession(read_stream, write_stream)
    await session.__aenter__()
    await session.initialize()
    listed = await session.list_tools()
    tools = list(getattr(listed, "tools", None) or [])

    async def cleanup() -> None:
        from butler.mcp.client_stdio_ops import (
            close_stdio_session_safe,
            close_stdio_transport_safe,
        )

        await close_stdio_session_safe(session)
        await close_stdio_transport_safe(transport)

    return session, tools, cleanup


async def call_stdio_tool(session: Any, tool_name: str, arguments: dict[str, Any]) -> str:
    result = await session.call_tool(tool_name, arguments)
    parts: list[str] = []
    for content in getattr(result, "content", None) or []:
        text = getattr(content, "text", None)
        parts.append(str(text) if text is not None else str(content))
    return "\n".join(parts) if parts else json_dumps_result(result)


def json_dumps_result(result: Any) -> str:
    from butler.mcp.client_stdio_ops import json_dumps_mcp_result_safe

    return cast(str, json_dumps_mcp_result_safe(result))