83 lines
2.9 KiB
Python
Executable file
83 lines
2.9 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Call a ReVa (Ghidra) MCP tool over plain HTTP.
|
|
|
|
Fallback for when the ReVa MCP client link drops but the server on CT111 is fine.
|
|
Usage:
|
|
tools/reva_call.py <tool-name> '<json-args>'
|
|
tools/reva_call.py --list
|
|
|
|
The API key is never stored here. It is read from, in order:
|
|
$REVA_KEY, then ~/.claude.json, then CT111's reva-lab.properties over ssh.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import urllib.request
|
|
|
|
URL = os.environ.get("REVA_URL", "http://192.168.10.138:8080/mcp/message")
|
|
|
|
|
|
def api_key():
|
|
if os.environ.get("REVA_KEY"):
|
|
return os.environ["REVA_KEY"]
|
|
cfg = os.path.expanduser("~/.claude.json")
|
|
if os.path.exists(cfg):
|
|
m = re.search(r'"X-API-Key"\s*:\s*"([0-9a-f]{16,})"', open(cfg, encoding="utf-8").read())
|
|
if m:
|
|
return m.group(1)
|
|
out = subprocess.run(
|
|
["ssh", "spicy", "pct exec 111 -- grep '^reva.server.options.api.key=' /opt/reva-src/reva-lab.properties"],
|
|
capture_output=True, text=True, timeout=30)
|
|
m = re.search(r"api\.key=(\S+)", out.stdout)
|
|
if m:
|
|
return m.group(1)
|
|
sys.exit("no ReVa API key found (set $REVA_KEY)")
|
|
|
|
|
|
def rpc(key, method, params, session=None, notify=False):
|
|
msg = {"jsonrpc": "2.0", "method": method, "params": params}
|
|
if not notify:
|
|
msg["id"] = 1
|
|
req = urllib.request.Request(URL, data=json.dumps(msg).encode(), headers={
|
|
"X-API-Key": key,
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/event-stream",
|
|
**({"Mcp-Session-Id": session} if session else {}),
|
|
})
|
|
with urllib.request.urlopen(req, timeout=300) as r:
|
|
sid = r.headers.get("Mcp-Session-Id") or session
|
|
raw = r.read().decode("utf-8", "replace").strip()
|
|
if not raw:
|
|
return {}, sid
|
|
# tools/* replies come back as SSE (id:/event:/data: lines); initialize is plain JSON.
|
|
if not raw.startswith("{"):
|
|
raw = "".join(ln[5:].strip() for ln in raw.splitlines() if ln.startswith("data:"))
|
|
return json.loads(raw), sid
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
sys.exit(__doc__)
|
|
key = api_key()
|
|
_, sid = rpc(key, "initialize", {
|
|
"protocolVersion": "2024-11-05", "capabilities": {},
|
|
"clientInfo": {"name": "reva_call", "version": "1"}})
|
|
rpc(key, "notifications/initialized", {}, sid, notify=True)
|
|
if sys.argv[1] == "--list":
|
|
res, _ = rpc(key, "tools/list", {}, sid)
|
|
for t in res.get("result", {}).get("tools", []):
|
|
print(f"{t['name']}: {t.get('description','')[:110]}")
|
|
return
|
|
args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}
|
|
res, _ = rpc(key, "tools/call", {"name": sys.argv[1], "arguments": args}, sid)
|
|
if "error" in res:
|
|
print(json.dumps(res["error"], indent=2))
|
|
sys.exit(1)
|
|
for c in res.get("result", {}).get("content", []):
|
|
print(c.get("text", json.dumps(c)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|