130 lines
3.7 KiB
Python
130 lines
3.7 KiB
Python
#!/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()
|