147 lines
4.7 KiB
Python
147 lines
4.7 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()
|
||
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 _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()
|
||
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")
|
||
return _ok({"image": {"type": "image", "data": data,
|
||
"mimeType": "image/jpeg"},
|
||
"width": w, "height": h,
|
||
"screen_state": screen_state})
|
||
|
||
|
||
@mcp.tool()
|
||
def de_tap(serial: str, x: int, y: int) -> dict:
|
||
"""点击设备屏幕指定坐标（设备原生分辨率像素，与截图 1:1 换算）。"""
|
||
try:
|
||
_check_write()
|
||
serial = _check_serial(serial)
|
||
if x < 0 or y < 0:
|
||
raise PlatformError("invalid_param", "坐标不能为负")
|
||
platform().tap(serial, x, y)
|
||
except PlatformError as e:
|
||
return _err(e)
|
||
audit.audit("de_tap", serial, f"({x},{y})", "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:
|
||
"""在设备屏幕上滑动（起终点为原生分辨率像素）。"""
|
||
try:
|
||
_check_write()
|
||
serial = _check_serial(serial)
|
||
platform().swipe(serial, x1, y1, x2, y2, 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})
|
||
|
||
|
||
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()
|