feat(M1): Agent 编排层（OpenAI 兼容）——DeepSeek 等第三方模型经 MCP 工具控制手机：工具桥转 function schema、截图图像转 image_url 多模态流、工具循环、CLI 入口（mock 验证通过）

mcp_agent/__init__.py

@@ -0,0 +1 @@
+"""Agent 编排层：第三方 LLM（OpenAI 兼容）经 MCP 工具控制手机。"""

mcp_agent/agent.py

@@ -0,0 +1,148 @@
+"""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:
+            schema = t.inputSchema if hasattr(t, "inputSchema") else {}
+            # 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 "（达到最大步骤数未完成，请检查操作是否卡在循环）"

mcp_agent/cli.py

@@ -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()

mcp_agent/config.py

@@ -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")

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")

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,

web/monitor.py

@@ -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):