From fab7e2b11a7e372e7d0d3dc321b68ae94fac8074 Mon Sep 17 00:00:00 2001 From: butubb <1422726308@qq.com> Date: Fri, 4 Sep 2026 13:48:55 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20MCP=20=E5=B7=A5=E5=85=B7=E8=A1=A5?= =?UTF-8?q?=E5=85=A8=EF=BC=889=E2=86=9218=EF=BC=89=E2=80=94=E2=80=94de=5Fo?= =?UTF-8?q?pen=5Fapp(adb=20monkey=20=E7=9B=B4=E5=90=AF=EF=BC=8C=E7=9C=81?= =?UTF-8?q?=20token)/de=5Fstop=5Fapp/de=5Fforeground=5Fapp/de=5Ftype=5Ftex?= =?UTF-8?q?t(=E4=B8=AD=E6=96=87=E7=9B=B4=E6=8E=A5=E8=BE=93=E5=85=A5)/de=5F?= =?UTF-8?q?set=5Fclipboard/de=5Fsleep/de=5Focr(=E5=B1=8F=E5=B9=95=E6=96=87?= =?UTF-8?q?=E5=AD=97=E8=AF=86=E5=88=AB)/de=5Flist=5Fapps/de=5Flist=5Ftasks?= =?UTF-8?q?=EF=BC=9Bdirect=5Fops=20=E7=9B=B4=E8=BF=9E=E5=B0=81=E8=A3=85?= =?UTF-8?q?=EF=BC=88adb/u2/OCR/=E5=89=AA=E8=B4=B4=E6=9D=BF=E8=BD=BB?= =?UTF-8?q?=E9=87=8F=E9=80=9A=E9=81=93=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mcp_server/direct_ops.py | 106 ++++++++++++++++++++++++ mcp_server/mcp_server.py | 150 ++++++++++++++++++++++++++++++++++ mcp_server/platform_client.py | 42 ++++++++++ 3 files changed, 298 insertions(+) create mode 100644 mcp_server/direct_ops.py diff --git a/mcp_server/direct_ops.py b/mcp_server/direct_ops.py new file mode 100644 index 0000000..d875500 --- /dev/null +++ b/mcp_server/direct_ops.py @@ -0,0 +1,106 @@ +"""MCP 直连操作封装：adb/u2/OCR/剪贴板——轻量通道优先。 + +为什么直连而不是全部走平台 HTTP： + - adb monkey 打开 App / am force-stop：一条 adb 命令，不建 u2 连接（省时省 token） + - 输入文字：u2 EditText.set_text 直接设文本（比剪贴板+粘贴少依赖） + - OCR：平台 RapidOCR（本地模型，截图即识别） +平台 REST 无法表达的操作（无独立端点）在此封装；安全（白名单/写门控） +仍在 MCP 工具层统一把关。绝不 kill-server / 绝不 disconnect（项目红线）。 +""" +import re +import subprocess + +from config import ADB_PATH + +_log = None # mcp 层 logging 由调用方配置 + + +def _adb(serial, *args, timeout=25): + """对指定设备执行 adb shell 命令，返回输出文本（超时返回空）。""" + try: + r = subprocess.run([ADB_PATH, "-s", serial, "shell", *args], + capture_output=True, timeout=timeout) + return (r.stdout or b"").decode("utf-8", errors="replace") + except subprocess.TimeoutExpired: + return "" + except Exception: + return "" + + +def open_app(serial, package): + """adb monkey 打开 App（无需知道 activity，最轻量）。""" + out = _adb(serial, "monkey", "-p", package, + "-c", "android.intent.category.LAUNCHER", "1") + return out or "" + + +def stop_app(serial, package): + """强制停止 App（am force-stop）。""" + return _adb(serial, "am", "force-stop", package) + + +def foreground_app(serial): + """当前前台 App 包名（dumpsys window 解析；失败返回空）。""" + out = _adb(serial, "dumpsys", "window", timeout=15) + m = re.search(r"mCurrentFocus=.*?([\w.]+)/", out or "") + return m.group(1) if m else "" + + +def list_apps(serial, keyword=""): + """第三方已装应用包名列表（pm list packages -3，可关键词过滤）。""" + out = _adb(serial, "pm", "list", "packages", "-3") + pkgs = [] + for line in (out or "").splitlines(): + p = line.replace("package:", "").strip() + if p and (not keyword or keyword.lower() in p.lower()): + pkgs.append(p) + return pkgs + + +def type_text(serial, text): + """向当前界面输入框输入文字（u2 定位 EditText set_text，支持中文）。 + + 返回 (ok, msg)。 + """ + import uiautomator2 as u2 + d = u2.connect(serial) + # 优先聚焦输入框；找不到则第一个 EditText（与任务 input_text 兜底一致） + try: + el = d(focused=True) + if el.exists: + el.set_text(text) + return True, "已输入到聚焦输入框" + except Exception: + pass + try: + edit = d(className="android.widget.EditText") + if edit.exists: + edit.set_text(text) + return True, "已输入到输入框" + except Exception as e: + return False, f"输入失败: {type(e).__name__}: {str(e)[:100]}" + return False, "未找到输入框（请先点击输入框或提供界面信息）" + + +def set_clipboard(serial, text): + """剪贴板注入（ClipInject 通道，读回验证）。""" + from core.clipboard_helper import inject_clipboard + return inject_clipboard(serial, text) + + +def ocr(serial): + """截屏 + RapidOCR 识别，返回 [{text, score, box}]。""" + import uiautomator2 as u2 + from core.ocr import recognize + d = u2.connect(serial) + img = d.screenshot() + if img is None: + return [] + return recognize(img) + + +def read_clipboard(serial): + """读设备剪贴板（u2）。""" + import uiautomator2 as u2 + d = u2.connect(serial) + return d.clipboard or "" diff --git a/mcp_server/mcp_server.py b/mcp_server/mcp_server.py index 74e0dcc..b22c152 100644 --- a/mcp_server/mcp_server.py +++ b/mcp_server/mcp_server.py @@ -273,6 +273,156 @@ def de_press_key(serial: str, key: str) -> dict: return _ok({"action": "press_key", "serial": serial, "key": key}) +# ================== L2 扩展工具（轻量通道：adb/u2 直连，省 token） ================== +@mcp.tool() +def de_open_app(serial: str, package: str) -> dict: + """打开 App（adb monkey 直启，最快路径）。package 为应用包名，如 com.ss.android.ugc.aweme。""" + try: + _check_write() + serial = _check_serial(serial) + if not package: + raise PlatformError("invalid_param", "缺少包名") + from mcp_server import direct_ops + direct_ops.open_app(serial, package) + except PlatformError as e: + return _err(e) + except Exception as e: + return _err(PlatformError("platform_unavailable", f"打开失败: {str(e)[:100]}")) + audit.audit("de_open_app", serial, package, "ok") + return _ok({"action": "open_app", "package": package}) + + +@mcp.tool() +def de_stop_app(serial: str, package: str) -> dict: + """强制停止 App（am force-stop）。""" + try: + _check_write() + serial = _check_serial(serial) + if not package: + raise PlatformError("invalid_param", "缺少包名") + from mcp_server import direct_ops + direct_ops.stop_app(serial, package) + except PlatformError as e: + return _err(e) + except Exception as e: + return _err(PlatformError("platform_unavailable", f"停止失败: {str(e)[:100]}")) + audit.audit("de_stop_app", serial, package, "ok") + return _ok({"action": "stop_app", "package": package}) + + +@mcp.tool() +def de_foreground_app(serial: str) -> dict: + """查询设备当前前台运行的 App 包名（轻量 dumpsys，不打扰设备）。""" + try: + serial = _check_serial(serial) + from mcp_server import direct_ops + pkg = direct_ops.foreground_app(serial) + except PlatformError as e: + return _err(e) + except Exception as e: + return _err(PlatformError("platform_unavailable", f"查询失败: {str(e)[:100]}")) + audit.audit("de_foreground_app", serial, "", pkg or "未知") + return _ok({"foreground_app": pkg or ""}) + + +@mcp.tool() +def de_type_text(serial: str, text: str) -> dict: + """向设备当前输入框输入文字（支持中文，直接 set_text 不依赖剪贴板）。""" + try: + _check_write() + serial = _check_serial(serial) + if not text: + raise PlatformError("invalid_param", "内容为空") + from mcp_server import direct_ops + ok, msg = direct_ops.type_text(serial, text) + if not ok: + raise PlatformError("device_offline", msg) + except PlatformError as e: + return _err(e) + except Exception as e: + return _err(PlatformError("platform_unavailable", f"输入失败: {str(e)[:100]}")) + audit.audit("de_type_text", serial, f"{text[:30]}...", "ok") + return _ok({"action": "type_text"}) + + +@mcp.tool() +def de_set_clipboard(serial: str, text: str) -> dict: + """写入设备剪贴板（ClipInject 通道，读回验证）。""" + try: + _check_write() + serial = _check_serial(serial) + if not text: + raise PlatformError("invalid_param", "内容为空") + from mcp_server import direct_ops + ok, msg = direct_ops.set_clipboard(serial, text) + if not ok: + raise PlatformError("device_offline", msg) + except PlatformError as e: + return _err(e) + except Exception as e: + return _err(PlatformError("platform_unavailable", f"注入失败: {str(e)[:100]}")) + audit.audit("de_set_clipboard", serial, f"{text[:30]}...", "ok") + return _ok({"action": "set_clipboard"}) + + +@mcp.tool() +def de_sleep(serial: str) -> dict: + """熄灭设备屏幕（运行中任务会中断，慎用）。""" + try: + _check_write() + serial = _check_serial(serial) + platform().sleep(serial) + except PlatformError as e: + return _err(e) + audit.audit("de_sleep", serial, "", "ok") + return _ok({"action": "sleep"}) + + +@mcp.tool() +def de_ocr(serial: str) -> dict: + """OCR 识别当前屏幕文字（图片/画布/WebView 里 UI 树没有的文字也能识别）。 + + 返回 [{text, score}]——搜屏幕关键词后可配合 de_tap_element/de_tap 操作。 + """ + try: + serial = _check_serial(serial) + from mcp_server import direct_ops + results = direct_ops.ocr(serial) + except PlatformError as e: + return _err(e) + except Exception as e: + return _err(PlatformError("platform_unavailable", f"OCR 失败: {str(e)[:100]}")) + slim = [{"text": r["text"], "score": round(r["score"], 2)} for r in results] + audit.audit("de_ocr", serial, "", f"{len(slim)} 条") + return _ok({"count": len(slim), "texts": slim[:100]}) + + +@mcp.tool() +def de_list_apps(serial: str, keyword: str = "") -> dict: + """列出设备第三方已装应用包名（可关键词过滤，如 keyword='douyin' 找抖音）。""" + try: + serial = _check_serial(serial) + from mcp_server import direct_ops + pkgs = direct_ops.list_apps(serial, keyword) + except PlatformError as e: + return _err(e) + except Exception as e: + return _err(PlatformError("platform_unavailable", f"查询失败: {str(e)[:100]}")) + audit.audit("de_list_apps", serial, keyword or "", f"{len(pkgs)} 个") + return _ok({"count": len(pkgs), "packages": pkgs[:200]}) + + +@mcp.tool() +def de_list_tasks() -> dict: + """列出平台任务计划（名称/启用状态/调度），供了解可自动化的任务。""" + try: + tasks = platform().list_tasks() + except PlatformError as e: + return _err(e) + audit.audit("de_list_tasks", "", "", f"{len(tasks)} 个") + return _ok({"tasks": tasks}) + + def main(): _log.info("MCP mobile-control 启动: allow_write=%s port=%s", S.allow_write, S.http_port) diff --git a/mcp_server/platform_client.py b/mcp_server/platform_client.py index 761b457..c14bdea 100644 --- a/mcp_server/platform_client.py +++ b/mcp_server/platform_client.py @@ -126,6 +126,48 @@ class PlatformClient: json={"mode": "on", "serials": [serial]}) return self._check_op(r, "wake") + def sleep(self, serial): + """熄屏（POST /api/device/screen_all mode=off）。""" + r = self._post("/api/device/screen_all", + json={"mode": "off", "serials": [serial]}) + return self._check_op(r, "sleep") + + def list_tasks(self): + """任务计划列表（GET /api/jobs）。""" + r = self._get("/api/jobs") + if r.status_code != 200: + raise PlatformError("platform_unavailable", + f"/api/jobs HTTP {r.status_code}") + j = r.json() or {} + tasks = [] + for t in j.get("jobs") or []: + tasks.append({"id": t.get("id"), "name": t.get("name"), + "task_type": t.get("task_type"), + "enabled": t.get("enabled"), + "schedule": (t.get("schedule") or {}).get("mode", "")}) + return tasks + + def sleep(self, serial): + """熄屏（POST /api/device/screen_all mode=off）。""" + r = self._post("/api/device/screen_all", + json={"mode": "off", "serials": [serial]}) + return self._check_op(r, "sleep") + + def list_tasks(self): + """任务计划列表（GET /api/jobs）。""" + r = self._get("/api/jobs") + if r.status_code != 200: + raise PlatformError("platform_unavailable", + f"/api/jobs HTTP {r.status_code}") + j = r.json() or {} + tasks = [] + for t in j.get("jobs") or []: + tasks.append({"id": t.get("id"), "name": t.get("name"), + "task_type": t.get("task_type"), + "enabled": t.get("enabled"), + "schedule": (t.get("schedule") or {}).get("mode", "")}) + return tasks + def press_key(self, serial, key): """按键（POST /api/screen/key）。""" r = self._post("/api/screen/key",