#!/usr/bin/env python3 """Sequential stdlib MCP stdio check; reads only an empty temporary Git repo.""" import argparse import json import os from pathlib import Path import queue import re import subprocess import threading import time def sanitize(value, run_root=None): text = json.dumps(value, ensure_ascii=False) if run_root: text = text.replace(str(Path(run_root).resolve()), "$RUN_ROOT") text = text.replace(str(run_root), "$RUN_ROOT") text = re.sub(r"/Users/[^/\s\\\"]+", "$USER_HOME", text) return json.loads(text) def probe(command, repo, timeout=20): """Require separate initialize/list/call responses and inspect actual content.""" env = {"PATH": os.environ.get("PATH", os.defpath), "LANG": "C.UTF-8", "PYTHONNOUSERSITE": "1", "PYTHONDONTWRITEBYTECODE": "1", "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull} process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, cwd=repo, env=env) lines = queue.Queue() errors = [] transcript = [] def read_stdout(): for line in process.stdout: lines.put(line) lines.put(None) def read_stderr(): errors.extend(process.stderr.readlines()) stdout_thread = threading.Thread(target=read_stdout, daemon=True) stderr_thread = threading.Thread(target=read_stderr, daemon=True) stdout_thread.start() stderr_thread.start() result = {"command": command, "initialize_ok": False, "tools_list_ok": False, "tool_call_ok": False, "success": False} def send(message): process.stdin.write(json.dumps(message) + "\n") process.stdin.flush() def response(request_id): deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError(f"No response for request {request_id}") try: line = lines.get(timeout=remaining) except queue.Empty as exc: raise TimeoutError(f"No response for request {request_id}") from exc if line is None: raise EOFError(f"Server closed stdout before response {request_id}") try: message = json.loads(line) except ValueError as exc: raise ValueError("Non-JSON server stdout") from exc transcript.append(message) if message.get("id") == request_id: if "error" in message: raise RuntimeError(f"JSON-RPC error: {message['error']}") if "result" not in message: raise ValueError("Response has no result") return message["result"] try: send({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "real-mcp-reproduction", "version": "1.0"}}}) initialized = response(1) result["initialize"] = initialized result["initialize_ok"] = bool(initialized.get("protocolVersion") and initialized.get("serverInfo")) if not result["initialize_ok"]: raise ValueError("Incomplete initialize response") send({"jsonrpc": "2.0", "method": "notifications/initialized"}) send({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}) listed = response(2) tools = listed.get("tools", []) result["tools"] = [tool["name"] for tool in tools] result["git_status_schema"] = next((tool.get("inputSchema") for tool in tools if tool.get("name") == "git_status"), None) result["tools_list_ok"] = "git_status" in result["tools"] if not result["tools_list_ok"]: raise ValueError("git_status missing from tools/list") send({"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "git_status", "arguments": {"repo_path": str(repo)}}}) called = response(3) result["tool_call"] = called content = "\n".join(item.get("text", "") for item in called.get("content", []) if item.get("type") == "text") expected_markers = ["On branch main", "No commits yet", "nothing to commit"] result["expected_markers_present"] = { marker: marker in content for marker in expected_markers} result["tool_call_ok"] = (called.get("isError") is not True and all(result["expected_markers_present"].values())) result["success"] = result["tool_call_ok"] except (OSError, EOFError, TimeoutError, RuntimeError, ValueError) as exc: result["client_error"] = f"{type(exc).__name__}: {exc}" finally: try: process.stdin.close() except OSError: pass try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.terminate() try: process.wait(timeout=3) except subprocess.TimeoutExpired: process.kill() process.wait() result["terminated_after_check"] = True stdout_thread.join(timeout=2) stderr_thread.join(timeout=2) result["exit_code"] = process.returncode result["stderr"] = "".join(errors) result["response_ids"] = [message.get("id") for message in transcript] result["list_tools_attribute_error"] = ( "AttributeError: 'Server' object has no attribute 'list_tools'" in result["stderr"]) result["mcp_error_import_failure"] = "cannot import name 'McpError'" in result["stderr"] return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo", required=True, type=Path) parser.add_argument("--run-root", type=Path) parser.add_argument("--output", type=Path) parser.add_argument("--expect", choices=["success", "list-tools-error"], default="success") parser.add_argument("command", nargs=argparse.REMAINDER) args = parser.parse_args() command = args.command[1:] if args.command[:1] == ["--"] else args.command if not command: parser.error("server command required after --") result = sanitize(probe(command, args.repo), args.run_root) output = json.dumps(result, indent=2, ensure_ascii=False) + "\n" if args.output: args.output.write_text(output) print(output, end="") expected = result["success"] if args.expect == "success" else ( result["list_tools_attribute_error"] and not result["initialize_ok"] and result["exit_code"] == 1) return 0 if expected else 1 if __name__ == "__main__": raise SystemExit(main())