feat: MCP 工具补全（9→18）——de_open_app(adb monkey 直启，省 token)/de_stop_app/de_foreground_app/de_type_text(中文直接输入)/de_set_clipboard/de_sleep/de_ocr(屏幕文字识别)/de_list_apps/de_list_tasks；direct_ops 直连封装（adb/u2/OCR/剪贴板轻量通道）

2026-09-04 13:48:55 +08:00
parent 1560a13d86
commit fab7e2b11a
3 changed files with 298 additions and 0 deletions
@@ -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 ""
@@ -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)
@@ -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",