test_mcp_rpc.py
3.81 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
"""
Test android_mcp.py via MCP JSON-RPC stdio protocol.
Simulates exactly what Hermes does when it spawns the MCP subprocess.
Usage:
python test_mcp_rpc.py
"""
import json
import subprocess
import sys
import os
PYTHON = "/data/data/com.termux/files/home/.hermes/hermes-agent/venv/bin/python"
MCP_SCRIPT = "/data/data/com.termux/files/home/.hermes/event_bus/android_mcp.py"
EVENT_DB = "/data/data/com.termux/files/home/.hermes/event_bus/events.db"
def send_rpc(proc, method: str, params: dict = None, msg_id: int = 1) -> dict:
msg = {"jsonrpc": "2.0", "id": msg_id, "method": method}
if params:
msg["params"] = params
line = json.dumps(msg) + "\n"
proc.stdin.write(line.encode())
proc.stdin.flush()
raw = proc.stdout.readline()
return json.loads(raw)
def send_notify(proc, method: str, params: dict = None):
"""Send a notification (no id, no response expected)."""
msg = {"jsonrpc": "2.0", "method": method}
if params:
msg["params"] = params
proc.stdin.write((json.dumps(msg) + "\n").encode())
proc.stdin.flush()
def main():
env = os.environ.copy()
env["EVENT_BUS_DB"] = EVENT_DB
env["HERMES_QUIET"] = "1"
print("Spawning android_mcp.py via stdio...")
proc = subprocess.Popen(
[PYTHON, MCP_SCRIPT],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
)
try:
# 1. MCP initialize handshake
resp = send_rpc(proc, "initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "test-client", "version": "0.1"},
}, msg_id=1)
print("\n[1] initialize →", json.dumps(resp.get("result", resp.get("error")), indent=2))
send_notify(proc, "notifications/initialized")
# 2. List tools
resp = send_rpc(proc, "tools/list", {}, msg_id=2)
tools = resp.get("result", {}).get("tools", [])
print(f"\n[2] tools/list → {len(tools)} tools:")
for t in tools:
print(f" • {t['name']}: {t.get('description','')[:60]}")
# 3. Call get_stats
resp = send_rpc(proc, "tools/call", {
"name": "get_stats",
"arguments": {}
}, msg_id=3)
content = resp.get("result", {}).get("content", [])
stats = json.loads(content[0]["text"]) if content else {}
print(f"\n[3] get_stats →")
print(f" total_events={stats.get('total_events')}, "
f"last_type={stats.get('last_event_type')}, "
f"last_app={stats.get('last_event_app')}")
# 4. Call get_latest
resp = send_rpc(proc, "tools/call", {
"name": "get_latest",
"arguments": {"limit": 5}
}, msg_id=4)
content = resp.get("result", {}).get("content", [])
result = json.loads(content[0]["text"]) if content else {}
print(f"\n[4] get_latest(5) → {result.get('count')} events:")
for ev in result.get("events", []):
print(f" [{ev['id']}] {ev['type']} | {ev['app']} | {ev['data']}")
# 5. Call get_events with since_id
last_id = result["events"][-1]["id"] if result.get("events") else 0
resp = send_rpc(proc, "tools/call", {
"name": "get_events",
"arguments": {"since_id": last_id, "limit": 10}
}, msg_id=5)
content = resp.get("result", {}).get("content", [])
result2 = json.loads(content[0]["text"]) if content else {}
print(f"\n[5] get_events(since_id={last_id}) → {result2.get('count')} new events")
print("\n✓ MCP stdio protocol working correctly")
finally:
proc.terminate()
stderr_out = proc.stderr.read().decode()
if stderr_out.strip():
print("\n[stderr]:", stderr_out[:500])
if __name__ == "__main__":
main()