137 lines
4.7 KiB
Python
137 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Android Fleet MCP server — lets Hermes agents (and any MCP client) drive the
|
|
fleet: see screens, tap, type, run commands, trigger onboarding, broadcast.
|
|
HTTP streamable-http on :8095/mcp. All calls proxy to the godhead REST API.
|
|
"""
|
|
import json
|
|
import urllib.request
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
GODHEAD = "http://127.0.0.1:8080"
|
|
|
|
mcp = FastMCP("android-fleet", host="0.0.0.0", port=8095)
|
|
|
|
def _api(path, method="GET", payload=None, timeout=120):
|
|
data = json.dumps(payload).encode() if payload is not None else None
|
|
req = urllib.request.Request(GODHEAD + path, data=data, method=method,
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
return json.loads(r.read())
|
|
|
|
def _err(e):
|
|
return {"error": str(e)[:200]}
|
|
|
|
@mcp.tool()
|
|
def fleet_devices() -> dict:
|
|
"""List all fleet devices with status, IP, proxy and IG account state."""
|
|
try:
|
|
devs = _api("/api/devices")
|
|
accts = {a["mac"]: a.get("status") for a in _api("/api/accounts")}
|
|
for d in devs:
|
|
d["ig_status"] = accts.get(d["mac"], "none")
|
|
return {"count": len(devs),
|
|
"online": sum(1 for d in devs if d["adb_state"] == "online"),
|
|
"devices": devs}
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def device_shell(mac: str, cmd: str) -> dict:
|
|
"""Run an adb shell command on one device (mac = device MAC address)."""
|
|
try:
|
|
return _api(f"/api/device/{mac}/exec", "POST", {"cmd": cmd})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def device_vision(mac: str, instruction: str) -> dict:
|
|
"""Screenshot the device and ask the vision model (qwen2.5vl:7b on GPU)
|
|
about what's on screen. Use before tapping to find coordinates."""
|
|
try:
|
|
return _api(f"/api/device/{mac}/vision", "POST", {"instruction": instruction})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def device_tap(mac: str, x: int, y: int) -> dict:
|
|
"""Tap screen coordinates on a device."""
|
|
try:
|
|
return _api(f"/api/device/{mac}/tap", "POST", {"x": x, "y": y})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def device_swipe(mac: str, x1: int, y1: int, x2: int, y2: int, ms: int = 300) -> dict:
|
|
"""Swipe on a device (e.g. scroll the feed)."""
|
|
try:
|
|
return _api(f"/api/device/{mac}/swipe", "POST",
|
|
{"x1": x1, "y1": y1, "x2": x2, "y2": y2, "ms": ms})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def device_type(mac: str, text: str) -> dict:
|
|
"""Type text on a device (into the focused field)."""
|
|
try:
|
|
return _api(f"/api/device/{mac}/text", "POST", {"text": text})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def device_launch(mac: str, package: str) -> dict:
|
|
"""Launch an app by package name on a device."""
|
|
try:
|
|
return _api(f"/api/device/{mac}/launch", "POST", {"package": package})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def device_screenshot(mac: str) -> dict:
|
|
"""Take a screenshot; returns a vision description + the PNG is cached
|
|
at /api/device/<mac>/screenshot/latest."""
|
|
try:
|
|
shot = _api(f"/api/device/{mac}/screenshot", "POST", timeout=60)
|
|
return {"saved": shot.get("path"), "view": f"/api/device/{mac}/screenshot/latest"}
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def fleet_broadcast_tap(macs: list[str], x: int, y: int) -> dict:
|
|
"""Tap the same coordinates on many devices at once."""
|
|
try:
|
|
return _api("/api/fleet/action", "POST",
|
|
{"macs": macs, "action": {"kind": "tap", "params": {"x": x, "y": y}}})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def onboard_device(mac: str) -> dict:
|
|
"""Run the full phone+IG onboarding on a device (creates IG account when
|
|
Instagram allows it; rate limits are reported honestly)."""
|
|
try:
|
|
return _api(f"/api/device/{mac}/onboard", "POST", {})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def spawn_devices(count: int) -> dict:
|
|
"""Clone new Android VMs from the template; they auto-register."""
|
|
try:
|
|
return _api("/api/spawn", "POST", {"count": count})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
@mcp.tool()
|
|
def fleet_message(text: str) -> dict:
|
|
"""Broadcast a message via saved IG sessions (requires created accounts)."""
|
|
try:
|
|
return _api("/api/fleet/message", "POST", {"text": text})
|
|
except Exception as e:
|
|
return _err(e)
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(mcp.streamable_http_app(), host="0.0.0.0", port=8095, log_level="warning")
|