278 lines
9.1 KiB
Python
278 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Live health-check script to verify Gitea MCP namespace connections.
|
|
|
|
Spawns MCP server processes from the IDE config, performs the JSON-RPC
|
|
handshake, verifies the required tool is registered, and invokes that tool.
|
|
The final invocation is what catches client/namespace EOF failures that a
|
|
static FastMCP registration check cannot see.
|
|
"""
|
|
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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,
|
|
):
|
|
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()
|