feat(M1): 补 de_wake/de_press_key 工具(Agent 熄屏可自唤醒);修复 MCP SDK v2 input_schema 字段警告

This commit is contained in:
2026-09-04 12:50:55 +08:00
parent 937a87b36c
commit 7017ff3315
3 changed files with 46 additions and 1 deletions
+2 -1
View File
@@ -48,7 +48,8 @@ class Agent:
tools = await self._mcp.list_tools()
self.tools_schema = []
for t in tools:
schema = t.inputSchema if hasattr(t, "inputSchema") else {}
# 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 ""
+32
View File
@@ -163,6 +163,38 @@ def de_swipe(serial: str, x1: int, y1: int, x2: int, y2: int,
return _ok({"action": "swipe", "serial": serial})
_KEYS = ("back", "home", "recent", "menu", "power", "volume_up",
"volume_down", "enter", "delete", "search", "camera")
@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)
+12
View File
@@ -107,6 +107,18 @@ class PlatformClient:
"duration": float(duration)})
return self._check_op(r, "swipe")
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: