Detect stale MCP runtime before mutation tools execute

2026-07-08 13:43:50 -04:00
parent cc63ab74e6
commit c64809134f
3 changed files with 339 additions and 0 deletions
@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""Live health-check script to verify Gitea MCP namespace connections.
+
+Spawns the MCP server processes as defined in the IDE's global config,
+performs the JSON-RPC handshake, and queries the tools list to verify
+that the connection is fully operational and doesn't return EOF.
+"""
+import json
+import os
+import subprocess
+import sys
+
+def test_connection(name, config):
+    print(f"Testing MCP connection for '{name}'...")
+    command = config.get("command")
+    args = config.get("args", [])
+    env = config.get("env", {})
+    
+    # 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 e:
+        print(f"  [FAIL] Failed to spawn process: {e}")
+        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:
+        proc.stdin.write(json.dumps(init_req) + "\n")
+        proc.stdin.flush()
+        
+        # Read response
+        line = proc.stdout.readline()
+        if not line:
+            stderr_content = proc.stderr.read()
+            print(f"  [FAIL] Received EOF from process. Stderr:\n{stderr_content}")
+            proc.terminate()
+            return False
+            
+        print(f"  [OK] Received initialize response: {line.strip()[:150]}...")
+        
+        # Send initialized notification
+        init_notif = {
+            "jsonrpc": "2.0",
+            "method": "notifications/initialized"
+        }
+        proc.stdin.write(json.dumps(init_notif) + "\n")
+        proc.stdin.flush()
+        
+        # Send tools/list request
+        list_req = {
+            "jsonrpc": "2.0",
+            "method": "tools/list",
+            "params": {},
+            "id": 2
+        }
+        proc.stdin.write(json.dumps(list_req) + "\n")
+        proc.stdin.flush()
+        
+        line = proc.stdout.readline()
+        if not line:
+            print("  [FAIL] Received EOF on tools/list request.")
+            proc.terminate()
+            return False
+            
+        res = json.loads(line)
+        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]}...")
+        
+        proc.terminate()
+        return True
+    except Exception as e:
+        print(f"  [FAIL] Error during handshake: {e}")
+        proc.terminate()
+        return False
+
+def main():
+    config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
+    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", {})
+    failed = False
+    for name in ["gitea-author", "gitea-reviewer"]:
+        if name in servers:
+            if not test_connection(name, servers[name]):
+                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()