Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 butubb
|
02776b12b8 | ||
|
butubb
|
7017ff3315 | ||
|
butubb
|
937a87b36c | ||
|
butubb
|
d7b6d83cd8 | ||
|
butubb
|
084ed59ad5 | ||
|
butubb
|
2ba9f0c239 | ||
|
butubb
|
c304996c7b |
@@ -0,0 +1,244 @@
|
||||
# MCP 手机控制（Mobile Control MCP Server）设计文档
|
||||
|
||||
> 分支：dev:mcp | 状态：设计稿 | 日期：2026-09-04
|
||||
|
||||
## 1. 背景与目标
|
||||
|
||||
平台（auto_control，生产 220:18050）已具备完整的手机远程控制与自动化能力：设备池管理、u2 控制（tap/swipe/key/输入/剪贴板）、minicap 看屏（截图/流）、uiautodev 元素树、RapidOCR、任务调度与动作编辑器。
|
||||
|
||||
本设计的目标：**新增一个 MCP Server，把平台能力以 MCP 协议暴露给多模态 AI（Claude 等），使 AI 能像人一样"看到手机屏幕 → 理解 → 操作 → 再看到"地闭环控制手机**。
|
||||
|
||||
核心判断：平台能力全部已有，MCP Server 是**薄封装层**——主要工作在：工具规格设计、多模态图像链路、认证与安全、部署。
|
||||
|
||||
## 2. 术语
|
||||
|
||||
| 术语 | 说明 |
|
||||
|---|---|
|
||||
| MCP | Model Context Protocol，模型上下文协议（客户端-服务器，工具调用） |
|
||||
| Tools | MCP 暴露的可调用能力（本设计按 L1/L2/L3 分层） |
|
||||
| serial | 设备标识，如 `192.168.20.66:5555`（设备池唯一主键） |
|
||||
| 平台 | 指 auto_control web 服务（18050）及其 core 层能力 |
|
||||
| 宿主 | MCP Server 运行进程/容器 |
|
||||
|
||||
## 3. 总体架构
|
||||
|
||||
```
|
||||
┌──────────────┐ MCP (Streamable HTTP / stdio) ┌──────────────────┐
|
||||
│ 多模态 AI │ ────────────────────────────────▶ │ MCP Server │
|
||||
│ Claude Desktop│ tools + image content block │ （220 独立进程） │
|
||||
│ Claude Code │ ◀──────────────────────────────── │ mcp/ 包 │
|
||||
└──────────────┘ 截图图像 / JSON 结果 └────────┬─────────┘
|
||||
│ 内部调用
|
||||
▼
|
||||
┌──────────────────┐
|
||||
│ auto_control │
|
||||
│ 18050 REST API │
|
||||
│ （登录会话 token）│
|
||||
├──────────────────┤
|
||||
│ core 层能力： │
|
||||
│ u2 / minicap / │
|
||||
│ uiautodev / OCR /│
|
||||
│ 任务调度 / 设备池 │
|
||||
└──────────────────┘
|
||||
```
|
||||
|
||||
- **图像方向**：MCP 协议支持 `image` content block（base64），截图以此返回，多模态模型原生可读——这是"AI 看到手机"的关键链路。
|
||||
- **控制方向**：AI 返回结构化工具调用（tap/swipe/type…），MCP Server 转发平台执行。
|
||||
- **循环**：截图 → 模型视觉推理 → 工具调用 → 平台执行 → 截图验证。
|
||||
|
||||
## 4. 平台现有能力复用清单
|
||||
|
||||
| 平台能力 | 现有入口 | MCP 复用方式 |
|
||||
|---|---|---|
|
||||
| 设备列表/状态 | `GET /api/status`、`/api/devices/pool` | MCP server 内部 HTTP 调用（带会话） |
|
||||
| 截图 | `GET /api/screen/thumb?serial=`（单帧 JPEG） | 直接转发为 image block（亦可直连 core u2 screenshot） |
|
||||
| 看屏流 | `/api/screen/stream`（MJPEG） | M0-M2 不需要；M3 可选（视频理解场景） |
|
||||
| 点击 | `POST /api/screen/tap {serial,x,y}` | 封装 tool `tap` |
|
||||
| 滑动 | `POST /api/screen/swipe` | 封装 `swipe` |
|
||||
| 按键 | `POST /api/screen/key` | 封装 `press_key` |
|
||||
| 输入文字 | ClipInject 通道（core.clipboard_helper）+ u2 input | 封装 `type_text` |
|
||||
| 剪贴板 | `POST /api/tools/clipboard/set`（ClipInject 通道） | 封装 `set_clipboard` |
|
||||
| 元素树 | `GET /api/uiauto/...`（uiautodev） | 封装 `get_ui_tree` |
|
||||
| OCR | core.ocr（RapidOCR） | 封装 `ocr_screen` |
|
||||
| 打开 App | u2 `app_start`（经任务层/直连） | 封装 `open_app`（走 core device 直连） |
|
||||
| 设备在线状态/亮熄屏 | `/api/device/screen_all`、dumpsys | 封装 `screen_state`/`wake` |
|
||||
|
||||
> 决策：MCP Server 优先走 **HTTP API**（薄封装，认证简单、与平台解耦、平台安全逻辑全复用）。对延迟敏感且 HTTP 无入口的能力（u2 直连截图/输入），经平台 core 模块进程内调用或新增少量只读端点，不绕过平台安全层。
|
||||
|
||||
## 5. MCP Server 设计
|
||||
|
||||
### 5.1 技术栈
|
||||
|
||||
- 语言/运行时：Python 3.11（与平台一致）
|
||||
- 框架：**FastMCP**（`fastmcp`，官方 SDK 之上，装饰器式 tools，自带 Streamable HTTP/stdio 双传输）
|
||||
- 依赖：`fastmcp`、`httpx`、`Pillow`（图像处理）、`mcp[cli]`
|
||||
- 日志：logging → 平台同款格式（时间/级别/模块）
|
||||
|
||||
### 5.2 进程与容器
|
||||
|
||||
- 独立容器 `mcp-server`（220 docker-compose 追加），image `python:3.11-slim`
|
||||
- 挂载：无数据挂载（无状态，配置走环境变量）；网络 host 或独立端口（**候选：8033**，避免与 18050/18051 冲突）
|
||||
- 独立于 auto_control 重启，互不阻塞；MCP Server 崩溃不影响平台，平台不可用时 MCP tools 返回明确错误
|
||||
|
||||
### 5.3 配置（环境变量）
|
||||
|
||||
| 变量 | 默认 | 说明 |
|
||||
|---|---|---|
|
||||
| `MCP_PLATFORM_URL` | `http://127.0.0.1:18050` | 平台地址 |
|
||||
| `MCP_PLATFORM_USER` / `MCP_PLATFORM_PASS` | 空 | 平台登录账号（admin） |
|
||||
| `MCP_ALLOW_WRITE` | `0` | 写操作总开关（0=只读感知，1=可操作） |
|
||||
| `MCP_ALLOWED_SERIALS` | 空=全部 | 设备白名单（逗号分隔；为空时自动=设备池内设备） |
|
||||
| `MCP_HTTP_PORT` | `8033` | HTTP 传输端口 |
|
||||
| `MCP_SCREENSHOT_WIDTH` | `540` | 截图宽度（等比缩放，控图像 token 成本） |
|
||||
| `MCP_JPEG_QUALITY` | `70` | 截图 JPEG 质量 |
|
||||
| `MCP_AUDIT_FILE` | `/var/log/mcp/audit.log` | 审计日志路径 |
|
||||
|
||||
## 6. Tools 规格
|
||||
|
||||
命名空间 `de_`（device）前缀避免与常见工具冲突。全部工具对**未配置白名单/离线设备**返回明确错误，不做静默跳过。
|
||||
|
||||
### L1 感知层
|
||||
|
||||
#### `de_list_devices`
|
||||
- 描述：列出可控制设备及其状态（在线/任务/型号/前台 App）
|
||||
- 返回：`[{serial, model, online, task_job, worker_status, foreground_app, screen_state}]`
|
||||
- 错误：平台不可达 → `platform_unavailable`
|
||||
|
||||
#### `de_screenshot(serial: str) -> image`
|
||||
- 描述：截取设备当前屏幕（JPEG），返回 image content block；同时返回 `width/height/serial` 元信息
|
||||
- 实现：平台 `/api/screen/thumb`（X-Screen-State 头复用判断亮熄屏）；失败（离线/超时）→ 明确错误
|
||||
- 图像规格：宽 ≤ `MCP_SCREENSHOT_WIDTH`（默认 540），质量 `MCP_JPEG_QUALITY`；**需保证图像方向正确**（设备可能横竖屏，含 EXIF 或由调用方按截图尺寸推断）
|
||||
- 限制：截图频率 ≥ 1s/次（防 AI 疯狂截图），可配置
|
||||
|
||||
#### `de_ui_tree(serial: str) -> str`
|
||||
- 描述：获取当前界面元素树（扁平 JSON：resource-id/text/content-desc/class/bounds），供定位
|
||||
- 返回：文本 JSON（大树截断至 `MCP_UI_TREE_MAX` 字符，默认 20000，防 context 爆炸）
|
||||
- 空界面/无元素 → 明确错误
|
||||
|
||||
#### `de_ocr(serial: str) -> str`
|
||||
- 描述：OCR 识别当前屏幕文字，返回 `[{text, score, box}]`
|
||||
- 场景：图片/画布/WebView 渲染文字（UI 树里没有的）
|
||||
|
||||
#### `de_screen_state(serial: str) -> str`
|
||||
- 描述：亮屏/熄屏/未知
|
||||
|
||||
### L2 操作层（受 `MCP_ALLOW_WRITE=1` 门控）
|
||||
|
||||
#### `de_tap(serial, x: int, y: int)`
|
||||
- 坐标：设备**原生分辨率**像素（与截图 1:1 换算——调用方用截图尺寸按比例换算，server 不做猜测）
|
||||
|
||||
#### `de_swipe(serial, x1,y1,x2,y2, duration: float=0.2)`
|
||||
|
||||
#### `de_press_key(serial, key: str)`
|
||||
- key ∈ back/home/recent/menu/power/enter/delete…
|
||||
|
||||
#### `de_type_text(serial, text: str, clear_first: bool=True)`
|
||||
- 实现：优先 ClipInject 通道（`core.clipboard_helper.inject_clipboard` 语义：写入剪贴板 + paste 到焦点输入框），失败回退 u2 input
|
||||
- 中文/emoji 全支持（ClipInject 实测通过）
|
||||
|
||||
#### `de_set_clipboard(serial, text: str)`
|
||||
- 写入设备剪贴板（ClipInject am start 通道），读回验证
|
||||
|
||||
#### `de_open_app(serial, package: str)`
|
||||
- 打开指定 App（u2 app_start）；`package` 需在已知清单或完全限定名
|
||||
|
||||
#### `de_wake(serial)` / `de_sleep(serial)`
|
||||
|
||||
### L3 语义层（M2 里程碑）
|
||||
|
||||
#### `de_find_and_tap(serial, target: str, by: "text"|"id"|"ocr"|"desc"=...)`
|
||||
- UI 树/OCR 定位含 target 的元素 → 自动点击其中心；多命中返回候选列表让 AI 选择
|
||||
- 复用平台元素选择器策略（属性唯一/重复序号消歧）
|
||||
|
||||
#### `de_wait_until(serial, condition: str, timeout: int=15)`
|
||||
- 轮询 OCR/UI 树直到出现条件文本（或消失），返回命中的坐标/文本
|
||||
|
||||
#### `de_get_task_state(serial)`（若二期接任务系统）
|
||||
- 设备当前任务状态；可选 `de_run_task(task_id)` 需单独授权
|
||||
|
||||
### 工具返回约定
|
||||
- 全部工具返回结构化 JSON：`{ok: true, data: ...}` 或 `{ok: false, error: {code, message}}`
|
||||
- 错误码：`platform_unavailable` / `device_offline` / `device_not_allowed` / `invalid_param` / `write_disabled` / `busy`（设备被任务占用）
|
||||
- image 返回：`{ok:true, image: <content block>, width, height}`
|
||||
|
||||
## 7. 图像链路细节
|
||||
|
||||
1. 平台截图（`/api/screen/thumb`）→ JPEG 字节
|
||||
2. MCP Server 校验尺寸 → 等比缩放到 `MCP_SCREENSHOT_WIDTH`（用 Pillow，不重压缩已够质量则跳过）
|
||||
3. 组装 MCP `image` content block（`{"type":"image","data":<base64>,"mimeType":"image/jpeg"}`）
|
||||
4. 客户端（Claude）原生把 image 传入视觉上下文
|
||||
|
||||
性能预算：540px JPEG ~40-80KB/张；模型一次多步操作 5-15 张 ≈ 1MB 以内，可接受。
|
||||
|
||||
## 8. 认证与安全
|
||||
|
||||
1. **平台会话**：MCP Server 启动时用 `MCP_PLATFORM_USER/PASS` 登录平台拿会话（Cookie），会话失效自动重登；不向客户端暴露平台凭据
|
||||
2. **写操作门控**：`MCP_ALLOW_WRITE=0`（默认）时 L2/L3 操作全部拒绝（`write_disabled`）——先部署只读，验证感知链路后再开写
|
||||
3. **设备白名单**：`MCP_ALLOWED_SERIALS` 指定；为空时自动限制为**设备池 enabled 设备**（平台口径）
|
||||
4. **占用互斥**：操作前查设备 `worker_status`——running/connecting 中拒绝操作（`busy`），避免 MCP 与任务打架
|
||||
5. **审计**：每次调用（含只读）写审计日志：`ts|tool|serial|args摘要|result`；落盘 `MCP_AUDIT_FILE`
|
||||
6. **传输安全**：内网部署（host 网络 8033）默认无 TLS；若外网暴露需前置 TLS/网络隔离（平台红线：设备控制接口不进公网）
|
||||
7. **无状态**：MCP Server 不存设备数据，全量透传平台
|
||||
|
||||
## 9. 部署（220 docker-compose 追加）
|
||||
|
||||
```yaml
|
||||
mcp-server:
|
||||
container_name: mcp-server
|
||||
image: dockerproxy.net/library/python:3.11-slim
|
||||
restart: unless-stopped
|
||||
working_dir: /app
|
||||
volumes:
|
||||
- "./mcp_server:/app"
|
||||
command: sh -c "pip install --no-cache-dir -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt && python -m mcp_server"
|
||||
environment:
|
||||
- MCP_PLATFORM_URL=http://127.0.0.1:18050
|
||||
- MCP_PLATFORM_USER=admin
|
||||
- MCP_PLATFORM_PASS=${MCP_PLATFORM_PASS} # .env 提供
|
||||
- MCP_ALLOW_WRITE=1
|
||||
- MCP_ALLOWED_SERIALS=
|
||||
network_mode: host
|
||||
```
|
||||
|
||||
客户端接入（Claude Code / Desktop）：
|
||||
|
||||
```json
|
||||
// .mcp.json / claude_desktop_config.json
|
||||
{ "mcpServers": { "mobile": {
|
||||
"url": "http://192.168.20.220:8033/mcp",
|
||||
"transport": "streamable-http"
|
||||
}}}
|
||||
```
|
||||
|
||||
## 10. 里程碑与验收
|
||||
|
||||
| 里程碑 | 内容 | 验收标准 |
|
||||
|---|---|---|
|
||||
| M0 | Server 骨架 + `de_list_devices` + `de_screenshot` + `de_tap`/`de_swipe`（HTTP 传输、登录会话、审计骨架） | Claude Code 连上后：能列设备、截图（AI 能描述屏幕内容）、点击指定坐标生效 |
|
||||
| M1 | L1 完备（ui_tree/ocr/screen_state）+ L2 全量（key/type/clipboard/open_app/wake） | AI 完成「打开抖音搜索『奚学东』」多步操作 |
|
||||
| M2 | L3 语义层（find_and_tap/wait_until）+ 图像尺寸/性能调优 | AI 用自然语言指令（含中文输入）完成跨 3 步以上操作且自适应页面变化 |
|
||||
| M3 | 生产化：安全加固复核、审计看板、部署脚本、接入文档、任务系统互斥回归 | 日常稳定使用 1 周无异常 |
|
||||
|
||||
## 11. 风险与开放问题
|
||||
|
||||
1. **AI 误操作**：模型点击错位/误触关键按钮（如删除）——缓解：L3 定位优先于裸坐标、写门控先行、审计可追溯；**不做**二次确认（破坏自动化体验），靠白名单+占用互斥兜底
|
||||
2. **截图成本**：长会话多截图 → token/延迟成本——缓解：尺寸/质量可调、截图上限、`de_ui_tree` 作为低成本替代（文本树比图像便宜）
|
||||
3. **横竖屏/分辨率异构**：设备多样 → 坐标换算依赖截图 1:1；`de_find_and_tap`（语义层）不受分辨率影响
|
||||
4. **图像 content block 兼容性**：Claude 系原生支持；其他多模态客户端需验证（M0 用 Claude Code 验证）
|
||||
5. **是否接任务系统**：二期决策——`de_run_task` 语义与"单设备操作"不同（任务面向多设备调度），倾向二期用独立命名空间
|
||||
6. **clipboard/input 通道差异**：个别 MIUI 需授权 ClipInject——M0 阶段在受控设备集验证，必要时 fallback 链已在 core 层
|
||||
|
||||
## 12. 附录：平台 API 映射（MCP → 平台）
|
||||
|
||||
| MCP tool | 平台调用 |
|
||||
|---|---|
|
||||
| de_screenshot | GET /api/screen/thumb?serial= |
|
||||
| de_tap / de_swipe / de_press_key | POST /api/screen/tap|swipe|key |
|
||||
| de_type_text / de_set_clipboard | POST /api/tools/clipboard/set（ClipInject）+ u2 input |
|
||||
| de_ui_tree | GET /api/uiauto/…（或 core 直连） |
|
||||
| de_ocr | core.ocr（进程内或新只读端点） |
|
||||
| de_list_devices | GET /api/status + /api/devices/pool |
|
||||
| de_open_app | core device app_start（或任务层 open_app 动作） |
|
||||
| de_screen_state | dumpsys（core.common._device_screen_state） |
|
||||
|
||||
> 实现时对平台缺失的只读入口（如 ocr/open_app 的直连面）优先在平台加**只读端点**（复用现有权限体系），MCP 不越权直连设备。
|
||||
@@ -0,0 +1 @@
|
||||
"""Agent 编排层：第三方 LLM（OpenAI 兼容）经 MCP 工具控制手机。"""
|
||||
@@ -0,0 +1,149 @@
|
||||
"""Agent 编排层：OpenAI 兼容模型（DeepSeek 等）经 MCP 工具控制手机。
|
||||
|
||||
工作流：
|
||||
1. 启动时从 MCP Server 拉工具列表 → 转 OpenAI function schema
|
||||
2. run(prompt)：循环 chat/completions
|
||||
- 模型返回 tool_calls → 依次执行（经 MCP）→ 结果回喂
|
||||
- de_screenshot 的返回图像转为 image_url 追加为下一轮 user 消息（多模态看图）
|
||||
- 无 tool_calls → 返回最终文本
|
||||
"""
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
from fastmcp import Client
|
||||
|
||||
from mcp_agent.config import AgentSettings
|
||||
|
||||
_log = logging.getLogger("agent")
|
||||
|
||||
S = AgentSettings()
|
||||
|
||||
SYSTEM_PROMPT = """你是手机自动化控制助手。你通过工具实时操作 Android 手机。
|
||||
|
||||
工作规范：
|
||||
1. 先 de_list_devices 确定目标设备（在线才可操作）
|
||||
2. 观察屏幕：先 de_screenshot 获取截图（图像会随后给你），基于截图理解当前界面
|
||||
3. 操作：de_tap/de_swipe 的坐标必须与最近一次 de_screenshot 图像一致（直接看图给坐标，服务器自动换算）
|
||||
4. 每次关键操作后再次 de_screenshot 验证结果，直到完成用户目标
|
||||
5. 完成或失败时用中文总结：做了什么、当前状态、需要用户注意的事项
|
||||
6. 设备不可用/操作失败时如实报告错误，不要臆测成功
|
||||
|
||||
可用工具清单将由系统提供。"""
|
||||
|
||||
|
||||
class Agent:
|
||||
def __init__(self, settings: AgentSettings = None):
|
||||
self.s = settings or S
|
||||
self.tools_schema = [] # OpenAI function schema
|
||||
self._tool_exec = {} # name -> callable
|
||||
self.messages = []
|
||||
|
||||
# ---------- MCP 工具桥 ----------
|
||||
async def _load_tools(self):
|
||||
"""从 MCP Server 拉工具，转 OpenAI function schema。"""
|
||||
self._mcp = Client(self.s.mcp_url)
|
||||
await self._mcp.__aenter__()
|
||||
tools = await self._mcp.list_tools()
|
||||
self.tools_schema = []
|
||||
for t in tools:
|
||||
# MCP SDK v2 改名 input_schema，兼容新旧字段
|
||||
schema = getattr(t, "input_schema", None) or getattr(t, "inputSchema", {})
|
||||
# fastmcp Tool 属性兼容：name/description/inputSchema
|
||||
name = getattr(t, "name", "")
|
||||
desc = getattr(t, "description", "") or ""
|
||||
self.tools_schema.append({
|
||||
"type": "function",
|
||||
"function": {"name": name, "description": desc,
|
||||
"parameters": schema}})
|
||||
self._tool_exec[name] = t
|
||||
_log.info("MCP 工具已加载: %s", [s["function"]["name"] for s in self.tools_schema])
|
||||
|
||||
async def close(self):
|
||||
if getattr(self, "_mcp", None):
|
||||
await self._mcp.__aexit__(None, None, None)
|
||||
|
||||
# ---------- 模型调用 ----------
|
||||
async def _chat(self):
|
||||
"""调用 OpenAI 兼容 chat/completions，返回完整 response JSON。"""
|
||||
body = {
|
||||
"model": self.s.model,
|
||||
"messages": self.messages,
|
||||
"tools": self.tools_schema if self.tools_schema else None,
|
||||
"max_tokens": 4096,
|
||||
}
|
||||
headers = {"Authorization": f"Bearer {self.s.api_key}",
|
||||
"Content-Type": "application/json"}
|
||||
async with httpx.AsyncClient(timeout=self.s.request_timeout) as client:
|
||||
r = await client.post(f"{self.s.api_base.rstrip('/')}/chat/completions",
|
||||
json=body, headers=headers)
|
||||
if r.status_code != 200:
|
||||
raise RuntimeError(f"模型 API HTTP {r.status_code}: {r.text[:300]}")
|
||||
return r.json()
|
||||
|
||||
# ---------- 工具执行 ----------
|
||||
async def _execute_tool(self, name, arguments):
|
||||
"""执行 MCP 工具，返回 (文本结果, image_data_or_None)。"""
|
||||
args = json.loads(arguments) if isinstance(arguments, str) else (arguments or {})
|
||||
_log.info("执行工具 %s %s", name, args)
|
||||
try:
|
||||
result = await self._mcp.call_tool(name, args)
|
||||
data = getattr(result, "data", result)
|
||||
except Exception as e:
|
||||
return {"ok": False, "error": f"工具执行失败: {e}"}, None
|
||||
# de_screenshot：图像分离（作为 image_url 追加给模型看）
|
||||
if name == "de_screenshot" and isinstance(data, dict) and data.get("ok"):
|
||||
img = (data.get("data") or {}).get("image") or {}
|
||||
if img.get("data"):
|
||||
text_result = {k: v for k, v in (data.get("data") or {}).items()
|
||||
if k != "image"}
|
||||
return text_result, img["data"]
|
||||
return data, None
|
||||
|
||||
# ---------- 主循环 ----------
|
||||
async def run(self, prompt: str, serial: str = "") -> str:
|
||||
"""执行一轮指令，返回最终回答文本。"""
|
||||
target = serial or self.s.default_serial
|
||||
sys_txt = SYSTEM_PROMPT
|
||||
if target:
|
||||
sys_txt += f"\n\n本次默认目标设备 serial：{target}（未指定设备时用它）。"
|
||||
self.messages = [{"role": "system", "content": sys_txt},
|
||||
{"role": "user", "content": prompt}]
|
||||
|
||||
for step in range(self.s.max_steps):
|
||||
resp = await self._chat()
|
||||
choice = (resp.get("choices") or [{}])[0]
|
||||
msg = choice.get("message") or {}
|
||||
|
||||
# 1) 工具调用
|
||||
tool_calls = msg.get("tool_calls")
|
||||
if tool_calls:
|
||||
self.messages.append({
|
||||
"role": "assistant",
|
||||
"content": msg.get("content") or "",
|
||||
"tool_calls": tool_calls})
|
||||
for tc in tool_calls:
|
||||
fn = tc.get("function") or {}
|
||||
name = fn.get("name", "")
|
||||
text_result, image_b64 = await self._execute_tool(
|
||||
name, fn.get("arguments", "{}"))
|
||||
self.messages.append({
|
||||
"role": "tool",
|
||||
"tool_call_id": tc.get("id", ""),
|
||||
"content": json.dumps(text_result, ensure_ascii=False)[:4000]})
|
||||
# 截图图像：作为下一轮 user 图像内容（OpenAI 协议 tool 结果只能文本）
|
||||
if image_b64:
|
||||
self.messages.append({
|
||||
"role": "user",
|
||||
"content": [{"type": "text",
|
||||
"text": "这是最新屏幕截图，请基于它继续判断"},
|
||||
{"type": "image_url",
|
||||
"image_url": {"url":
|
||||
f"data:image/jpeg;base64,{image_b64}"}}]})
|
||||
continue
|
||||
|
||||
# 2) 最终回答
|
||||
return msg.get("content") or "（模型无输出）"
|
||||
|
||||
return "（达到最大步骤数未完成，请检查操作是否卡在循环）"
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Agent CLI：命令行给 AI 下指令控制手机。
|
||||
|
||||
用法：
|
||||
export AGENT_API_KEY=sk-xxx # DeepSeek API Key
|
||||
export AGENT_DEFAULT_SERIAL=192.168.20.66:5555 # 可选：默认设备
|
||||
python -m mcp_agent.cli "打开抖音，搜索奚学东，截个图"
|
||||
python -m mcp_agent.cli -s 192.168.20.66:5555 "打开微信"
|
||||
python -m mcp_agent.cli # 交互模式（exit 退出）
|
||||
"""
|
||||
import argparse
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
|
||||
from mcp_agent.agent import Agent
|
||||
|
||||
logging.basicConfig(level=logging.INFO,
|
||||
format="%(asctime)s %(levelname)s [%(name)s] %(message)s")
|
||||
_log = logging.getLogger("cli")
|
||||
|
||||
|
||||
async def _run_once(prompt, serial):
|
||||
if not os.environ.get("AGENT_API_KEY") and not os.environ.get("DEEPSEEK_API_KEY"):
|
||||
print("❌ 未配置 API Key：export AGENT_API_KEY=sk-xxx")
|
||||
return
|
||||
agent = Agent()
|
||||
try:
|
||||
await agent._load_tools()
|
||||
print(f"🤖 指令: {prompt}\n")
|
||||
answer = await agent.run(prompt, serial)
|
||||
print(f"\n✅ 结果:\n{answer}")
|
||||
finally:
|
||||
await agent.close()
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="MCP 手机控制 Agent CLI")
|
||||
ap.add_argument("prompt", nargs="?", default="", help="指令（不填则交互模式）")
|
||||
ap.add_argument("-s", "--serial", default="", help="目标设备 serial")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.prompt:
|
||||
asyncio.run(_run_once(args.prompt, args.serial))
|
||||
return
|
||||
|
||||
print("交互模式：输入指令（如「打开抖音搜索奚学东」），exit 退出")
|
||||
while True:
|
||||
try:
|
||||
prompt = input("\n指令> ").strip()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
break
|
||||
if not prompt:
|
||||
continue
|
||||
if prompt.lower() in ("exit", "quit", "退出"):
|
||||
break
|
||||
asyncio.run(_run_once(prompt, args.serial))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,30 @@
|
||||
"""Agent 层配置（第三方 LLM API，OpenAI 兼容格式）。
|
||||
|
||||
DeepSeek 官方 API：https://api.deepseek.com（OpenAI 兼容）。
|
||||
生产用 .env 注入 DEEPSEEK_API_KEY，不要提交 git。
|
||||
"""
|
||||
import os
|
||||
|
||||
|
||||
def _env(key, default):
|
||||
return os.environ.get(key, default)
|
||||
|
||||
|
||||
class AgentSettings:
|
||||
# 模型 API（OpenAI 兼容）
|
||||
api_base = _env("AGENT_API_BASE", "https://api.deepseek.com")
|
||||
api_key = _env("AGENT_API_KEY", _env("DEEPSEEK_API_KEY", ""))
|
||||
model = _env("AGENT_MODEL", "deepseek-v4-flash-vision-exp")
|
||||
|
||||
# MCP Server（工具源）
|
||||
mcp_url = _env("AGENT_MCP_URL", "http://127.0.0.1:8033/mcp")
|
||||
|
||||
# 默认目标设备（命令行不指定 serial 时用它；空则让模型先 de_list_devices）
|
||||
default_serial = _env("AGENT_DEFAULT_SERIAL", "")
|
||||
|
||||
# Agent 循环上限与请求超时
|
||||
max_steps = int(_env("AGENT_MAX_STEPS", "20"))
|
||||
request_timeout = float(_env("AGENT_TIMEOUT", "120"))
|
||||
|
||||
# system prompt 语言
|
||||
language = _env("AGENT_LANG", "zh")
|
||||
@@ -0,0 +1 @@
|
||||
"""MCP 手机控制 Server 包。"""
|
||||
@@ -0,0 +1,27 @@
|
||||
"""审计日志：每次 MCP 调用记录一行（含只读）。"""
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
_file = None
|
||||
|
||||
|
||||
def init(path):
|
||||
global _file
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
_file = open(path, "a", encoding="utf-8")
|
||||
audit("system", "startup", "", "")
|
||||
|
||||
|
||||
def audit(tool, serial, args_summary, result):
|
||||
global _file
|
||||
if _file is None:
|
||||
return
|
||||
rec = {"ts": datetime.now().isoformat(timespec="seconds"),
|
||||
"tool": tool, "serial": serial,
|
||||
"args": str(args_summary)[:200], "result": str(result)[:200]}
|
||||
try:
|
||||
_file.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
||||
_file.flush()
|
||||
except Exception:
|
||||
pass
|
||||
@@ -0,0 +1,32 @@
|
||||
"""MCP Server 配置（环境变量，生产用 .env 注入）。"""
|
||||
import os
|
||||
|
||||
|
||||
def _env(key, default):
|
||||
return os.environ.get(key, default)
|
||||
|
||||
|
||||
class Settings:
|
||||
# 平台（auto_control）地址与账号
|
||||
platform_url = _env("MCP_PLATFORM_URL", "http://127.0.0.1:18050")
|
||||
platform_user = _env("MCP_PLATFORM_USER", "admin")
|
||||
platform_pass = _env("MCP_PLATFORM_PASS", "")
|
||||
|
||||
# 安全
|
||||
allow_write = _env("MCP_ALLOW_WRITE", "0") == "1" # 写操作门控（默认只读）
|
||||
allowed_serials = [s.strip() for s in
|
||||
_env("MCP_ALLOWED_SERIALS", "").split(",") if s.strip()] # 空=不限
|
||||
|
||||
# 传输
|
||||
http_host = _env("MCP_HTTP_HOST", "0.0.0.0")
|
||||
http_port = int(_env("MCP_HTTP_PORT", "8033"))
|
||||
|
||||
# 截图
|
||||
screenshot_width = int(_env("MCP_SCREENSHOT_WIDTH", "540"))
|
||||
jpeg_quality = int(_env("MCP_JPEG_QUALITY", "70"))
|
||||
|
||||
# 审计
|
||||
audit_file = _env("MCP_AUDIT_FILE", "/var/log/mcp/audit.log")
|
||||
|
||||
# 平台请求超时（秒）
|
||||
platform_timeout = float(_env("MCP_PLATFORM_TIMEOUT", "30"))
|
||||
@@ -0,0 +1,266 @@
|
||||
"""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 开启）")
|
||||
|
||||
|
||||
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:
|
||||
"""列出可控制设备：serial/在线状态/型号/任务状态/前台 App。
|
||||
|
||||
返回 [{serial, model, online, task_job, worker_status, foreground_app}]。
|
||||
"""
|
||||
try:
|
||||
devs = platform().list_devices()
|
||||
except PlatformError as e:
|
||||
return _err(e)
|
||||
out = []
|
||||
for d in devs:
|
||||
out.append({
|
||||
"serial": d.get("serial"),
|
||||
"model": d.get("model") or d.get("device_name") 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 返回的图像坐标（display 空间）——先截图拿到
|
||||
native_size 后再点击，server 自动换算为设备原生坐标。
|
||||
"""
|
||||
try:
|
||||
_check_write()
|
||||
serial = _check_serial(serial)
|
||||
if x < 0 or y < 0:
|
||||
raise PlatformError("invalid_param", "坐标不能为负")
|
||||
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})->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)
|
||||
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")
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def de_ui_tree(serial: str) -> dict:
|
||||
"""获取当前界面元素树（文本 JSON）：每元素含 text/resource_id/description/class/bounds。
|
||||
|
||||
优先用它定位元素（元素驱动操作），比纯坐标点击更可靠。
|
||||
"""
|
||||
try:
|
||||
serial = _check_serial(serial)
|
||||
els = platform().ui_elements(serial)
|
||||
except PlatformError as e:
|
||||
return _err(e)
|
||||
# 精简输出：去掉 suggested/深度噪音，保留可定位属性
|
||||
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],
|
||||
"bounds": e.get("bounds", ""),
|
||||
})
|
||||
audit.audit("de_ui_tree", serial, "", f"{len(slim)} 元素")
|
||||
return _ok({"count": len(slim), "elements": slim[:300]})
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def de_tap_element(serial: str, by: str, value: str, index: int = 1) -> dict:
|
||||
"""按元素点击（不需要坐标）：by=text|id|desc，value 为匹配文本/资源 id/描述。
|
||||
|
||||
元素驱动操作比坐标可靠（界面变化自适应）；元素不存在时返回错误，
|
||||
可改用 de_ui_tree 查元素或 de_tap 坐标兜底。index 用于多命中取第几个（默认 1）。
|
||||
"""
|
||||
try:
|
||||
_check_write()
|
||||
serial = _check_serial(serial)
|
||||
if by not in ("text", "id", "desc"):
|
||||
raise PlatformError("invalid_param", "by 可选 text/id/desc")
|
||||
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 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_wake(serial: str) -> dict:
|
||||
"""点亮设备屏幕并解锁（熄屏时先调用它再截图）。"""
|
||||
try:
|
||||
_check_write()
|
||||
serial = _check_serial(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)
|
||||
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})
|
||||
|
||||
|
||||
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()
|
||||
@@ -0,0 +1,145 @@
|
||||
"""平台（auto_control）HTTP 客户端：登录会话 + CSRF + API 封装。
|
||||
|
||||
会话失效自动重登；CSRF token 每次登录后获取，POST 必带。
|
||||
只读接口（GET）与写接口（POST）分离封装，写操作由 MCP 门控层决定是否调用。
|
||||
"""
|
||||
import httpx
|
||||
import logging
|
||||
|
||||
_log = logging.getLogger("mcp.platform")
|
||||
|
||||
_CSRF_HEADER = "X-CSRF-Token"
|
||||
|
||||
|
||||
class PlatformError(Exception):
|
||||
def __init__(self, code, message):
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
self.message = message
|
||||
|
||||
|
||||
class PlatformClient:
|
||||
def __init__(self, url, user, password, timeout=30.0):
|
||||
self.url = url.rstrip("/")
|
||||
self.user = user
|
||||
self.password = password
|
||||
self._client = httpx.Client(base_url=self.url, timeout=timeout,
|
||||
follow_redirects=True)
|
||||
self._csrf = ""
|
||||
self._login()
|
||||
|
||||
# ---------- 会话 ----------
|
||||
def _login(self):
|
||||
"""登录平台，拿会话 cookie + CSRF token。"""
|
||||
r = self._client.post("/login", data={
|
||||
"username": self.user, "password": self.password})
|
||||
if r.status_code != 302 and r.status_code != 200:
|
||||
raise PlatformError("platform_unavailable",
|
||||
f"平台登录失败 HTTP {r.status_code}: {r.text[:120]}")
|
||||
csrf = self._client.get("/api/csrf")
|
||||
if csrf.status_code == 200:
|
||||
self._csrf = (csrf.json() or {}).get("token", "")
|
||||
_log.info("平台登录成功: %s", self.user)
|
||||
|
||||
def _ensure_session(self):
|
||||
"""会话过期（401/403/302 到登录页）时重登。"""
|
||||
r = self._client.get("/api/status")
|
||||
if r.status_code in (401, 403) or r.url.path.endswith("/login"):
|
||||
_log.warning("会话失效，重新登录")
|
||||
self._login()
|
||||
return True
|
||||
return False
|
||||
|
||||
# ---------- 基础请求 ----------
|
||||
def _get(self, path, **params):
|
||||
self._ensure_session()
|
||||
return self._client.get(path, params=params)
|
||||
|
||||
def _post(self, path, json=None):
|
||||
self._ensure_session()
|
||||
headers = {_CSRF_HEADER: self._csrf} if self._csrf else {}
|
||||
return self._client.post(path, json=json or {}, headers=headers)
|
||||
|
||||
# ---------- 平台 API 封装（M0） ----------
|
||||
def list_devices(self):
|
||||
"""设备列表与状态（GET /api/status）。"""
|
||||
r = self._get("/api/status")
|
||||
if r.status_code != 200:
|
||||
raise PlatformError("platform_unavailable",
|
||||
f"/api/status HTTP {r.status_code}")
|
||||
j = r.json() or {}
|
||||
return j.get("devices", [])
|
||||
|
||||
def screenshot(self, serial):
|
||||
"""截图（GET /api/screen/thumb），返回 (jpeg_bytes, screen_state)。"""
|
||||
r = self._get("/api/screen/thumb", 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}")
|
||||
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,
|
||||
"x": int(x), "y": int(y)})
|
||||
return self._check_op(r, "tap")
|
||||
|
||||
def swipe(self, serial, x1, y1, x2, y2, duration=0.2):
|
||||
"""滑动（POST /api/screen/swipe）。"""
|
||||
r = self._post("/api/screen/swipe", json={
|
||||
"serial": serial, "x1": int(x1), "y1": int(y1),
|
||||
"x2": int(x2), "y2": int(y2),
|
||||
"duration": float(duration)})
|
||||
return self._check_op(r, "swipe")
|
||||
|
||||
def ui_elements(self, serial):
|
||||
"""UI 元素树（GET /api/uiauto/elements，uiautodev 服务）。"""
|
||||
r = self._get("/api/uiauto/elements", 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 j.get("elements", [])
|
||||
|
||||
def wake(self, serial):
|
||||
"""亮屏并解锁（POST /api/device/screen_all mode=on）。"""
|
||||
r = self._post("/api/device/screen_all",
|
||||
json={"mode": "on", "serials": [serial]})
|
||||
return self._check_op(r, "wake")
|
||||
|
||||
def press_key(self, serial, key):
|
||||
"""按键（POST /api/screen/key）。"""
|
||||
r = self._post("/api/screen/key",
|
||||
json={"serial": serial, "key": key})
|
||||
return self._check_op(r, "key")
|
||||
|
||||
@staticmethod
|
||||
def _check_op(r, name):
|
||||
if r.status_code == 503:
|
||||
raise PlatformError("device_offline", r.text[:120])
|
||||
if r.status_code != 200:
|
||||
raise PlatformError("platform_unavailable",
|
||||
f"{name} HTTP {r.status_code}: {r.text[:120]}")
|
||||
j = r.json() or {}
|
||||
if not j.get("ok"):
|
||||
raise PlatformError("device_offline", str(j.get("error", "操作失败"))[:120])
|
||||
return j
|
||||
@@ -0,0 +1,3 @@
|
||||
fastmcp>=2.0
|
||||
httpx>=0.27
|
||||
Pillow>=10.0
|
||||
@@ -542,6 +542,25 @@ def api_screen_text():
|
||||
return jsonify({"ok": False, "error": f"输入失败: {e}"}), 503
|
||||
|
||||
|
||||
@bp.route("/api/screen/size")
|
||||
@perm_required(PERM_DEVICES)
|
||||
def api_screen_size():
|
||||
"""屏幕原生分辨率（只读，供 MCP 坐标换算：截图是缩放图，操作需原生坐标）。
|
||||
|
||||
返回 {ok, width, height}（如 1080x2400）；离线/不可达 → 503。
|
||||
"""
|
||||
serial = request.args.get("serial", "")
|
||||
if not serial:
|
||||
return jsonify({"ok": False, "error": "缺少 serial"}), 400
|
||||
try:
|
||||
d = _screen_get_device(serial)
|
||||
w, h = d.window_size()
|
||||
return jsonify({"ok": True, "width": w, "height": h})
|
||||
except Exception as e:
|
||||
_screen_invalidate(serial)
|
||||
return jsonify({"ok": False, "error": f"获取分辨率失败: {e}"}), 503
|
||||
|
||||
|
||||
# ================== API：应用管理（APK 上传/安装）==================
|
||||
|
||||
def _current_focus_pkg(serial):
|
||||
|
||||
Reference in New Issue
Block a user