From d7b6d83cd8e8cdc78f7254e7bb0cf0dcff482b53 Mon Sep 17 00:00:00 2001 From: butubb <1422726308@qq.com> Date: Fri, 4 Sep 2026 12:47:09 +0800 Subject: [PATCH] =?UTF-8?q?feat(M1):=20MCP=20=E6=88=AA=E5=9B=BE=E5=9D=90?= =?UTF-8?q?=E6=A0=87=E7=A9=BA=E9=97=B4=E7=BB=9F=E4=B8=80=E2=80=94=E2=80=94?= =?UTF-8?q?de=5Fscreenshot=20=E9=99=84=20native=5Fsize=EF=BC=8Cde=5Ftap/de?= =?UTF-8?q?=5Fswipe=20=E6=8E=A5=E5=8F=97=E6=88=AA=E5=9B=BE=E5=9D=90?= =?UTF-8?q?=E6=A0=87=E7=94=B1=20server=20=E6=8C=89=E6=AF=94=E4=BE=8B?= =?UTF-8?q?=E6=8D=A2=E7=AE=97=E5=8E=9F=E7=94=9F=EF=BC=88=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E5=8F=AA=E6=84=9F=E7=9F=A5=E6=88=AA=E5=9B=BE=E5=9D=90=E6=A0=87?= =?UTF-8?q?=E7=B3=BB=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mcp_server/mcp_server.py | 39 +++++++++++++++++++++++++++++------ mcp_server/platform_client.py | 13 ++++++++++++ 2 files changed, 46 insertions(+), 6 deletions(-) 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,