phone_lazy_mcp.py
2.53 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
"""Lazy proxy for android_mcp.py: 3 tools instead of 65 schemas in every request.
The real server (65 tools) is only spawned on the first phone_* call.
ponytail: one shared child session, no reconnect on crash (restart dsh); fine for a phone-local agent.
"""
import asyncio
import json
import sys
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.server.fastmcp import FastMCP
REAL = Path.home() / "pixelphone/event_bus/android_mcp.py"
mcp = FastMCP("android_phone")
_session = None
_ready = asyncio.Event()
async def _owner():
# the stdio/session contexts must be entered and exited by the SAME task (anyio cancel scopes),
# so one long-lived task owns them instead of whichever request came first.
global _session
params = StdioServerParameters(command=sys.executable, args=[str(REAL)], cwd=str(REAL.parent))
async with stdio_client(params) as (r, w), ClientSession(r, w) as s:
await s.initialize()
_session = s
_ready.set()
await asyncio.Event().wait()
_task = None
async def real() -> ClientSession:
global _task
if _task is None:
_task = asyncio.create_task(_owner())
await asyncio.wait_for(_ready.wait(), 90)
return _session
def _text(res) -> str:
return "\n".join(getattr(c, "text", str(c)) for c in res.content)
@mcp.tool()
async def phone_list(keyword: str = "") -> str:
"""List Android-phone tools (name + first description line). Filter by keyword; call phone_describe(name) before phone_call."""
tools = (await (await real()).list_tools()).tools
kw = keyword.lower()
rows = [f"{t.name}: {(t.description or '').strip().splitlines()[0] if t.description else ''}"
for t in tools if kw in t.name.lower() or kw in (t.description or "").lower()]
return "\n".join(rows) or "no match"
@mcp.tool()
async def phone_describe(name: str) -> str:
"""Full description and JSON input schema of one phone tool."""
for t in (await (await real()).list_tools()).tools:
if t.name == name:
return f"{t.description}\n\ninputSchema: {json.dumps(t.inputSchema, ensure_ascii=False)}"
return f"unknown tool: {name} (use phone_list)"
@mcp.tool()
async def phone_call(name: str, arguments: dict | None = None) -> str:
"""Call one phone tool by name with a JSON arguments object."""
res = await (await real()).call_tool(name, arguments or {})
out = _text(res)
return f"[tool error] {out}" if res.isError else out
if __name__ == "__main__":
mcp.run()