butler/mcp/config.py
Rama main · réplica de lectura del repositorio original.
"""Load and validate MCP server configuration."""
from __future__ import annotations
import logging
import os
import re
import ipaddress
from pathlib import Path
from typing import Any, cast
from butler.utilities.env_parse import env_truthy, int_env
from butler.io.safe_load import safe_load_yaml
from butler.mcp.types import McpServerConfig, McpToolPolicy
from butler.defaults.env_defaults import (
MCP_CONFIG_DEFAULT,
MCP_HTTP_HOSTS_ALLOW_DEFAULT,
MCP_STDIO_ALLOW_COMMANDS_DEFAULT,
)
logger = logging.getLogger(__name__)
_PRIVATE_HOSTS = frozenset({
"localhost",
"127.0.0.1",
"0.0.0.0",
"::1",
"metadata.google.internal",
})
def mcp_sdk_available() -> bool:
try:
import mcp # noqa: F401
return True
except ImportError:
return False
def mcp_enabled() -> bool:
return env_truthy("BUTLER_MCP_ENABLED", default=False) and mcp_sdk_available()
def max_servers() -> int:
try:
return int(int_env("BUTLER_MCP_MAX_SERVERS", 3, min=0, max=20))
except ValueError:
return 3
def max_tools() -> int:
try:
return int(int_env("BUTLER_MCP_MAX_TOOLS", 20, min=0, max=100))
except ValueError:
return 20
def session_scoped() -> bool:
return bool(env_truthy("BUTLER_MCP_SESSION_SCOPED", default=True))
def stdio_allow_commands() -> frozenset[str]:
raw = os.getenv("BUTLER_MCP_STDIO_ALLOW_COMMANDS", MCP_STDIO_ALLOW_COMMANDS_DEFAULT)
return frozenset(x.strip() for x in raw.split(",") if x.strip())
def http_hosts_allow_extra() -> list[str]:
raw = os.getenv("BUTLER_MCP_HTTP_HOSTS_ALLOW", MCP_HTTP_HOSTS_ALLOW_DEFAULT).strip()
if not raw:
return []
return [x.strip() for x in raw.split(",") if x.strip()]
def allow_private_http() -> bool:
return bool(env_truthy("BUTLER_MCP_HTTP_ALLOW_PRIVATE", default=False))
def _expand_env(value: str) -> str:
import re as _re
def repl(m: _re.Match[str]) -> str:
key = m.group(1)
return os.getenv(key, "")
return _re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", repl, str(value or ""))
def _resolve_config_paths(workspace: Path | None = None) -> list[Path]:
paths: list[Path] = []
if workspace is not None:
for rel in (".butler/mcp.yaml", ".butler/mcp.yml"):
p = workspace / rel
if p.is_file():
paths.append(p)
env_path = os.getenv("BUTLER_MCP_CONFIG", MCP_CONFIG_DEFAULT).strip()
if env_path:
paths.append(Path(env_path).expanduser())
else:
from butler.mcp.config_ops import butler_home_mcp_config_paths_safe
paths.extend(butler_home_mcp_config_paths_safe())
seen: set[str] = set()
out: list[Path] = []
for p in paths:
key = str(p.resolve())
if key not in seen:
seen.add(key)
out.append(p)
return out
def _parse_tool_policy(raw: Any) -> McpToolPolicy:
if not isinstance(raw, dict):
return McpToolPolicy()
allow = raw.get("allow") or []
deny = raw.get("deny") or []
if not isinstance(allow, list):
allow = []
if not isinstance(deny, list):
deny = []
return McpToolPolicy(
allow=tuple(str(x).strip() for x in allow if str(x).strip()),
deny=tuple(str(x).strip() for x in deny if str(x).strip()),
)
def _parse_server(server_id: str, raw: dict[str, Any]) -> McpServerConfig | None:
sid = re.sub(r"[^a-zA-Z0-9._-]+", "_", str(server_id or "").strip())[:64]
if not sid:
return None
transport = str(raw.get("transport") or "stdio").strip().lower()
if transport not in ("stdio", "http"):
logger.warning("MCP server %s: invalid transport %s", sid, transport)
return None
timeout = float(raw.get("timeout_seconds") or raw.get("timeout") or 60)
tools = _parse_tool_policy(raw.get("tools"))
classify_raw = raw.get("classify")
classify: dict[str, str] = {}
if isinstance(classify_raw, dict):
classify = {str(k): str(v) for k, v in classify_raw.items()}
if transport == "stdio":
command = str(raw.get("command") or "").strip()
if not command:
return None
args_raw = raw.get("args") or []
args = tuple(str(a) for a in args_raw) if isinstance(args_raw, list) else ()
env_raw = raw.get("env") or {}
env = (
{_expand_env(str(k)): _expand_env(str(v)) for k, v in env_raw.items()}
if isinstance(env_raw, dict)
else {}
)
return McpServerConfig(
server_id=sid,
transport="stdio",
timeout_seconds=timeout,
command=command,
args=args,
env=env,
cwd=str(raw.get("cwd") or "").strip(),
tools=tools,
classify=classify,
)
url = _expand_env(str(raw.get("url") or "").strip())
if not url:
return None
headers_raw = raw.get("headers") or {}
headers = (
{_expand_env(str(k)): _expand_env(str(v)) for k, v in headers_raw.items()}
if isinstance(headers_raw, dict)
else {}
)
hosts_raw = raw.get("hosts_allow") or []
hosts = tuple(str(h).strip() for h in hosts_raw) if isinstance(hosts_raw, list) else ()
return McpServerConfig(
server_id=sid,
transport="http",
timeout_seconds=timeout,
url=url,
headers=headers,
sse=bool(raw.get("sse")),
hosts_allow=hosts,
tools=tools,
classify=classify,
)
def http_mcp_servers_configured(*, workspace: Path | None = None) -> bool:
"""True if any loaded MCP server uses HTTP transport."""
from butler.mcp.config_ops import http_mcp_servers_configured_safe
return bool(http_mcp_servers_configured_safe(workspace=workspace))
def load_mcp_servers(*, workspace: Path | None = None) -> list[McpServerConfig]:
servers: dict[str, McpServerConfig] = {}
for path in _resolve_config_paths(workspace):
# Audit R2-19: corrupt MCP config used to silently skip
# every server in that file. Now the safe_load helper
# renames the corrupt file for forensic retention, logs
# WARNING with exc_info, and records the event for /诊断.
data = safe_load_yaml(path, default={}, kind="mcp_servers")
if not isinstance(data, dict):
continue
block = data.get("servers")
if not isinstance(block, dict):
continue
for sid, raw in block.items():
if not isinstance(raw, dict):
continue
cfg = _parse_server(str(sid), raw)
if cfg is not None:
servers[cfg.server_id] = cfg
ordered = list(servers.values())[: max_servers()]
return ordered
def validate_stdio_command(config: McpServerConfig) -> str | None:
cmd = Path(config.command).name if "/" in config.command or "\\" in config.command else config.command
allowed = stdio_allow_commands()
if cmd not in allowed:
return f"stdio command '{cmd}' not in BUTLER_MCP_STDIO_ALLOW_COMMANDS"
return None
def _host_is_private_or_metadata(host: str) -> bool:
norm = str(host or "").strip().lower()
if not norm:
return True
if norm in _PRIVATE_HOSTS:
return True
if norm.endswith(".local") or norm.endswith(".internal"):
return True
try:
addr = ipaddress.ip_address(norm)
except ValueError:
return False
return bool(
addr.is_private
or addr.is_loopback
or addr.is_link_local
or addr.is_reserved
or addr.is_multicast
)
def validate_http_url(config: McpServerConfig) -> str | None:
from butler.mcp.config_ops import parse_http_url_safe
parsed, err = parse_http_url_safe(config.url)
if err:
return cast(str | None, err)
if parsed is None:
return "invalid url"
if parsed.scheme not in ("http", "https"):
return f"unsupported scheme: {parsed.scheme}"
host = (parsed.hostname or "").lower()
if not host:
return "missing host"
if not allow_private_http() and _host_is_private_or_metadata(host):
return f"private host blocked: {host}"
allowed = set(config.hosts_allow) | set(http_hosts_allow_extra())
if allowed and not any(host == h or host.endswith("." + h) for h in allowed):
return f"host '{host}' not in hosts_allow"
return None
def tool_allowed_by_policy(policy: McpToolPolicy, tool_name: str) -> bool:
name = str(tool_name or "").strip()
if not name:
return False
if policy.deny and name in policy.deny:
return False
if policy.allow:
return name in policy.allow
return True