diff --git a/mcp_server/mcp_server.py b/mcp_server/mcp_server.py index 11ffb38..1fe77e5 100644 --- a/mcp_server/mcp_server.py +++ b/mcp_server/mcp_server.py @@ -18,6 +18,11 @@ logging.basicConfig(level=logging.INFO, _log = logging.getLogger("mcp") S = config.Settings() + +# 坐标空间缓存:serial -> (display_w, display_h, native_w, native_h) +# de_screenshot 返回的图与 de_tap/de_swipe 的坐标同一空间(display), +# server 按比例换算为设备原生坐标(模型只感知截图坐标系,不感知原生分辨率)。 +_coord = {} audit.init(S.audit_file) mcp = FastMCP("mobile-control") @@ -47,6 +52,16 @@ def _check_write(): raise PlatformError("write_disabled", "写操作未启用(MCP_ALLOW_WRITE=1 开启)") +def _to_native(serial, x, y): + """截图坐标 → 设备原生坐标(按最近一次截图的比例换算)。""" + c = _coord.get(serial) + if not c: + raise PlatformError("invalid_param", + "请先对该设备执行 de_screenshot(需要建立坐标空间)") + dw, dh, nw, nh = c + return (round(x * nw / dw), round(y * nh / dh)) + + def _err(e: PlatformError): return {"ok": False, "error": {"code": e.code, "message": e.message}} @@ -96,40 +111,52 @@ def de_screenshot(serial: str) -> dict: buf = io.BytesIO() img.convert("RGB").save(buf, "JPEG", quality=S.jpeg_quality) data = base64.b64encode(buf.getvalue()).decode() + # 记录坐标空间(display=返回图尺寸,native=设备原生),供 tap/swipe 换算 + nw, nh = platform().screen_size(serial) + _coord[serial] = (img.size[0], img.size[1], nw, nh) except PlatformError as e: return _err(e) except Exception as e: return _err(PlatformError("platform_unavailable", f"截图处理失败: {e}")) audit.audit("de_screenshot", serial, f"{w}x{h}", "ok") + dw, dh, nw, nh = _coord[serial] return _ok({"image": {"type": "image", "data": data, "mimeType": "image/jpeg"}, - "width": w, "height": h, + "width": dw, "height": dh, + "native_size": {"width": nw, "height": nh}, "screen_state": screen_state}) @mcp.tool() def de_tap(serial: str, x: int, y: int) -> dict: - """点击设备屏幕指定坐标(设备原生分辨率像素,与截图 1:1 换算)。""" + """点击设备屏幕指定坐标。 + + 坐标空间 = de_screenshot 返回的图像坐标(display 空间)——先截图拿到 + native_size 后再点击,server 自动换算为设备原生坐标。 + """ try: _check_write() serial = _check_serial(serial) if x < 0 or y < 0: raise PlatformError("invalid_param", "坐标不能为负") - platform().tap(serial, x, y) + nx, ny = _to_native(serial, x, y) + platform().tap(serial, nx, ny) except PlatformError as e: return _err(e) - audit.audit("de_tap", serial, f"({x},{y})", "ok") + audit.audit("de_tap", serial, f"({x},{y})->native({nx},{ny})", "ok") return _ok({"action": "tap", "serial": serial, "x": x, "y": y}) @mcp.tool() def de_swipe(serial: str, x1: int, y1: int, x2: int, y2: int, duration: float = 0.2) -> dict: - """在设备屏幕上滑动(起终点为原生分辨率像素)。""" + """在设备屏幕上滑动(坐标空间同 de_tap:截图坐标,server 换算原生)。""" try: _check_write() serial = _check_serial(serial) - platform().swipe(serial, x1, y1, x2, y2, duration) + nx1, ny1 = _to_native(serial, x1, y1) + nx2, ny2 = _to_native(serial, x2, y2) + platform().swipe(serial, nx1, ny1, nx2, ny2, duration) except PlatformError as e: return _err(e) audit.audit("de_swipe", serial, f"({x1},{y1})->({x2},{y2})", "ok") diff --git a/mcp_server/platform_client.py b/mcp_server/platform_client.py index 3cf4f0d..ff3fb9f 100644 --- a/mcp_server/platform_client.py +++ b/mcp_server/platform_client.py @@ -80,6 +80,19 @@ class PlatformClient: f"截图 HTTP {r.status_code}") return r.content, r.headers.get("X-Screen-State", "unknown") + def screen_size(self, serial): + """屏幕原生分辨率(GET /api/screen/size),返回 (w, h)。""" + r = self._get("/api/screen/size", serial=serial) + if r.status_code == 503: + raise PlatformError("device_offline", r.text[:120]) + if r.status_code != 200: + raise PlatformError("platform_unavailable", + f"分辨率 HTTP {r.status_code}") + j = r.json() or {} + if not j.get("ok"): + raise PlatformError("device_offline", str(j.get("error", "取分辨率失败"))[:120]) + return int(j["width"]), int(j["height"]) + def tap(self, serial, x, y): """点击(POST /api/screen/tap)。""" r = self._post("/api/screen/tap", json={"serial": serial,