#!/usr/bin/env python3 """Offline-only MCP namespace spawn probe (NOT IDE-namespace proof). Spawns a *separate* MCP server process from config via subprocess.Popen, performs the JSON-RPC handshake, verifies the required tool is registered, and invokes that tool. Results are classified with ``probe_source=offline_spawn``. This path is useful for offline launch/registration debugging. It does **not** prove the IDE-managed MCP client namespace is healthy (#543). For workflow gates, pass live IDE call evidence with ``probe_source=client_namespace`` to ``gitea_assess_mcp_namespace_health``. """ import argparse import json import os import subprocess import sys from mcp_namespace_health import REQUIRED_NAMESPACE_TOOLS, classify_namespace_probe def _read_json_line(proc): line = proc.stdout.readline() if not line: stderr_content = proc.stderr.read() return None, stderr_content return json.loads(line), None def _write_message(proc, payload): proc.stdin.write(json.dumps(payload) + "\n") proc.stdin.flush() def run_connection_test(name, config, *, required_tool=None, config_path=None): print(f"Testing MCP connection for '{name}'...") command = config.get("command") args = config.get("args", []) env = config.get("env", {}) tool_name = required_tool or REQUIRED_NAMESPACE_TOOLS.get(name) or "gitea_whoami" # Merge current environment run_env = os.environ.copy() run_env.update(env) # Spawn subprocess try: proc = subprocess.Popen( [command] + args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, env=run_env ) except Exception as exc: print(f" [FAIL] Failed to spawn process: {exc}") assessment = classify_namespace_probe( name, required_tool=tool_name, probe_result={"success": False, "error": str(exc)}, process={"profile": env.get("GITEA_MCP_PROFILE"), "env": env}, config_path=config_path, probe_source="offline_spawn", ) print(f" diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}") return False # Send initialize request init_req = { "jsonrpc": "2.0", "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "healthcheck", "version": "1.0"} }, "id": 1 } try: _write_message(proc, init_req) # Read response res, stderr_content = _read_json_line(proc) if res is None: print(f" [FAIL] Received EOF from process. Stderr:\n{stderr_content}") assessment = classify_namespace_probe( name, required_tool=tool_name, probe_result={"success": False, "error": stderr_content or "EOF"}, process={ "pid": proc.pid, "profile": env.get("GITEA_MCP_PROFILE"), "env": env, }, config_path=config_path, probe_source="offline_spawn", ) print(f" remediation: {' '.join(assessment['remediation'])}") proc.terminate() return False print(f" [OK] Received initialize response: {str(res)[:150]}...") # Send initialized notification init_notif = { "jsonrpc": "2.0", "method": "notifications/initialized" } _write_message(proc, init_notif) # Send tools/list request list_req = { "jsonrpc": "2.0", "method": "tools/list", "params": {}, "id": 2 } _write_message(proc, list_req) res, stderr_content = _read_json_line(proc) if res is None: print(" [FAIL] Received EOF on tools/list request.") assessment = classify_namespace_probe( name, required_tool=tool_name, probe_result={"success": False, "error": stderr_content or "EOF"}, process={ "pid": proc.pid, "profile": env.get("GITEA_MCP_PROFILE"), "env": env, }, config_path=config_path, probe_source="offline_spawn", ) print(f" remediation: {' '.join(assessment['remediation'])}") proc.terminate() return False if "error" in res: print(f" [FAIL] Server returned error: {res['error']}") proc.terminate() return False tools = res.get("result", {}).get("tools", []) tool_names = [t.get("name") for t in tools] print(f" [OK] Successfully retrieved {len(tool_names)} tools: {tool_names[:5]}...") if tool_name not in tool_names: assessment = classify_namespace_probe( name, required_tool=tool_name, registered_tools=tool_names, probe_result={"success": False, "error": "required tool missing"}, process={ "pid": proc.pid, "profile": env.get("GITEA_MCP_PROFILE"), "env": env, }, config_path=config_path, probe_source="offline_spawn", ) print(f" [FAIL] Required tool '{tool_name}' is not registered.") print(f" remediation: {' '.join(assessment['remediation'])}") proc.terminate() return False call_req = { "jsonrpc": "2.0", "method": "tools/call", "params": {"name": tool_name, "arguments": {}}, "id": 3, } _write_message(proc, call_req) call_res, stderr_content = _read_json_line(proc) if call_res is None: assessment = classify_namespace_probe( name, required_tool=tool_name, registered_tools=tool_names, probe_result={"success": False, "error": stderr_content or "EOF"}, process={ "pid": proc.pid, "profile": env.get("GITEA_MCP_PROFILE"), "env": env, }, config_path=config_path, probe_source="offline_spawn", ) print(f" [FAIL] Received EOF on {tool_name} invocation.") print(f" diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}") print(f" remediation: {' '.join(assessment['remediation'])}") proc.terminate() return False if "error" in call_res: assessment = classify_namespace_probe( name, required_tool=tool_name, registered_tools=tool_names, probe_result={"success": False, "error": call_res["error"]}, process={ "pid": proc.pid, "profile": env.get("GITEA_MCP_PROFILE"), "env": env, }, config_path=config_path, probe_source="offline_spawn", ) print(f" [FAIL] {tool_name} invocation returned error: {call_res['error']}") print(f" remediation: {' '.join(assessment['remediation'])}") proc.terminate() return False assessment = classify_namespace_probe( name, required_tool=tool_name, registered_tools=tool_names, probe_result={"success": True, "result": call_res.get("result")}, process={ "pid": proc.pid, "profile": env.get("GITEA_MCP_PROFILE"), "env": env, }, config_path=config_path, probe_source="offline_spawn", ) print(f" [OK] Successfully invoked required tool '{tool_name}'.") print(f" diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}") proc.terminate() return True except Exception as exc: print(f" [FAIL] Error during handshake: {exc}") proc.terminate() return False def main(): parser = argparse.ArgumentParser() parser.add_argument( "--config", default=os.environ.get( "MCP_CONFIG_PATH", os.path.expanduser("~/.gemini/config/mcp_config.json"), ), help="Path to MCP config JSON.", ) parser.add_argument( "--namespace", action="append", dest="namespaces", help="Namespace to test. May be repeated.", ) args = parser.parse_args() config_path = args.config try: with open(config_path) as f: mcp_config = json.load(f) except Exception as e: print(f"Failed to load mcp_config.json: {e}") sys.exit(1) servers = mcp_config.get("mcpServers", {}) namespaces = args.namespaces or list(REQUIRED_NAMESPACE_TOOLS) failed = False for name in namespaces: if name in servers: if not run_connection_test( name, servers[name], required_tool=REQUIRED_NAMESPACE_TOOLS.get(name), config_path=config_path, probe_source="offline_spawn", ): failed = True else: print(f"Server '{name}' not found in mcp_config.json") if failed: sys.exit(1) else: print("All Gitea MCP connection tests passed!") if __name__ == "__main__": main()