From dee48d8cea136c1407192bbda4f0f201e74d1b84 Mon Sep 17 00:00:00 2001 From: rob Date: Mon, 20 Jul 2026 00:20:40 -0300 Subject: [PATCH] Add M7.2 MCP client support with McpStep, CLI commands, and stdio transport --- pyproject.toml | 4 + src/cmdforge/cli/__init__.py | 48 +++ src/cmdforge/cli/mcp_commands.py | 169 +++++++++++ src/cmdforge/mcp_client.py | 484 ++++++++++++++++++++++++++++++ src/cmdforge/runner.py | 79 ++++- src/cmdforge/tool.py | 55 +++- tests/fixtures/mcp_test_server.py | 44 +++ tests/test_mcp.py | 399 ++++++++++++++++++++++++ 8 files changed, 1279 insertions(+), 3 deletions(-) create mode 100644 src/cmdforge/cli/mcp_commands.py create mode 100644 src/cmdforge/mcp_client.py create mode 100644 tests/fixtures/mcp_test_server.py create mode 100644 tests/test_mcp.py diff --git a/pyproject.toml b/pyproject.toml index beb64df..7fc38eb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,6 +57,9 @@ flow = [ pty = [ "pexpect>=4.8", # For PTY providers (interactive CLIs like Aider) ] +mcp = [ + "mcp>=1.27,<2", # Model Context Protocol support (McpStep) +] all = [ "Flask>=2.3", "argon2-cffi>=21.0", @@ -65,6 +68,7 @@ all = [ "NodeGraphQt-QuiltiX-fork[pyside6]>=0.7.0", "setuptools", "pexpect>=4.8", + "mcp>=1.27,<2", ] [project.scripts] diff --git a/src/cmdforge/cli/__init__.py b/src/cmdforge/cli/__init__.py index 4d87e3c..35580a1 100644 --- a/src/cmdforge/cli/__init__.py +++ b/src/cmdforge/cli/__init__.py @@ -16,6 +16,7 @@ from .project_commands import cmd_deps, cmd_deps_tree, cmd_install_deps, cmd_add from .config_commands import cmd_config from .settings_commands import cmd_settings from .system_deps_commands import cmd_system_deps +from .mcp_commands import cmd_mcp def main(): @@ -450,6 +451,53 @@ def main(): # Default for system-deps with no subcommand (show status) p_sysdeps.set_defaults(func=cmd_system_deps) + # mcp command + p_mcp = subparsers.add_parser("mcp", help="Manage MCP servers") + mcp_sub = p_mcp.add_subparsers(dest="mcp_cmd", help="MCP commands") + + # mcp list + p_mcp_list = mcp_sub.add_parser("list", help="List configured MCP servers") + p_mcp_list.set_defaults(func=cmd_mcp) + + # mcp connect + p_mcp_connect = mcp_sub.add_parser("connect", help="Connect to an MCP server and discover tools") + p_mcp_connect.add_argument("name", help="Server name from mcp.yaml") + p_mcp_connect.set_defaults(func=cmd_mcp) + + # mcp add + p_mcp_add = mcp_sub.add_parser("add", help="Add an MCP server") + p_mcp_add.add_argument("name", help="Server name") + p_mcp_add.add_argument( + "--transport", choices=["stdio"], default="stdio", + help="Transport type (M7.2 supports stdio)", + ) + p_mcp_add.add_argument("--command", help="Executable for stdio servers") + p_mcp_add.add_argument( + "--arg", dest="server_args", action="append", default=[], metavar="VALUE", + help="Command argument; repeat for each argument (use --arg=-y for leading dashes)", + ) + p_mcp_add.add_argument("--cwd", help="Working directory") + p_mcp_add.add_argument( + "--env", action="append", default=[], metavar="NAME=VALUE", + help="Server environment entry; repeat as needed (supports ${NAME} references)", + ) + p_mcp_add.add_argument( + "--inherit-env", action="append", default=[], metavar="NAME", + help="Additionally inherit this environment variable; repeat as needed", + ) + p_mcp_add.add_argument("--timeout", type=float, default=30, help="Per-operation timeout in seconds") + p_mcp_add.add_argument("-d", "--description", help="Server description") + p_mcp_add.add_argument("--force", action="store_true", help="Overwrite existing server") + p_mcp_add.set_defaults(func=cmd_mcp) + + # mcp remove + p_mcp_remove = mcp_sub.add_parser("remove", help="Remove an MCP server") + p_mcp_remove.add_argument("name", help="Server name") + p_mcp_remove.set_defaults(func=cmd_mcp) + + # Default for mcp with no subcommand (list) + p_mcp.set_defaults(func=cmd_mcp, mcp_cmd="list") + args = parser.parse_args() # If no command, launch UI diff --git a/src/cmdforge/cli/mcp_commands.py b/src/cmdforge/cli/mcp_commands.py new file mode 100644 index 0000000..44dd5ec --- /dev/null +++ b/src/cmdforge/cli/mcp_commands.py @@ -0,0 +1,169 @@ +"""MCP management commands.""" + +import re + +from ..mcp_client import ( + DEFAULT_INHERITED_ENV, + McpServerConfig, + McpClientManager, + _sanitize, + load_mcp_config, + save_mcp_config, +) + + +def cmd_mcp(args): + """Manage MCP servers.""" + if getattr(args, "mcp_cmd", "list") == "list": + return _cmd_mcp_list(args) + elif args.mcp_cmd == "connect": + return _cmd_mcp_connect(args) + elif args.mcp_cmd == "add": + return _cmd_mcp_add(args) + elif args.mcp_cmd == "remove": + return _cmd_mcp_remove(args) + return 0 + + +def _cmd_mcp_list(args): + try: + servers = load_mcp_config() + except (OSError, ValueError) as exc: + print(f"Error loading MCP config: {exc}") + return 1 + if not servers: + print("No MCP servers configured.") + print("\nAdd a server: cmdforge mcp add --command npx " + "--arg=-y --arg @scope/server") + print("Config file: ~/.cmdforge/mcp.yaml") + return 0 + + print(f"MCP servers ({len(servers)}):\n") + for s in servers: + print(f" [+] {s.name}") + if s.description: + print(f" Description: {s.description}") + print(f" Transport: {s.transport}") + if s.transport == "stdio": + command = " ".join([s.command or ""] + s.args) + print(f" Command: {_sanitize(command)}") + if s.cwd: + print(f" Working dir: {s.cwd}") + print(f" Approved: {'yes' if s.approved else 'no'}") + print(f" Timeout: {s.timeout}s") + print() + return 0 + + +def _cmd_mcp_connect(args): + name = args.name + try: + with McpClientManager() as manager: + tools = manager.discover(name) + except (ImportError, KeyError, OSError, PermissionError, RuntimeError, + TimeoutError, ValueError) as exc: + print(f"Error connecting to server '{name}': {exc}") + return 1 + + print(f"Connected to '{name}'. Available tools ({len(tools)}):\n") + for t in tools: + print(f" [+] {t.get('name', 'unknown')}") + desc = t.get("description", "") + if desc: + print(f" {desc[:120]}") + print() + print(f"Use these tools in your CmdForge tool configs with:") + print(f" steps:") + print(f" - type: mcp") + print(f" server: {name}") + print(f" tool: ") + print(f" output_var: result") + return 0 + + +def _cmd_mcp_add(args): + name = args.name + try: + servers = load_mcp_config() + except (OSError, ValueError) as exc: + print(f"Error loading MCP config: {exc}") + return 1 + existing = {s.name: s for s in servers} + + if name in existing and not getattr(args, "force", False): + print(f"Server '{name}' already exists. Use --force to overwrite.") + return 1 + + try: + server_env = _parse_env_entries(getattr(args, "env", [])) + except ValueError as exc: + print(f"Error: {exc}") + return 1 + + inherited = list(DEFAULT_INHERITED_ENV) + for env_name in getattr(args, "inherit_env", []): + if env_name not in inherited: + inherited.append(env_name) + + config = McpServerConfig( + name=name, + transport=getattr(args, "transport", "stdio"), + command=getattr(args, "command", None), + args=getattr(args, "server_args", []) or [], + cwd=getattr(args, "cwd", None), + env=server_env, + inherit_env=inherited, + timeout=getattr(args, "timeout", 30), + description=getattr(args, "description", "") or "", + approved=True, + ) + + try: + config.validate() + except ValueError as exc: + print(f"Error: {exc}") + return 1 + + if name in existing: + servers = [s for s in servers if s.name != name] + servers.append(config) + try: + save_mcp_config(servers) + except OSError as exc: + print(f"Error saving MCP config: {exc}") + return 1 + print(f"MCP server '{name}' saved and approved to run.") + return 0 + + +def _cmd_mcp_remove(args): + name = args.name + try: + servers = load_mcp_config() + except (OSError, ValueError) as exc: + print(f"Error loading MCP config: {exc}") + return 1 + if not any(s.name == name for s in servers): + print(f"Server '{name}' not found.") + return 1 + + servers = [s for s in servers if s.name != name] + try: + save_mcp_config(servers) + except OSError as exc: + print(f"Error saving MCP config: {exc}") + return 1 + print(f"MCP server '{name}' removed.") + return 0 + + +def _parse_env_entries(entries): + result = {} + for entry in entries: + if "=" not in entry: + raise ValueError(f"invalid --env value '{entry}'; expected NAME=VALUE") + name, value = entry.split("=", 1) + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name): + raise ValueError(f"invalid environment variable name '{name}'") + result[name] = value + return result diff --git a/src/cmdforge/mcp_client.py b/src/cmdforge/mcp_client.py new file mode 100644 index 0000000..c776ec8 --- /dev/null +++ b/src/cmdforge/mcp_client.py @@ -0,0 +1,484 @@ +"""MCP client support built on the official MCP Python SDK.""" + +import asyncio +import hashlib +import json +import os +import re +import tempfile +import time +from dataclasses import dataclass, field +from datetime import timedelta +from pathlib import Path +from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar + +import yaml + +MCP_CONFIG_FILE = Path.home() / ".cmdforge" / "mcp.yaml" +MCP_CONFIG_VERSION = 1 +RESULT_MODES = ("auto", "structured", "content", "text") +SUPPORTED_TRANSPORTS = ("stdio",) +DEFAULT_INHERITED_ENV = ( + "PATH", + "HOME", + "USER", + "LOGNAME", + "TMPDIR", + "TMP", + "TEMP", + "SYSTEMROOT", + "WINDIR", + "COMSPEC", + "PATHEXT", +) + +_ENV_REFERENCE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") +_ENV_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +_SECRET_PATTERNS = ( + re.compile(r"(?i)bearer\s+\S+"), + re.compile(r"(?i)\b(?:sk|ghp|github_pat)-[A-Za-z0-9_-]+"), + re.compile(r"(?i)(?:api[_-]?key|token|secret|password)\s*[:=]\s*\S+"), +) + +T = TypeVar("T") + + +@dataclass +class McpServerConfig: + """Configuration for one MCP server. + + M7.2 intentionally supports stdio only. ``approved`` records explicit + user consent to execute the configured local command. + """ + + name: str + transport: str = "stdio" + command: Optional[str] = None + args: List[str] = field(default_factory=list) + cwd: Optional[str] = None + env: Dict[str, str] = field(default_factory=dict) + inherit_env: List[str] = field(default_factory=lambda: list(DEFAULT_INHERITED_ENV)) + timeout: float = 30 + description: str = "" + approved: bool = False + + def validate(self) -> None: + if not isinstance(self.name, str) or not self.name.strip(): + raise ValueError("MCP server name must be a non-empty string") + if self.transport not in SUPPORTED_TRANSPORTS: + raise ValueError( + f"Unsupported MCP transport '{self.transport}'. " + f"Supported in M7.2: {', '.join(SUPPORTED_TRANSPORTS)}" + ) + if not isinstance(self.command, str) or not self.command.strip(): + raise ValueError(f"MCP server '{self.name}' requires a command") + if not isinstance(self.args, list) or not all(isinstance(arg, str) for arg in self.args): + raise ValueError(f"MCP server '{self.name}' args must be a list of strings") + if not isinstance(self.env, dict) or not all( + isinstance(key, str) + and _ENV_NAME.fullmatch(key) + and isinstance(value, str) + for key, value in self.env.items() + ): + raise ValueError( + f"MCP server '{self.name}' env must map valid variable names to strings" + ) + if not isinstance(self.inherit_env, list) or not all( + isinstance(name, str) and _ENV_NAME.fullmatch(name) + for name in self.inherit_env + ): + raise ValueError( + f"MCP server '{self.name}' inherit_env must contain valid variable names" + ) + if isinstance(self.timeout, bool) or not isinstance(self.timeout, (int, float)): + raise ValueError(f"MCP server '{self.name}' timeout must be a number") + if self.timeout <= 0 or self.timeout > 3600: + raise ValueError( + f"MCP server '{self.name}' timeout must be greater than 0 and at most 3600 seconds" + ) + if self.cwd is not None and not isinstance(self.cwd, str): + raise ValueError(f"MCP server '{self.name}' cwd must be a string") + if not isinstance(self.approved, bool): + raise ValueError(f"MCP server '{self.name}' approved must be a boolean") + + +def _fingerprint(cfg: McpServerConfig) -> str: + payload = json.dumps( + [ + cfg.transport, + cfg.command, + cfg.args, + cfg.cwd, + cfg.env, + cfg.inherit_env, + cfg.timeout, + cfg.approved, + ], + sort_keys=True, + ) + return hashlib.sha256(payload.encode()).hexdigest()[:16] + + +def _sanitize(value: str, secrets: Optional[List[str]] = None) -> str: + """Redact common credential forms and explicit resolved secret values.""" + sanitized = str(value) + for secret in secrets or []: + if secret: + sanitized = sanitized.replace(secret, "[redacted]") + for pattern in _SECRET_PATTERNS: + sanitized = pattern.sub("[redacted]", sanitized) + return sanitized + + +def _expand_env_value(value: str) -> str: + def replace(match: re.Match) -> str: + name = match.group(1) + if name not in os.environ: + raise ValueError(f"Environment variable '{name}' referenced by MCP config is not set") + return os.environ[name] + + return _ENV_REFERENCE.sub(replace, value) + + +def _build_server_env(cfg: McpServerConfig) -> Dict[str, str]: + """Build a minimal subprocess environment without leaking all credentials.""" + environment = { + name: os.environ[name] + for name in cfg.inherit_env + if name in os.environ + } + for name, value in cfg.env.items(): + environment[name] = _expand_env_value(value) + return environment + + +def load_mcp_config() -> List[McpServerConfig]: + if not MCP_CONFIG_FILE.exists(): + return [] + + data = yaml.safe_load(MCP_CONFIG_FILE.read_text()) or {} + if not isinstance(data, dict): + raise ValueError("MCP config must be a YAML mapping") + version = data.get("version", MCP_CONFIG_VERSION) + if version != MCP_CONFIG_VERSION: + raise ValueError(f"Unsupported MCP config version: {version}") + raw_servers = data.get("servers") or {} + if not isinstance(raw_servers, dict): + raise ValueError("MCP config 'servers' must be a mapping") + + servers = [] + for name, raw in raw_servers.items(): + if not isinstance(raw, dict): + raise ValueError(f"MCP server '{name}' config must be a mapping") + cfg = McpServerConfig( + name=name, + transport=raw.get("transport", "stdio"), + command=raw.get("command"), + args=raw.get("args", []), + cwd=raw.get("cwd"), + env=raw.get("env", {}), + inherit_env=raw.get("inherit_env", list(DEFAULT_INHERITED_ENV)), + timeout=raw.get("timeout", 30), + description=raw.get("description", ""), + approved=raw.get("approved", False), + ) + cfg.validate() + servers.append(cfg) + return servers + + +def save_mcp_config(servers: List[McpServerConfig]) -> None: + for server in servers: + server.validate() + MCP_CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + body = { + "version": MCP_CONFIG_VERSION, + "servers": {server.name: _server_to_dict(server) for server in servers}, + } + serialized = yaml.safe_dump(body, default_flow_style=False, sort_keys=False) + temp_path = None + try: + MCP_CONFIG_FILE.parent.chmod(0o700) + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=MCP_CONFIG_FILE.parent, + prefix=".mcp-", + suffix=".yaml.tmp", + delete=False, + ) as temp_file: + temp_path = Path(temp_file.name) + temp_path.chmod(0o600) + temp_file.write(serialized) + temp_file.flush() + os.fsync(temp_file.fileno()) + os.replace(temp_path, MCP_CONFIG_FILE) + MCP_CONFIG_FILE.chmod(0o600) + finally: + if temp_path is not None and temp_path.exists(): + temp_path.unlink() + + +def _server_to_dict(server: McpServerConfig) -> dict: + data: dict = { + "transport": server.transport, + "command": server.command, + "approved": server.approved, + } + if server.args: + data["args"] = server.args + if server.cwd: + data["cwd"] = server.cwd + if server.env: + data["env"] = server.env + if server.inherit_env != list(DEFAULT_INHERITED_ENV): + data["inherit_env"] = server.inherit_env + if server.timeout != 30: + data["timeout"] = server.timeout + if server.description: + data["description"] = server.description + return data + + +def _require_mcp_sdk() -> None: + try: + import mcp # noqa: F401 + except ImportError as exc: + raise ImportError( + "MCP support requires the mcp package. Install with: pip install 'cmdforge[mcp]'" + ) from exc + + +def _serialize_tool(tool: Any) -> Dict[str, Any]: + if hasattr(tool, "model_dump"): + return tool.model_dump(mode="json", by_alias=True, exclude_none=True) + if isinstance(tool, dict): + return tool + return { + key: getattr(tool, key) + for key in ("name", "description", "inputSchema", "outputSchema") + if hasattr(tool, key) + } + + +class McpClientManager: + """Invocation-scoped MCP configuration and schema manager. + + SDK sessions are intentionally scoped to each discovery or tool operation + in M7.2. This guarantees subprocess cleanup while the manager reuses loaded + configuration and discovered schemas across all MCP steps in one tool run. + """ + + def __init__(self): + self._configs: Dict[str, McpServerConfig] = {} + self._schemas: Dict[str, Dict[str, Any]] = {} + self._schema_fingerprints: Dict[str, str] = {} + self._loaded = False + + def __enter__(self) -> "McpClientManager": + return self + + def __exit__(self, exc_type, exc_value, traceback) -> None: + self.shutdown() + + def _ensure_loaded(self) -> None: + if self._loaded: + return + for cfg in load_mcp_config(): + self._configs[cfg.name] = cfg + self._loaded = True + + def shutdown(self) -> None: + self._schemas.clear() + self._schema_fingerprints.clear() + + def list_servers(self) -> List[McpServerConfig]: + self._ensure_loaded() + return list(self._configs.values()) + + def _get_approved_config(self, server_name: str) -> McpServerConfig: + self._ensure_loaded() + cfg = self._configs.get(server_name) + if not cfg: + raise KeyError(f"MCP server '{server_name}' not configured. Add it to {MCP_CONFIG_FILE}") + cfg.validate() + if not cfg.approved: + command = " ".join([cfg.command or ""] + cfg.args) + raise PermissionError( + f"MCP server '{server_name}' is not approved to execute: {_sanitize(command)}. " + "Re-add it with 'cmdforge mcp add ...' or set approved: true after review." + ) + return cfg + + def discover(self, server_name: str, refresh: bool = False) -> List[Dict[str, Any]]: + _require_mcp_sdk() + cfg = self._get_approved_config(server_name) + fingerprint = _fingerprint(cfg) + if ( + not refresh + and server_name in self._schemas + and self._schema_fingerprints.get(server_name) == fingerprint + ): + return list(self._schemas[server_name].values()) + + async def list_all_tools(session: Any) -> List[Dict[str, Any]]: + tools: List[Dict[str, Any]] = [] + cursor = None + seen_cursors = set() + while True: + page = await session.list_tools(cursor=cursor) + tools.extend(_serialize_tool(tool) for tool in page.tools) + cursor = getattr(page, "nextCursor", None) + if not cursor: + break + if cursor in seen_cursors: + raise RuntimeError("MCP server repeated a tools/list cursor") + seen_cursors.add(cursor) + return tools + + tools = _run_stdio_operation(cfg, list_all_tools, "discover tools") + self._schemas[server_name] = {tool["name"]: tool for tool in tools} + self._schema_fingerprints[server_name] = fingerprint + return tools + + def get_tool_schema(self, server_name: str, tool_name: str) -> Dict[str, Any]: + if server_name not in self._schemas: + self.discover(server_name) + server_schemas = self._schemas.get(server_name, {}) + if tool_name not in server_schemas: + raise KeyError( + f"Tool '{tool_name}' not found on MCP server '{server_name}'. " + f"Available: {list(server_schemas.keys())}" + ) + return server_schemas[tool_name] + + def call_tool( + self, + server_name: str, + tool_name: str, + arguments: Dict[str, Any], + result_mode: str = "auto", + ) -> Any: + if result_mode not in RESULT_MODES: + raise ValueError(f"result_mode must be one of {RESULT_MODES}") + if not isinstance(arguments, dict): + raise ValueError("MCP tool arguments must be a mapping") + + _require_mcp_sdk() + cfg = self._get_approved_config(server_name) + + async def call(session: Any) -> Any: + result = await session.call_tool( + tool_name, + arguments, + read_timeout_seconds=timedelta(seconds=cfg.timeout), + ) + return _normalize_result(result, result_mode) + + return _run_stdio_operation(cfg, call, f"call {tool_name}") + + +def _run_stdio_operation( + cfg: McpServerConfig, + operation: Callable[[Any], Awaitable[T]], + operation_name: str, +) -> T: + from mcp import ClientSession, StdioServerParameters + from mcp.client.stdio import stdio_client + + resolved_env = _build_server_env(cfg) + server_params = StdioServerParameters( + command=cfg.command, + args=cfg.args, + cwd=cfg.cwd, + env=resolved_env, + ) + + async def run(stderr_file: Any) -> T: + async with stdio_client(server_params, errlog=stderr_file) as (read, write): + async with ClientSession( + read, + write, + read_timeout_seconds=timedelta(seconds=cfg.timeout), + ) as session: + await session.initialize() + return await operation(session) + + # The SDK passes errlog directly to subprocess.Popen, so it must expose a + # real file descriptor (StringIO is insufficient). + with tempfile.TemporaryFile(mode="w+", encoding="utf-8") as stderr_file: + started = time.monotonic() + try: + return asyncio.run(asyncio.wait_for(run(stderr_file), timeout=cfg.timeout)) + except (asyncio.TimeoutError, TimeoutError) as exc: + stderr_file.seek(0) + detail = _sanitize( + stderr_file.read().strip(), + [resolved_env[name] for name in cfg.env if name in resolved_env], + ) + suffix = f" (server stderr: {detail[:500]})" if detail else "" + raise TimeoutError( + f"MCP {operation_name} on '{cfg.name}' timed out after {cfg.timeout}s{suffix}" + ) from exc + except Exception as exc: + # Some SDK/AnyIO cleanup paths replace asyncio's cancellation with + # an ExceptionGroup (for example BrokenResourceError). The elapsed + # deadline remains authoritative: the requested operation timed out. + if time.monotonic() - started >= cfg.timeout: + raise TimeoutError( + f"MCP {operation_name} on '{cfg.name}' timed out after {cfg.timeout}s" + ) from exc + # Only configured values are secrets. Redacting inherited PATH/HOME + # values would make ordinary diagnostics unreadable. + secrets = [resolved_env[name] for name in cfg.env if name in resolved_env] + stderr_file.seek(0) + detail = _sanitize(stderr_file.read().strip(), secrets) + message = _sanitize(str(exc), secrets) + if detail: + message = f"{message} (server stderr: {detail[:500]})" + raise RuntimeError( + f"MCP {operation_name} on '{cfg.name}' failed: {message}" + ) from exc + + +def _normalize_result(result: Any, mode: str) -> Any: + if mode not in RESULT_MODES: + raise ValueError(f"result_mode must be one of {RESULT_MODES}") + + content = getattr(result, "content", []) or [] + if getattr(result, "isError", False): + error_content = [_serialize_content_block(block) for block in content] + raise RuntimeError(f"MCP tool returned an error: {error_content}") + + structured = getattr(result, "structuredContent", None) + if mode in ("auto", "structured") and structured is not None: + return structured + if mode == "structured": + raise ValueError("MCP tool did not return structuredContent") + + text_parts = [ + block.text + for block in content + if getattr(block, "type", None) == "text" and hasattr(block, "text") + ] + if mode == "text": + return "\n".join(text_parts) + + serialized = [_serialize_content_block(block) for block in content] + if mode == "content": + return serialized + if content and len(text_parts) == len(content): + return "\n".join(text_parts) + return serialized + + +def _serialize_content_block(block: Any) -> Dict[str, Any]: + if hasattr(block, "model_dump"): + return block.model_dump(mode="json", by_alias=True, exclude_none=True) + if isinstance(block, dict): + return block + result = {} + for name in ("type", "text", "mimeType", "data", "uri", "resource"): + if hasattr(block, name): + result[name] = getattr(block, name) + return result diff --git a/src/cmdforge/runner.py b/src/cmdforge/runner.py index c82e2ef..b68a83d 100644 --- a/src/cmdforge/runner.py +++ b/src/cmdforge/runner.py @@ -9,11 +9,12 @@ from typing import Optional import yaml -from .tool import Tool, PromptStep, CodeStep, ToolStep +from .tool import Tool, PromptStep, CodeStep, ToolStep, McpStep from .providers import call_provider, mock_provider from .resolver import resolve_tool, ToolNotFoundError, ToolSpec, install_from_registry from .manifest import load_manifest from .profiles import load_profile +from .mcp_client import McpClientManager # Maximum recursion depth for nested tool calls MAX_TOOL_DEPTH = 10 @@ -352,7 +353,9 @@ def substitute_variables(template: str, variables: dict, warn_non_scalar: bool = if base_name == "settings": # Settings dict - leave unchanged (handled specially) return match.group(0) - # Replace with empty string for None, str() for other values + if isinstance(base_value, (dict, list)): + return json.dumps(base_value) + # Replace with empty string for None, str() for other scalar values return "" if base_value is None else str(base_value) # Handle nested access @@ -901,6 +904,9 @@ def run_tool( output = substitute_variables(tool.output, variables, warn_non_scalar=verbose) return output, 0 + # Reuse MCP configuration and schema caches across all MCP steps in this run. + mcp_manager = None + # Execute each step for i, step in enumerate(tool.steps): if verbose: @@ -910,6 +916,8 @@ def run_tool( step_type = "CODE" elif isinstance(step, ToolStep): step_type = f"TOOL({step.tool})" + elif isinstance(step, McpStep): + step_type = f"MCP({step.server}/{step.tool})" else: step_type = "UNKNOWN" print(f"[verbose] Step {i+1}: {step_type} -> {{{step.output_var}}}", file=sys.stderr) @@ -981,12 +989,79 @@ def run_tool( return "", 3 variables[step.output_var] = output + elif isinstance(step, McpStep): + if verbose or dry_run: + print(f"=== MCP (Step {i+1}) -> {{{step.output_var}}} ===", file=sys.stderr) + print(f" Server: {step.server} Tool: {step.tool}", file=sys.stderr) + print(f" Args: {step.arguments}", file=sys.stderr) + print("=== END MCP ===", file=sys.stderr) + + if dry_run: + variables[step.output_var] = f"[DRY RUN - would call mcp:{step.server}/{step.tool}]" + else: + args = _substitute_mcp_args(step.arguments, variables) + try: + if mcp_manager is None: + mcp_manager = McpClientManager() + result = mcp_manager.call_tool( + step.server, step.tool, args, + result_mode=step.result_mode, + ) + variables[step.output_var] = result + except ImportError as e: + print(f"{e}", file=sys.stderr) + return "", 4 + except Exception as e: + print(f"MCP step failed (server={step.server}, tool={step.tool}): {e}", file=sys.stderr) + return "", 4 + # Generate final output output = substitute_variables(tool.output, variables, warn_non_scalar=verbose) return output, 0 +def _substitute_mcp_args(arguments: dict, variables: dict) -> dict: + """Substitute variables in MCP arguments, preserving types.""" + result = {} + for key, value in arguments.items(): + result[key] = _deep_substitute(value, variables) + return result + + +def _deep_substitute(value, variables: dict): + if isinstance(value, str): + import re + + exact_reference = re.fullmatch( + r"\s*\{([a-zA-Z_][a-zA-Z0-9_]*(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)\}\s*", + value, + ) + if exact_reference: + path = exact_reference.group(1).split(".") + if path[0] in variables: + resolved = variables[path[0]] + found = True + for key in path[1:]: + if isinstance(resolved, dict) and key in resolved: + resolved = resolved[key] + elif hasattr(resolved, key): + resolved = getattr(resolved, key) + else: + found = False + break + if found: + return resolved + + substituted = substitute_variables(value, variables) + return substituted if substituted != value else value + elif isinstance(value, dict): + return {k: _deep_substitute(v, variables) for k, v in value.items()} + elif isinstance(value, list): + return [_deep_substitute(v, variables) for v in value] + return value + + def create_argument_parser(tool: Tool) -> argparse.ArgumentParser: """ Create an argument parser for a tool. diff --git a/src/cmdforge/tool.py b/src/cmdforge/tool.py index 2362ced..4378729 100644 --- a/src/cmdforge/tool.py +++ b/src/cmdforge/tool.py @@ -226,8 +226,59 @@ class ToolStep: ) -Step = PromptStep | CodeStep | ToolStep +@dataclass +class McpStep: + """A step that calls an MCP server tool.""" + server: str # Server name in ~/.cmdforge/mcp.yaml + tool: str # Tool name on that server + arguments: dict = field(default_factory=dict) + output_var: str = "mcp_result" + result_mode: str = "auto" # auto | structured | content | text + name: Optional[str] = None + def to_dict(self) -> dict: + d = { + "type": "mcp", + "server": self.server, + "tool": self.tool, + "output_var": self.output_var, + } + if self.arguments: + d["arguments"] = self.arguments + if self.result_mode != "auto": + d["result_mode"] = self.result_mode + if self.name: + d["name"] = self.name + return d + + @classmethod + def from_dict(cls, data: dict) -> "McpStep": + server = data.get("server") + tool = data.get("tool") + arguments = data.get("arguments", {}) + output_var = data.get("output_var", "mcp_result") + result_mode = data.get("result_mode", "auto") + if not isinstance(server, str) or not server.strip(): + raise ValueError("MCP step server must be a non-empty string") + if not isinstance(tool, str) or not tool.strip(): + raise ValueError("MCP step tool must be a non-empty string") + if not isinstance(arguments, dict): + raise ValueError("MCP step arguments must be a mapping") + if not isinstance(output_var, str) or not output_var.strip(): + raise ValueError("MCP step output_var must be a non-empty string") + if result_mode not in ("auto", "structured", "content", "text"): + raise ValueError("MCP step result_mode must be auto, structured, content, or text") + return cls( + server=server, + tool=tool, + arguments=arguments, + output_var=output_var, + result_mode=result_mode, + name=data.get("name"), + ) + + +Step = PromptStep | CodeStep | ToolStep | McpStep @dataclass class ToolSource: @@ -318,6 +369,8 @@ class Tool: steps.append(CodeStep.from_dict(step)) elif step.get("type") == "tool": steps.append(ToolStep.from_dict(step)) + elif step.get("type") == "mcp": + steps.append(McpStep.from_dict(step)) # Parse source attribution if present source = None diff --git a/tests/fixtures/mcp_test_server.py b/tests/fixtures/mcp_test_server.py new file mode 100644 index 0000000..fcc6139 --- /dev/null +++ b/tests/fixtures/mcp_test_server.py @@ -0,0 +1,44 @@ +"""Small real MCP stdio server used by the client integration tests.""" + +import os +import time + +from mcp.server.fastmcp import FastMCP + + +server = FastMCP("CmdForge MCP test server") + + +@server.tool() +def echo(message: str) -> str: + """Return a message unchanged.""" + return message + + +@server.tool() +def add(a: int, b: int) -> dict[str, int]: + """Add two integers.""" + return {"sum": a + b} + + +@server.tool() +def read_env(name: str) -> str: + """Return an environment value, or an empty string if absent.""" + return os.environ.get(name, "") + + +@server.tool() +def current_directory() -> str: + """Return the server process working directory.""" + return os.getcwd() + + +@server.tool() +def wait(seconds: float) -> str: + """Block for the requested duration.""" + time.sleep(seconds) + return "done" + + +if __name__ == "__main__": + server.run(transport="stdio") diff --git a/tests/test_mcp.py b/tests/test_mcp.py new file mode 100644 index 0000000..c6930cf --- /dev/null +++ b/tests/test_mcp.py @@ -0,0 +1,399 @@ +"""Tests for MCP client and McpStep execution.""" + +import sys +import time +from pathlib import Path + +import pytest +from cmdforge.mcp_client import ( + DEFAULT_INHERITED_ENV, + McpServerConfig, + McpClientManager, + _build_server_env, + _normalize_result, + _sanitize, + _fingerprint, + load_mcp_config, + save_mcp_config, +) +from cmdforge.runner import _substitute_mcp_args, _deep_substitute, run_tool +from cmdforge.tool import McpStep, Tool + + +class TestMcpServerConfig: + def test_defaults(self): + cfg = McpServerConfig(name="test") + assert cfg.transport == "stdio" + assert cfg.timeout == 30 + assert cfg.args == [] + assert cfg.env == {} + assert cfg.approved is False + + @pytest.mark.parametrize( + "change, message", + [ + ({"transport": "streamable-http"}, "Unsupported MCP transport"), + ({"args": "-y"}, "list of strings"), + ({"timeout": 0}, "greater than 0"), + ({"timeout": True}, "must be a number"), + ], + ) + def test_validation_rejects_invalid_config(self, change, message): + cfg = McpServerConfig(name="test", command="server", **change) + with pytest.raises(ValueError, match=message): + cfg.validate() + + def test_fingerprint_changes_with_command(self): + a = McpServerConfig(name="a", command="cmd-a") + b = McpServerConfig(name="b", command="cmd-b") + assert _fingerprint(a) != _fingerprint(b) + + def test_fingerprint_same_for_same_config(self): + a = McpServerConfig(name="a", command="cmd", args=["-v"]) + b = McpServerConfig(name="b", command="cmd", args=["-v"]) + assert _fingerprint(a) == _fingerprint(b) + + +class TestSanitize: + def test_redacts_bearer(self): + assert "redacted" in _sanitize("Bearer abcdefghijklmnopqrstuvwxyz") + + def test_redacts_sk_prefix(self): + assert "redacted" in _sanitize("sk-abcdefghijklmnopqrstuvwxyz") + + def test_preserves_normal_text(self): + assert _sanitize("hello world") == "hello world" + + def test_redacts_explicit_secret(self): + assert _sanitize("failed with hunter2", ["hunter2"]) == "failed with [redacted]" + + +class TestEnvironmentIsolation: + def test_only_allowlisted_environment_is_inherited(self, monkeypatch): + monkeypatch.setenv("CMDFORGE_TEST_SECRET", "do-not-leak") + cfg = McpServerConfig(name="test", command="server") + environment = _build_server_env(cfg) + assert "CMDFORGE_TEST_SECRET" not in environment + assert set(environment).issubset(set(DEFAULT_INHERITED_ENV)) + + def test_explicit_environment_reference_is_resolved(self, monkeypatch): + monkeypatch.setenv("CMDFORGE_TEST_SECRET", "allowed") + cfg = McpServerConfig( + name="test", command="server", env={"SERVER_TOKEN": "${CMDFORGE_TEST_SECRET}"} + ) + assert _build_server_env(cfg)["SERVER_TOKEN"] == "allowed" + + def test_missing_environment_reference_fails(self, monkeypatch): + monkeypatch.delenv("CMDFORGE_MISSING", raising=False) + cfg = McpServerConfig( + name="test", command="server", env={"SERVER_TOKEN": "${CMDFORGE_MISSING}"} + ) + with pytest.raises(ValueError, match="is not set"): + _build_server_env(cfg) + + +class TestMcpConfigPersistence: + @pytest.fixture + def temp_mcp_file(self, tmp_path, monkeypatch): + config_file = tmp_path / "mcp.yaml" + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file) + yield config_file + + def test_save_and_load(self, temp_mcp_file): + servers = [ + McpServerConfig( + name="filesystem", + command="npx", + args=["-y", "@scope/server", "/tmp"], + timeout=30, + approved=True, + ) + ] + save_mcp_config(servers) + loaded = load_mcp_config() + assert len(loaded) == 1 + assert loaded[0].name == "filesystem" + assert loaded[0].command == "npx" + assert loaded[0].args == ["-y", "@scope/server", "/tmp"] + assert loaded[0].approved is True + + def test_load_no_file(self, temp_mcp_file): + assert load_mcp_config() == [] + + def test_save_creates_0600_permissions(self, temp_mcp_file): + save_mcp_config([McpServerConfig(name="test", command="echo")]) + perms = oct(temp_mcp_file.stat().st_mode & 0o777) + assert perms == "0o600" + + +class TestResultNormalization: + def test_auto_prefers_structured_content(self): + class FakeResult: + isError = False + structuredContent = {"sum": 3} + content = [type("Block", (), {"text": "plain text"})()] + + result = _normalize_result(FakeResult(), "auto") + assert result == {"sum": 3} + + def test_auto_falls_back_to_text(self): + class FakeResult: + isError = False + structuredContent = None + content = [type("Block", (), {"type": "text", "text": "hello"})()] + + result = _normalize_result(FakeResult(), "auto") + assert result == "hello" + + def test_auto_preserves_mixed_content(self): + class FakeResult: + isError = False + structuredContent = None + content = [ + {"type": "text", "text": "hello"}, + {"type": "image", "mimeType": "image/png", "data": "AA=="}, + ] + + result = _normalize_result(FakeResult(), "auto") + assert result == FakeResult.content + + def test_content_mode_always_returns_blocks(self): + class FakeResult: + isError = False + structuredContent = None + content = [{"type": "text", "text": "hello"}] + + assert _normalize_result(FakeResult(), "content") == FakeResult.content + + def test_structured_fails_without_structured_content(self): + class FakeResult: + isError = False + structuredContent = None + content = [type("Block", (), {"type": "text", "text": "hello"})()] + + with pytest.raises(ValueError, match="structuredContent"): + _normalize_result(FakeResult(), "structured") + + def test_text_mode(self): + class FakeResult: + isError = False + structuredContent = None + content = [ + type("Block", (), {"type": "text", "text": "line1"})(), + type("Block", (), {"type": "text", "text": "line2"})(), + ] + + result = _normalize_result(FakeResult(), "text") + assert result == "line1\nline2" + + def test_iserror_causes_failure(self): + class FakeResult: + isError = True + structuredContent = None + content = [type("Block", (), {"type": "text", "text": "error message"})()] + + with pytest.raises(RuntimeError, match="error"): + _normalize_result(FakeResult(), "auto") + + +class TestMcpClientManager: + @pytest.fixture + def manager(self, tmp_path, monkeypatch): + config_file = tmp_path / "mcp.yaml" + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file) + return McpClientManager() + + def test_list_servers_empty(self, manager): + assert manager.list_servers() == [] + + def test_list_servers_with_config(self, manager, tmp_path, monkeypatch): + config_file = tmp_path / "mcp.yaml" + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file) + save_mcp_config([McpServerConfig(name="test", command="echo")]) + + mgr = McpClientManager() + servers = mgr.list_servers() + assert len(servers) == 1 + assert servers[0].name == "test" + + def test_call_tool_unknown_server(self, manager): + with pytest.raises(KeyError, match="not configured"): + manager.call_tool("unknown", "tool", {}) + + def test_unapproved_server_is_not_executed(self, tmp_path, monkeypatch): + config_file = tmp_path / "mcp.yaml" + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file) + save_mcp_config([McpServerConfig(name="test", command="echo")]) + with pytest.raises(PermissionError, match="not approved"): + McpClientManager().discover("test") + + def test_missing_sdk_has_actionable_error(self, tmp_path, monkeypatch): + config_file = tmp_path / "mcp.yaml" + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file) + save_mcp_config([ + McpServerConfig(name="test", command="echo", approved=True) + ]) + monkeypatch.setitem(sys.modules, "mcp", None) + with pytest.raises(ImportError, match=r"cmdforge\[mcp\]"): + McpClientManager().discover("test") + + +class TestArgumentSubstitution: + def test_single_variable_preserves_integer(self): + result = _deep_substitute("42", {"limit": 42}) + assert result == "42" # no braces, so no substitution + + def test_braced_variable_preserves_type(self): + variables = {"settings": {"limit": 50}} + result = _substitute_mcp_args( + {"options": {"limit": "{settings.limit}"}}, variables + ) + assert result["options"]["limit"] == 50 + + def test_nested_dict_substitution(self): + variables = {"path": "/tmp/data.csv"} + result = _substitute_mcp_args( + {"files": [{"source": "{path}"}]}, variables + ) + assert result["files"][0]["source"] == "/tmp/data.csv" + + def test_string_with_variable_inside_text(self): + variables = {"name": "report"} + result = _deep_substitute("/tmp/{name}.csv", variables) + assert result == "/tmp/report.csv" + + @pytest.mark.parametrize( + "value", + [ + {"a": 1}, + [1, 2], + -5, + "001", + "true", + False, + None, + ], + ) + def test_exact_reference_preserves_original_type(self, value): + assert _deep_substitute("{value}", {"value": value}) == value + + +class TestMcpStep: + def test_round_trip(self): + original = McpStep( + server="fixture", tool="add", arguments={"a": 1, "b": 2}, + output_var="sum", result_mode="structured", + ) + assert McpStep.from_dict(original.to_dict()) == original + + @pytest.mark.parametrize( + "field, value", + [("server", ""), ("tool", None), ("arguments", []), ("result_mode", "raw")], + ) + def test_rejects_invalid_fields(self, field, value): + data = {"server": "fixture", "tool": "echo", field: value} + with pytest.raises(ValueError): + McpStep.from_dict(data) + + def test_runner_preserves_structured_result(self, monkeypatch): + calls = [] + + class FakeManager: + def call_tool(self, server, tool, arguments, result_mode="auto"): + calls.append((server, tool, arguments, result_mode)) + return {"sum": 3} + + monkeypatch.setattr("cmdforge.runner.McpClientManager", FakeManager) + tool = Tool( + name="mcp-add", + steps=[ + McpStep( + server="fixture", + tool="add", + arguments={"a": "{input}", "b": 2}, + output_var="result", + ) + ], + output="{result.sum}", + ) + output, exit_code = run_tool(tool, 1, {}) + assert (output, exit_code) == ("3", 0) + assert calls == [("fixture", "add", {"a": 1, "b": 2}, "auto")] + + +class TestRealMcpSdk: + @pytest.fixture + def manager(self, tmp_path, monkeypatch): + pytest.importorskip("mcp") + config_file = tmp_path / "mcp.yaml" + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file) + fixture = Path(__file__).parent / "fixtures" / "mcp_test_server.py" + save_mcp_config([ + McpServerConfig( + name="fixture", + command=sys.executable, + args=[str(fixture)], + timeout=10, + approved=True, + ) + ]) + return McpClientManager() + + def test_discovery_uses_mcp_handshake_and_tools_list(self, manager): + tools = manager.discover("fixture") + schemas = {tool["name"]: tool for tool in tools} + assert {"echo", "add", "read_env", "current_directory", "wait"}.issubset(schemas) + assert schemas["add"]["inputSchema"]["required"] == ["a", "b"] + + def test_calls_text_and_structured_tools(self, manager): + assert manager.call_tool("fixture", "echo", {"message": "hello"}, "text") == "hello" + assert manager.call_tool("fixture", "add", {"a": 1, "b": 2}) == {"sum": 3} + + def test_parent_secret_is_not_leaked(self, manager, monkeypatch): + monkeypatch.setenv("CMDFORGE_TEST_SECRET", "do-not-leak") + result = manager.call_tool( + "fixture", "read_env", {"name": "CMDFORGE_TEST_SECRET"}, "text" + ) + assert result == "" + + def test_configured_working_directory_is_used(self, manager, tmp_path): + manager.list_servers() + manager._configs["fixture"].cwd = str(tmp_path) + assert manager.call_tool("fixture", "current_directory", {}, "text") == str(tmp_path) + + def test_timeout_is_enforced(self, manager): + manager.list_servers() + manager._configs["fixture"].timeout = 0.5 + started = time.monotonic() + with pytest.raises(TimeoutError, match="timed out"): + manager.call_tool("fixture", "wait", {"seconds": 10}, "text") + assert time.monotonic() - started < 4 + + +class TestMcpCli: + def test_add_accepts_leading_dash_argument(self, tmp_path, monkeypatch, capsys): + from cmdforge.cli import main + + config_file = tmp_path / "mcp.yaml" + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file) + monkeypatch.setattr( + sys, + "argv", + [ + "cmdforge", "mcp", "add", "fixture", "--command", "npx", + "--arg=-y", "--arg", "@scope/server", + ], + ) + assert main() == 0 + loaded = load_mcp_config() + assert loaded[0].args == ["-y", "@scope/server"] + assert loaded[0].approved is True + assert "saved and approved" in capsys.readouterr().out + + def test_bare_mcp_lists_servers(self, tmp_path, monkeypatch, capsys): + from cmdforge.cli import main + + monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", tmp_path / "mcp.yaml") + monkeypatch.setattr(sys, "argv", ["cmdforge", "mcp"]) + assert main() == 0 + assert "No MCP servers configured" in capsys.readouterr().out