butubb
ed9e8bacb1
用户报的"探索完无法点击创建任务"真因：一个 run 的事件原先只有**一条** queue.Queue， 聊天页与建任务页同时开着时两个 EventSource 会**瓜分**它——建任务页的回放卡在中间、 `done` 被聊天页取走 → 永远等不到草稿，页面上自然没有可点的"创建"。 一、修（根因 + 表现） - `web/agent_api.py` 新增 `_Fanout`：**每个订阅者一个专属队列**，多开页面各看各的， 还带单轮事件缓冲（晚订阅/刷新重连也能补齐回放，终止事件一定送达）。 实测两路订阅者收到完全一致的 1039 条事件（含 done）。 - `static/admin/agent.js`：断线重连的兜底订阅也按 `mode` 让开（此前漏了这一处）。 二、补齐上一批的三项 - **「直接创建」**：`POST /api/agent/task_draft/create`（草稿体只在服务端、创建前再校验一次、 成功后清草稿避免重复建）+ 草稿预览里的「✓ 直接创建任务」按钮 + 「探索完直接创建任务」勾选框。 - **草稿沉淀经验/动作**：designer 轮次也走 `_distill_experience/_distill_actions`， 但**只在草稿通过校验时**（没走通的试错不入库，免得把误点当经验）。 - **MCP `de_snapshot`**（第 20 个工具）：截图+元素树一次取齐（省一次来回、不会因界面在动而错位）， 附带 `screen_state`/`unstable`；两套提示词都改为优先用它。 平台侧 `/api/uiauto/snapshot` 随之多返回 `screen_state`。 三、文档 - AI_TASK_GEN §10：§10.3 记两个 bug 的真因与修法、§10.4 三项标完成、§10.5 剩余项。 - AI_CONSOLE（扇出语义、多页面同时看一轮）、API（task_draft/create、snapshot 字段）、 MCP/MCP_DESIGN/staffdeck/README/ARCHITECTURE：工具数 19→20 + de_snapshot 条目。 - backlog：记一条新发现的缺陷——`mcp_server/platform_client._login()` 会把"登录页 200" 当成登录成功（现场进程缺 `MCP_PLATFORM_PASS` 时表现为含糊的 platform_unavailable）。 自测：真机浏览器端到端（勾上"探索完直接创建"）→ 探索 12 步 → 草稿 → 自动建任务成功； `de_snapshot` 直连真机校验；校验器 21 条用例、扇出单元用例、本地工具契约用例全绿。 自测产生的任务/草稿已全部清理（未碰用户既有数据）。
582 lines
23 KiB
Python
582 lines
23 KiB
Python
"""MCP 手机控制 Server（M0：设备列表/截图/点击/滑动）。
|
||
|
||
运行：MCP_ALLOW_WRITE=1 python -m mcp_server.mcp_server
|
||
客户端：Streamable HTTP @ http://<host>:8033/mcp
|
||
"""
|
||
import base64
|
||
import io
|
||
import logging
|
||
|
||
from fastmcp import FastMCP
|
||
from PIL import Image
|
||
|
||
from mcp_server import audit, config
|
||
from mcp_server.platform_client import PlatformClient, PlatformError
|
||
|
||
logging.basicConfig(level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s [%(name)s] %(message)s")
|
||
_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")
|
||
_platform = None
|
||
|
||
|
||
def platform():
|
||
"""惰性初始化平台客户端（启动即登录，失败明确报错）。"""
|
||
global _platform
|
||
if _platform is None:
|
||
_platform = PlatformClient(S.platform_url, S.platform_user,
|
||
S.platform_pass, S.platform_timeout)
|
||
return _platform
|
||
|
||
|
||
def _check_serial(serial):
|
||
"""白名单校验：未配置时限制为平台设备池（enabled）设备。"""
|
||
if not serial:
|
||
raise PlatformError("invalid_param", "缺少 serial")
|
||
if S.allowed_serials and serial not in S.allowed_serials:
|
||
raise PlatformError("device_not_allowed", f"设备 {serial} 不在白名单")
|
||
return serial
|
||
|
||
|
||
def _check_write():
|
||
if not S.allow_write:
|
||
raise PlatformError("write_disabled", "写操作未启用（MCP_ALLOW_WRITE=1 开启）")
|
||
|
||
|
||
# 设备任务占用锁：serial -> (ts, worker_status, task_job)。AI 写操作前检查，
|
||
# 任务 running/connecting 的设备拒绝操作（AI 不与任务抢设备）。5s 缓存。
|
||
_busy_cache = {}
|
||
|
||
|
||
def _ensure_device_free(serial):
|
||
"""写操作前检查设备是否有任务在跑（worker running/connecting → 拒绝）。"""
|
||
import time
|
||
now = time.time()
|
||
c = _busy_cache.get(serial)
|
||
if not c or now - c[0] > 5:
|
||
try:
|
||
devs = platform().list_devices()
|
||
info = next((d for d in devs if d.get("serial") == serial), {})
|
||
c = (now, info.get("worker_status") or "idle",
|
||
info.get("task_job") or "")
|
||
_busy_cache[serial] = c
|
||
except Exception:
|
||
return # 状态查询失败不阻塞（操作失败会另行报错）
|
||
if c[1] in ("running", "connecting"):
|
||
raise PlatformError(
|
||
"device_busy",
|
||
f"设备正在执行任务「{c[2] or '未知'}」——AI 不与任务抢设备，"
|
||
f"任务结束后才能操作（可在平台任务页先停止任务）")
|
||
|
||
|
||
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}}
|
||
|
||
|
||
def _ok(data):
|
||
return {"ok": True, "data": data}
|
||
|
||
|
||
@mcp.tool()
|
||
def de_list_devices() -> dict:
|
||
"""列出可控制设备：名称/地址/在线状态/型号/任务状态/前台 App。
|
||
|
||
返回 [{serial, name, model, online, task_job, worker_status, foreground_app}]。
|
||
`name` 是设备在平台里的名称（设备池里的身份标识）；对你的用户汇报时**优先用名称**
|
||
（多台同型号设备时靠它区分）。后续所有工具仍然用 `serial` 指定设备。
|
||
"""
|
||
try:
|
||
devs = platform().list_devices()
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
out = []
|
||
for d in devs:
|
||
out.append({
|
||
"serial": d.get("serial"),
|
||
"name": d.get("device_name") or "",
|
||
"model": d.get("model") or "",
|
||
"online": bool(d.get("present")),
|
||
"task_job": d.get("task_job") or "",
|
||
"worker_status": d.get("worker_status") or "idle",
|
||
"foreground_app": d.get("foreground_app") or "",
|
||
})
|
||
audit.audit("de_list_devices", "", "", f"{len(out)} 台")
|
||
return _ok(out)
|
||
|
||
|
||
@mcp.tool()
|
||
def de_screenshot(serial: str) -> dict:
|
||
"""截取设备屏幕并返回图像（image/jpeg，宽 ≤540px）。
|
||
|
||
同时返回 {width, height, screen_state}。多模态客户端可直接看图。
|
||
"""
|
||
try:
|
||
serial = _check_serial(serial)
|
||
jpeg, screen_state = platform().screenshot(serial)
|
||
img = Image.open(io.BytesIO(jpeg))
|
||
w, h = img.size
|
||
if w > S.screenshot_width:
|
||
ratio = S.screenshot_width / w
|
||
img = img.resize((S.screenshot_width, int(h * ratio)))
|
||
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": 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:
|
||
"""点击设备屏幕指定坐标（坐标空间 = de_screenshot 的图像坐标）。
|
||
|
||
自动吸附：若该点落在某个可点击元素内，实际点击会改为该元素的中心——
|
||
坐标只需大致对准目标即可（模型视觉定位常有偏差，吸附保证点准）；
|
||
点在空白处则按原坐标点击。返回中的 snapped/label 可核对吸附结果。
|
||
"""
|
||
try:
|
||
_check_write()
|
||
serial = _check_serial(serial)
|
||
_ensure_device_free(serial)
|
||
if x < 0 or y < 0:
|
||
raise PlatformError("invalid_param", "坐标不能为负")
|
||
nx, ny = _to_native(serial, x, y)
|
||
res = platform().tap(serial, nx, ny, snap=True)
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
audit.audit("de_tap", serial,
|
||
f"({x},{y})->native({nx},{ny})"
|
||
+ (f" 吸附[{res.get('label')}]" if res.get("snapped") else ""),
|
||
"ok")
|
||
return _ok({"action": "tap", "serial": serial, "x": x, "y": y,
|
||
"snapped": bool(res.get("snapped")),
|
||
"label": res.get("label") or ""})
|
||
|
||
|
||
@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)
|
||
_ensure_device_free(serial)
|
||
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")
|
||
return _ok({"action": "swipe", "serial": serial})
|
||
|
||
|
||
_KEYS = ("back", "home", "recent", "menu", "power", "volume_up",
|
||
"volume_down", "enter", "delete", "search", "camera")
|
||
|
||
|
||
def _slim_elements(els, limit):
|
||
"""精简元素列表（de_ui_tree / de_snapshot 共用）：去 suggested/深度噪音，
|
||
可点击优先、有文案优先，控制 token。"""
|
||
slim = []
|
||
for e in els:
|
||
slim.append({
|
||
"text": e.get("text", "")[:50],
|
||
"id": e.get("resource_id", "")[:80],
|
||
"desc": e.get("description", "")[:50],
|
||
"class": e.get("class", "").split(".")[-1],
|
||
"clickable": e.get("clickable", "") == "true",
|
||
"bounds": e.get("bounds", ""),
|
||
})
|
||
slim.sort(key=lambda x: (not x["clickable"], not (x["text"] or x["desc"])))
|
||
return slim[:limit]
|
||
|
||
|
||
@mcp.tool()
|
||
def de_ui_tree(serial: str, limit: int = 150) -> dict:
|
||
"""获取当前界面元素树（文本 JSON）：每元素含 text/resource_id/description/class/bounds。
|
||
|
||
可点击元素排在前面（可点性优先）。多数场景不需要读整棵树——直接给
|
||
de_tap_text 一个屏幕上可见的文字即可自动定位点击；本工具用于确认界面
|
||
上有什么、元素文案是否与预想一致。limit 控制返回条数（默认 150，防 token 膨胀）。
|
||
"""
|
||
try:
|
||
serial = _check_serial(serial)
|
||
if limit < 1 or limit > 300:
|
||
raise PlatformError("invalid_param", "limit 需在 1-300 之间")
|
||
els = platform().ui_elements(serial)
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
slim = _slim_elements(els, limit)
|
||
audit.audit("de_ui_tree", serial, "", f"{len(slim)} 元素")
|
||
return _ok({"count": len(slim), "elements": slim})
|
||
|
||
|
||
@mcp.tool()
|
||
def de_snapshot(serial: str, limit: int = 120) -> dict:
|
||
"""一次取齐：屏幕截图 + 元素树（**推荐用它代替 de_screenshot + de_ui_tree**）。
|
||
|
||
那两条是两次独立取数，中间隔着元素树 dump 本身的 1~2 秒——界面只要在动
|
||
（信息流/视频/加载动画），返回的元素坐标就和截图对不上，照它点击会点偏。
|
||
本工具在平台侧用**同一个连接背靠背取**，并额外做双截图校验：
|
||
`unstable=true` 表示"抓取过程中界面在变化"，此时别把元素坐标当准的，
|
||
建议让设备停在静止界面再取一次。
|
||
|
||
返回 image（宽 ≤540 的可视截图，同时建立后续 de_tap/de_swipe 的坐标空间）、
|
||
native_size、screen_state、unstable、cost_ms，以及 elements
|
||
（结构与 de_ui_tree 相同；limit 控制条数，默认 120）。
|
||
"""
|
||
try:
|
||
serial = _check_serial(serial)
|
||
if limit < 1 or limit > 300:
|
||
raise PlatformError("invalid_param", "limit 需在 1-300 之间")
|
||
snap = platform().snapshot(serial)
|
||
raw = snap.get("image") or ""
|
||
b64 = raw.split(",", 1)[1] if "," in raw else raw
|
||
img = Image.open(io.BytesIO(base64.b64decode(b64)))
|
||
nw, nh = img.size
|
||
if nw > S.screenshot_width:
|
||
ratio = S.screenshot_width / nw
|
||
img = img.resize((S.screenshot_width, int(nh * ratio)))
|
||
buf = io.BytesIO()
|
||
img.convert("RGB").save(buf, "JPEG", quality=S.jpeg_quality)
|
||
data = base64.b64encode(buf.getvalue()).decode()
|
||
# 与 de_screenshot 用同一套坐标口径：后续 de_tap/de_swipe 传的是这张图的坐标
|
||
_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}"))
|
||
slim = _slim_elements(snap.get("elements") or [], limit)
|
||
unstable = bool(snap.get("unstable"))
|
||
audit.audit("de_snapshot", serial, f"{len(slim)} 元素",
|
||
"unstable" if unstable else "ok")
|
||
out = {"image": {"type": "image", "data": data, "mimeType": "image/jpeg"},
|
||
"width": img.size[0], "height": img.size[1],
|
||
"native_size": {"width": nw, "height": nh},
|
||
"screen_state": snap.get("screen_state", "unknown"),
|
||
"unstable": unstable,
|
||
"cost_ms": snap.get("cost_ms", 0),
|
||
"count": len(slim), "elements": slim}
|
||
if unstable:
|
||
out["hint"] = ("抓取期间界面在变化（unstable）：元素坐标可能已过时，"
|
||
"让设备停在静止界面再取一次更稳")
|
||
return _ok(out)
|
||
|
||
|
||
@mcp.tool()
|
||
def de_tap_element(serial: str, by: str, value: str, index: int = 1) -> dict:
|
||
"""按元素点击（不需要坐标）：by=text|id|desc|text_contains|desc_contains。
|
||
|
||
text/id/desc 为精确匹配；text_contains/desc_contains 为子串模糊匹配
|
||
（只记得部分文字时用，如 by=text_contains value=搜索）。
|
||
元素驱动操作比坐标可靠（界面变化自适应）；元素不存在时返回错误，
|
||
可改用 de_ui_tree 查元素 / de_tap_text 按屏幕文字点 / de_tap 坐标兜底。
|
||
index 用于多命中取第几个（默认 1）。
|
||
"""
|
||
try:
|
||
_check_write()
|
||
serial = _check_serial(serial)
|
||
_ensure_device_free(serial)
|
||
if by not in ("text", "id", "desc", "text_contains", "desc_contains"):
|
||
raise PlatformError("invalid_param",
|
||
"by 可选 text/id/desc/text_contains/desc_contains")
|
||
if not value or index < 1:
|
||
raise PlatformError("invalid_param", "value 不能为空且 index>=1")
|
||
import uiautomator2 as u2
|
||
d = u2.connect(serial)
|
||
kw = {"text": value} if by == "text" else (
|
||
{"resourceId": value} if by == "id" else (
|
||
{"description": value} if by == "desc" else (
|
||
{"textContains": value} if by == "text_contains"
|
||
else {"descriptionContains": value})))
|
||
if index > 1:
|
||
kw["instance"] = index - 1
|
||
el = d(**kw)
|
||
if not el.exists:
|
||
raise PlatformError("device_offline",
|
||
f"未找到元素（{by}={value}，index={index}）——"
|
||
f"建议 de_ui_tree 查看实际元素或 de_tap 用坐标")
|
||
el.click()
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
except Exception as e:
|
||
return _err(PlatformError("platform_unavailable",
|
||
f"元素点击失败: {type(e).__name__}: {str(e)[:100]}"))
|
||
audit.audit("de_tap_element", serial, f"{by}={value}#{index}", "ok")
|
||
return _ok({"action": "tap_element", "serial": serial,
|
||
"by": by, "value": value, "index": index})
|
||
|
||
|
||
@mcp.tool()
|
||
def de_read_clipboard(serial: str) -> dict:
|
||
"""读取设备当前剪贴板内容（ClipInject/atx-agent 通道读回，M1 起支持）。"""
|
||
try:
|
||
serial = _check_serial(serial)
|
||
import uiautomator2 as u2
|
||
d = u2.connect(serial)
|
||
text = d.clipboard
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
except Exception as e:
|
||
return _err(PlatformError("platform_unavailable",
|
||
f"剪贴板读取失败: {type(e).__name__}: {str(e)[:100]}"))
|
||
audit.audit("de_read_clipboard", serial, "", f"{len(text or '')} 字符")
|
||
return _ok({"clipboard": text or ""})
|
||
|
||
|
||
@mcp.tool()
|
||
def de_wake(serial: str) -> dict:
|
||
"""点亮设备屏幕并解锁（熄屏时先调用它再截图）。"""
|
||
try:
|
||
_check_write()
|
||
serial = _check_serial(serial)
|
||
_ensure_device_free(serial)
|
||
platform().wake(serial)
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
audit.audit("de_wake", serial, "", "ok")
|
||
return _ok({"action": "wake", "serial": serial})
|
||
|
||
|
||
@mcp.tool()
|
||
def de_press_key(serial: str, key: str) -> dict:
|
||
"""按设备按键：back/home/recent/menu/power/enter/delete 等。"""
|
||
try:
|
||
_check_write()
|
||
serial = _check_serial(serial)
|
||
_ensure_device_free(serial)
|
||
if key not in _KEYS:
|
||
raise PlatformError("invalid_param", f"不支持的按键: {key}（可选 {_KEYS}）")
|
||
platform().press_key(serial, key)
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
audit.audit("de_press_key", serial, key, "ok")
|
||
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)
|
||
_ensure_device_free(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)
|
||
_ensure_device_free(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)
|
||
_ensure_device_free(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)
|
||
_ensure_device_free(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)
|
||
_ensure_device_free(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_tap_text(serial: str, text: str) -> dict:
|
||
"""点击屏幕上显示该文字的位置（语义点击：一次调用完成「找到并点击」，无需坐标）。
|
||
|
||
想点带文字的按钮/列表项/标签/链接时用它：text 只需是屏幕上可见文字的
|
||
一部分（子串匹配，如「搜索」「立即购买」）。原生控件直接命中；
|
||
WebView/图片/画布里渲染的文字自动走 OCR 兜底。多命中点第一处（想点
|
||
更靠下的请把文字换独特些）。屏幕确实没有该文字时返回错误提示，
|
||
请截图确认后换关键词。比 de_tap 坐标点击可靠，涉及文字目标时优先使用。
|
||
"""
|
||
try:
|
||
_check_write()
|
||
serial = _check_serial(serial)
|
||
_ensure_device_free(serial)
|
||
if not text or len(text) > 100:
|
||
raise PlatformError("invalid_param", "text 不能为空且 ≤100 字符")
|
||
res = platform().tap_text(serial, text)
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
if not res.get("found"):
|
||
err = PlatformError("text_not_found",
|
||
f"屏幕上未找到文字「{text}」——先 de_screenshot 看当前界面，"
|
||
f"换用屏幕上实际存在的文字；若文字在需滑动后才可见请先滑动")
|
||
audit.audit("de_tap_text", serial, f"「{text[:30]}」", "未找到")
|
||
return _err(err)
|
||
audit.audit("de_tap_text", serial,
|
||
f"「{text[:30]}」via {res.get('method')} @({res.get('x')},{res.get('y')})", "ok")
|
||
return _ok({"action": "tap_text", "serial": serial, "text": text,
|
||
"method": res.get("method"), "matched": res.get("matched") or text,
|
||
"x": res.get("x"), "y": res.get("y")})
|
||
|
||
|
||
@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)
|
||
mcp.run(transport="http", host=S.http_host, port=S.http_port)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|