From 6b8436ebd86f74bde368b703e9a7969857839d4e Mon Sep 17 00:00:00 2001 From: chenwei Date: Thu, 30 Jul 2026 15:48:26 +0000 Subject: [PATCH] =?UTF-8?q?docs:=20ppclock-mcp=20=E5=AE=9E=E6=96=BD?= =?UTF-8?q?=E8=AE=A1=E5=88=92=EF=BC=885=20=E4=BB=BB=E5=8A=A1=EF=BC=8CTDD?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude --- .../plans/2026-07-30-ppclock-mcp.md | 1086 +++++++++++++++++ 1 file changed, 1086 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-30-ppclock-mcp.md diff --git a/docs/superpowers/plans/2026-07-30-ppclock-mcp.md b/docs/superpowers/plans/2026-07-30-ppclock-mcp.md new file mode 100644 index 0000000..01bc391 --- /dev/null +++ b/docs/superpowers/plans/2026-07-30-ppclock-mcp.md @@ -0,0 +1,1086 @@ +# ppclock-mcp Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 把 ppclock SDK 包装成单层部署的 MCP 服务器（stdio + streamable HTTP 双模），供各 agent 平台调用，Windows 蓝牙服务器先行实测。 + +**Architecture:** 新增三个模块（`device_manager.py` 连接管理 / `mcp_tools.py` 工具层 / `mcp_server.py` 装配入口），作为 SDK 的纯消费者；连接管理用方案 C（按需连接 + 空闲超时保持 + 全局串行锁 + 掉线重连重试一次）。 + +**Tech Stack:** Python ≥3.10、官方 `mcp` 包（FastMCP）、bleak（Windows=WinRT / Linux=BlueZ）、pytest + pytest-asyncio。 + +**Spec:** `docs/superpowers/specs/2026-07-30-ppclock-mcp-design.md` + +## Global Constraints + +- **SDK 零改动**：不改 `client.py`/`commands.py`/`protocol.py`/`image_pipeline.py`/`transports/*` 任何一行。 +- `mcp>=1.10` 只能进 optional extra，基础 SDK 依赖（bleak、pillow）不变。 +- 输出契约：成功 `{"ok": true, "tool": , "data": {...}}`；失败 `{"ok": false, "tool": , "error": {"code","message","hint"?}}`。 +- 错误码仅：`INVALID_PARAM` / `DEVICE_NOT_FOUND` / `CONNECT_TIMEOUT` / `BLE_ERROR` / `DEVICE_DISCONNECTED` / `INTERNAL`。 +- HTTP 默认 `--host 127.0.0.1 --port 8972`；绑非回环地址必须设环境变量 `PPCLOCK_MCP_TOKEN`。 +- 代码注释/docstring 用中文，风格对齐现有源码；commit 前缀用 `feat:`/`test:`/`docs:`。 +- 测试用 `@pytest.mark.asyncio`（对齐 tests/ 现有模式），FakeTransport 模式复用 tests/test_client.py。 +- 全部测试命令用 `.venv/bin/python -m pytest tests/ -q`（仓库根目录）。 + +--- + +### Task 1: DeviceManager（连接管理器） + +**Files:** +- Create: `src/ppclock/device_manager.py` +- Test: `tests/test_device_manager.py` + +**Interfaces:** +- Consumes: `ppclock.client.PPClient`、`ppclock.transports.base.TransportError` +- Produces: + - `DeviceManager(mac: str|None=None, connect_timeout: float=90.0, idle_timeout: float=300.0, client_factory: Callable[[str|None, float], PPClient]|None=None)` + - `async manager.run(op) -> Any` — `op` 是 `async (PPClient) -> Any`；串行执行，掉线自动重连重试一次 + - `async manager.status() -> dict` — `{connected, mac, idle_seconds, connect_count, connect_timeout, idle_timeout, device_id?}` + - `async manager.close() -> None` + - `manager.connected: bool`（property） +- 下游（Task 2/3）只依赖以上名字与签名。 + +- [ ] **Step 1: 写失败测试** + +```python +"""DeviceManager 测试：按需连接/串行/空闲超时/掉线重连。""" +import asyncio + +import pytest + +from ppclock.client import PPClient +from ppclock.device_manager import DeviceManager +from ppclock.transports.base import TransportError + + +class FakeTransport: + def __init__(self): + self.address = "AA:BB:CC:DD:EE:FF" + self.enters = 0 + self.exits = 0 + self.rxtx_writes = [] + self.fail_next_write = False + + async def __aenter__(self): + self.enters += 1 + return self + + async def __aexit__(self, *exc): + self.exits += 1 + return False + + async def write_epd(self, data, response=True): + pass + + async def write_rxtx(self, data, response=True): + if self.fail_next_write: + self.fail_next_write = False + raise TransportError("connection lost") + self.rxtx_writes.append(data) + + async def read_rxtx(self): + return b"\x00" + + async def request_device_id(self, timeout=8.0): + return "81233F3C267112" + + async def run_ota(self, image, on_progress=None): + return {} + + async def delay(self, seconds): + pass + + +def make_manager(transports, **kw): + """client_factory 依次弹出预置 transport 包成 PPClient。""" + def factory(mac, timeout): + return PPClient(transports.pop(0)) + kw.setdefault("client_factory", factory) + return DeviceManager(**kw) + + +class TestConnect: + @pytest.mark.asyncio + async def test_connect_on_demand_and_reuse(self): + t = FakeTransport() + m = make_manager([t]) + assert not m.connected + await m.run(lambda dev: dev.set_mode("clock1")) + await m.run(lambda dev: dev.set_mode("clock2")) + assert t.enters == 1 # 第二次复用连接 + assert m.connected + await m.close() + + @pytest.mark.asyncio + async def test_remembers_address(self): + t = FakeTransport() + m = make_manager([t], mac=None) + await m.run(lambda dev: dev.set_mode("clock1")) + assert (await m.status())["mac"] == "AA:BB:CC:DD:EE:FF" + await m.close() + + +class TestSerialization: + @pytest.mark.asyncio + async def test_ops_are_serialized(self): + t = FakeTransport() + m = make_manager([t]) + concurrent = 0 + peak = 0 + + async def op(dev): + nonlocal concurrent, peak + concurrent += 1 + peak = max(peak, concurrent) + await asyncio.sleep(0.02) + concurrent -= 1 + return 1 + + await asyncio.gather(*(m.run(op) for _ in range(5))) + assert peak == 1 + await m.close() + + +class TestIdleTimeout: + @pytest.mark.asyncio + async def test_idle_disconnect(self): + t = FakeTransport() + m = make_manager([t], idle_timeout=0.05) + await m.run(lambda dev: dev.set_mode("clock1")) + assert m.connected + await asyncio.sleep(0.15) + assert not m.connected + assert t.exits == 1 + + @pytest.mark.asyncio + async def test_activity_resets_idle(self): + t = FakeTransport() + m = make_manager([t], idle_timeout=0.1) + await m.run(lambda dev: dev.set_mode("clock1")) + await asyncio.sleep(0.06) + await m.run(lambda dev: dev.set_mode("clock2")) + await asyncio.sleep(0.06) + assert m.connected # 第二次活动续期 + await m.close() + + +class TestReconnect: + @pytest.mark.asyncio + async def test_reconnect_once_on_transport_error(self): + t1, t2 = FakeTransport(), FakeTransport() + t1.fail_next_write = True + m = make_manager([t1, t2]) + await m.run(lambda dev: dev.set_mode("clock1")) + assert t1.enters == 1 and t2.enters == 1 # 掉线重连一次并成功 + await m.close() + + @pytest.mark.asyncio + async def test_retry_exhaustion_raises(self): + t1, t2 = FakeTransport(), FakeTransport() + t1.fail_next_write = True + t2.fail_next_write = True + m = make_manager([t1, t2]) + with pytest.raises(TransportError): + await m.run(lambda dev: dev.set_mode("clock1")) + await m.close() + + +class TestStatus: + @pytest.mark.asyncio + async def test_status_fields(self): + t = FakeTransport() + m = make_manager([t], connect_timeout=90.0, idle_timeout=300.0) + s = await m.status() + assert s["connected"] is True # status 自身触发连接 + assert s["device_id"] == "81233F3C267112" + assert s["connect_timeout"] == 90.0 + assert s["idle_timeout"] == 300.0 + assert isinstance(s["idle_seconds"], float) + await m.close() +``` + +注：`status()` 在设备 ID 读取上复用已建立连接；实现上 status 在连接时顺手 `get_device_id()`。`test_status_fields` 假定 status 会确保连接（见实现）。 + +- [ ] **Step 2: 跑测试确认失败** + +Run: `.venv/bin/python -m pytest tests/test_device_manager.py -q` +Expected: FAIL（`ModuleNotFoundError: No module named 'ppclock.device_manager'`） + +- [ ] **Step 3: 实现 device_manager.py** + +```python +"""设备连接管理器（方案 C：按需连接 + 空闲保持，全操作串行）。 + +- 一把 asyncio.Lock 串行所有设备操作（BLE 适配器独占） +- 首次操作守候连接（connect_timeout 应覆盖设备 30-60s 广播窗口） +- 操作后保持连接；空闲 idle_timeout 秒自动断开让设备睡眠 +- 操作中 TransportError（连接丢失）→ 重连并重试一次，再失败向上抛 +""" +from __future__ import annotations + +import asyncio +import time + +from .client import PPClient +from .transports.base import TransportError + + +class DeviceManager: + def __init__(self, mac: str | None = None, connect_timeout: float = 90.0, + idle_timeout: float = 300.0, client_factory=None): + self._mac = mac + self._connect_timeout = connect_timeout + self._idle_timeout = idle_timeout + self._factory = client_factory or ( + lambda mac, timeout: PPClient.via_local(mac=mac, timeout=timeout)) + self._client = None + self._lock = asyncio.Lock() + self._last_activity: float | None = None + self._idle_task: asyncio.Task | None = None + self._connect_count = 0 + + @property + def connected(self) -> bool: + return self._client is not None + + async def run(self, op): + """串行执行 op(client)；掉线重连重试一次。""" + async with self._lock: + try: + return await self._run_with_retry(op) + finally: + self._touch() + + async def status(self) -> dict: + async with self._lock: + await self._ensure_connected() + self._touch() + info = { + "connected": self.connected, + "mac": self._mac, + "idle_seconds": None if self._last_activity is None + else round(time.monotonic() - self._last_activity, 1), + "connect_count": self._connect_count, + "connect_timeout": self._connect_timeout, + "idle_timeout": self._idle_timeout, + } + try: + info["device_id"] = await self._client.get_device_id() + except Exception: # noqa: BLE001 - 状态查询尽力而为 + pass + return info + + async def close(self) -> None: + if self._idle_task is not None: + self._idle_task.cancel() + async with self._lock: + await self._drop() + + # ---------- 内部 ---------- + + async def _run_with_retry(self, op): + await self._ensure_connected() + try: + return await op(self._client) + except TransportError: + await self._drop() + await self._ensure_connected() + return await op(self._client) + + async def _ensure_connected(self): + if self._client is not None: + return + client = self._factory(self._mac, self._connect_timeout) + await client.__aenter__() + self._client = client + self._connect_count += 1 + if self._mac is None: # 自动扫描后记住实际地址 + self._mac = getattr(client._t, "address", None) + + async def _drop(self): + client, self._client = self._client, None + if client is not None: + try: + await client.__aexit__(None, None, None) + except Exception: # noqa: BLE001 - 断开失败不影响状态 + pass + + def _touch(self): + self._last_activity = time.monotonic() + if self._idle_task is not None: + self._idle_task.cancel() + self._idle_task = None + if self.connected: + self._idle_task = asyncio.create_task(self._idle_watch()) + + async def _idle_watch(self): + try: + await asyncio.sleep(self._idle_timeout) + async with self._lock: + if (self._last_activity is not None + and time.monotonic() - self._last_activity >= self._idle_timeout): + await self._drop() + except asyncio.CancelledError: + pass +``` + +注：`client._t` 是同包内访问 PPClient 持有的传输层（Transport 协议公开 `address` 属性），SDK 无公开访问器，此处为有意为之的包内约定。 + +- [ ] **Step 4: 跑测试确认通过** + +Run: `.venv/bin/python -m pytest tests/test_device_manager.py -q` +Expected: 9 passed + +- [ ] **Step 5: 回归 + 提交** + +Run: `.venv/bin/python -m pytest tests/ -q` +Expected: 165 + 9 全绿 + +```bash +git add src/ppclock/device_manager.py tests/test_device_manager.py +git commit -m "feat(mcp): DeviceManager——按需连接+空闲保持+串行锁+掉线重连（方案 C）" +``` + +--- + +### Task 2: mcp_tools（工具层：图源解码 + 错误契约 + 13 工具） + +**Files:** +- Create: `src/ppclock/mcp_tools.py` +- Test: `tests/test_mcp_tools.py` + +**Interfaces:** +- Consumes: Task 1 的 `DeviceManager.run(op)` / `DeviceManager.status()`；SDK `PPClient` 全部高层方法 +- Produces: + - `decode_image_source(source: str) -> PIL.Image.Image` — 路径或 base64（可带 `data:*;base64,` 前缀）；非法输入抛 `ValueError` + - `register_tools(mcp, manager) -> None` — 把 13 个工具注册到 FastMCP 实例 + - 工具名（Task 3 与文档依赖，逐字）：`scan_devices / device_status / set_time / set_mode / toggle / upload_image / render_template / countdown / countdown_off / calendar_text / set_sleep / clear_screen / raw_send` + +- [ ] **Step 1: 写失败测试** + +```python +"""mcp_tools 测试：图源解码 + 工具契约（内存 MCP 会话端到端）。""" +import base64 +import io +import json + +import pytest +from mcp.server.fastmcp import FastMCP +from mcp.shared.memory import create_connected_server_and_client_session +from PIL import Image + +from ppclock.client import PPClient +from ppclock.mcp_tools import decode_image_source, register_tools +from ppclock.transports.base import TransportError + + +class FakeTransport: + def __init__(self): + self.address = "AA:BB:CC:DD:EE:FF" + self.epd_writes = [] + self.rxtx_writes = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def write_epd(self, data, response=True): + self.epd_writes.append(data) + + async def write_rxtx(self, data, response=True): + self.rxtx_writes.append(data) + + async def read_rxtx(self): + return b"\x00" + + async def request_device_id(self, timeout=8.0): + return "81233F3C267112" + + async def run_ota(self, image, on_progress=None): + return {} + + async def delay(self, seconds): + pass + + +class FakeManager: + """模拟 DeviceManager：直接对 fake client 执行 op，或抛预置错误。""" + + def __init__(self, error=None): + self.client = PPClient(FakeTransport()) + self.error = error + self.ops = 0 + + async def run(self, op): + self.ops += 1 + if self.error is not None: + raise self.error + return await op(self.client) + + async def status(self): + return {"connected": True, "mac": "AA:BB:CC:DD:EE:FF", + "idle_seconds": 0.0, "connect_count": 1, + "connect_timeout": 90.0, "idle_timeout": 300.0, + "device_id": "81233F3C267112"} + + +def make_session_coro(manager): + mcp = FastMCP("ppclock-test") + register_tools(mcp, manager) + return mcp + + +async def call(mcp, tool, args=None): + async with create_connected_server_and_client_session(mcp._mcp_server) as s: + result = await s.call_tool(tool, args or {}) + assert not result.isError, result.content + return json.loads(result.content[0].text) + + +def png_b64() -> str: + buf = io.BytesIO() + Image.new("RGB", (400, 300), "white").save(buf, "PNG") + return base64.b64encode(buf.getvalue()).decode() + + +class TestDecodeImageSource: + def test_path(self, tmp_path): + p = tmp_path / "a.png" + Image.new("RGB", (10, 10)).save(p) + assert decode_image_source(str(p)).size == (10, 10) + + def test_base64(self): + assert decode_image_source(png_b64()).size == (400, 300) + + def test_data_uri(self): + src = "data:image/png;base64," + png_b64() + assert decode_image_source(src).size == (400, 300) + + def test_garbage(self): + with pytest.raises(ValueError): + decode_image_source("!!!not-base64!!!") + + +class TestTools: + @pytest.mark.asyncio + async def test_set_time_ok(self): + r = await call(make_session_coro(FakeManager()), "set_time", {"tz": 8.0}) + assert r["ok"] is True and r["tool"] == "set_time" + + @pytest.mark.asyncio + async def test_upload_image_base64(self): + r = await call(make_session_coro(FakeManager()), "upload_image", + {"source": png_b64(), "slot": 0}) + assert r["ok"] is True + assert r["data"]["bytes_bw"] == 15000 + assert r["data"]["slot"] == 0 + + @pytest.mark.asyncio + async def test_set_mode_invalid_param(self): + r = await call(make_session_coro(FakeManager()), "set_mode", + {"mode": "nonsense"}) + assert r["ok"] is False + assert r["error"]["code"] == "INVALID_PARAM" + + @pytest.mark.asyncio + async def test_transport_error_mapped(self): + m = FakeManager(error=TransportError("Connection timed out")) + r = await call(make_session_coro(m), "set_time", {}) + assert r["ok"] is False + assert r["error"]["code"] == "CONNECT_TIMEOUT" + assert "hint" in r["error"] + + @pytest.mark.asyncio + async def test_device_status(self): + r = await call(make_session_coro(FakeManager()), "device_status", {}) + assert r["ok"] is True + assert r["data"]["device_id"] == "81233F3C267112" + + @pytest.mark.asyncio + async def test_raw_send(self): + r = await call(make_session_coro(FakeManager()), "raw_send", + {"data_hex": "e201", "channel": "alt"}) + assert r["ok"] is True + + @pytest.mark.asyncio + async def test_raw_send_bad_hex(self): + r = await call(make_session_coro(FakeManager()), "raw_send", + {"data_hex": "zz", "channel": "alt"}) + assert r["ok"] is False + assert r["error"]["code"] == "INVALID_PARAM" + + @pytest.mark.asyncio + async def test_render_template(self): + r = await call(make_session_coro(FakeManager()), "render_template", + {"name": "custom", "payload": {"text": "你好"}, "slot": 0}) + assert r["ok"] is True + assert r["data"]["template"] == "custom" + + @pytest.mark.asyncio + async def test_countdown(self): + r = await call(make_session_coro(FakeManager()), "countdown", + {"date": "2026-12-31", "mode": "clock", "prefix": "目标"}) + assert r["ok"] is True + + @pytest.mark.asyncio + async def test_scan_devices(self, monkeypatch): + async def fake_scan(uri="local", timeout=5.0): + return [{"name": "NRF-5DBF28", "address": "18:BC:5A:5D:BF:28", + "rssi": -60}] + monkeypatch.setattr(PPClient, "scan_devices", staticmethod(fake_scan)) + r = await call(make_session_coro(FakeManager()), "scan_devices", {}) + assert r["ok"] is True + assert r["data"]["devices"][0]["address"] == "18:BC:5A:5D:BF:28" +``` + +- [ ] **Step 2: 装依赖 + 跑测试确认失败** + +Run: `.venv/bin/pip install "mcp>=1.10" -q && .venv/bin/python -m pytest tests/test_mcp_tools.py -q` +Expected: FAIL（`ModuleNotFoundError: No module named 'ppclock.mcp_tools'`） + +- [ ] **Step 3: 实现 mcp_tools.py** + +```python +"""MCP 工具层：13 个高层工具，统一输出契约（与 CLI --json 同构）。 + +成功 {"ok": true, "tool": name, "data": {...}} +失败 {"ok": false, "tool": name, "error": {"code", "message", "hint"?}} +""" +from __future__ import annotations + +import base64 +import binascii +import datetime +import io +import os + +from PIL import Image + +from .client import PPClient +from .transports.base import TransportError + +_TIMEOUT_HINT = "设备长睡眠、广播窗口 30-60s；调大 --connect-timeout 或稍后重试" +_NOTFOUND_HINT = "未发现 NRF- 前缀设备；确认设备在位且未处于深睡" + + +def decode_image_source(source: str) -> Image.Image: + """source = 服务器本地路径，或 base64（可带 data URI 前缀）。""" + if os.path.exists(source): + return Image.open(source) + data = source + if data.startswith("data:"): + if "," not in data: + raise ValueError("data URI 缺少 ',' 分隔") + data = data.split(",", 1)[1] + try: + raw = base64.b64decode(data, validate=True) + except (binascii.Error, ValueError): + raise ValueError("source 既不是存在的文件路径，也不是合法 base64") from None + return Image.open(io.BytesIO(raw)) + + +def _err(tool: str, code: str, message: str, hint: str | None = None) -> dict: + error = {"code": code, "message": message} + if hint: + error["hint"] = hint + return {"ok": False, "tool": tool, "error": error} + + +def _map_transport_error(tool: str, exc: TransportError) -> dict: + msg = str(exc) + low = msg.lower() + if "timeout" in low or "timed out" in low: + return _err(tool, "CONNECT_TIMEOUT", msg, _TIMEOUT_HINT) + if "not found" in low or "no device" in low: + return _err(tool, "DEVICE_NOT_FOUND", msg, _NOTFOUND_HINT) + return _err(tool, "BLE_ERROR", msg) + + +def register_tools(mcp, manager) -> None: + """把 13 个工具注册到 FastMCP 实例；manager 为 DeviceManager（测试可注入 fake）。""" + + async def _call(tool: str, op) -> dict: + try: + data = await manager.run(op) + return {"ok": True, "tool": tool, "data": data if data is not None else {}} + except TransportError as exc: + return _map_transport_error(tool, exc) + except ValueError as exc: + return _err(tool, "INVALID_PARAM", str(exc)) + except Exception as exc: # noqa: BLE001 - 工具层兜底，错误须回 agent 而非炸会话 + return _err(tool, "INTERNAL", f"{type(exc).__name__}: {exc}") + + @mcp.tool() + async def scan_devices(timeout: float = 5.0) -> dict: + """扫描附近 NRF- 前缀墨水屏设备，返回 [{name,address,rssi}]。""" + try: + devices = await PPClient.scan_devices("local", timeout=timeout) + return {"ok": True, "tool": "scan_devices", "data": {"devices": devices}} + except TransportError as exc: + return _map_transport_error("scan_devices", exc) + except Exception as exc: # noqa: BLE001 + return _err("scan_devices", "INTERNAL", f"{type(exc).__name__}: {exc}") + + @mcp.tool() + async def device_status() -> dict: + """查询连接状态/绑定 MAC/设备 ID/空闲秒数。""" + try: + return {"ok": True, "tool": "device_status", "data": await manager.status()} + except Exception as exc: # noqa: BLE001 + return _err("device_status", "INTERNAL", f"{type(exc).__name__}: {exc}") + + @mcp.tool() + async def set_time(tz: float = 8.0) -> dict: + """对时；tz 为时区偏移（默认 8=北京时间）。""" + return await _call("set_time", lambda dev: dev.set_time(tz=tz)) + + @mcp.tool() + async def set_mode(mode: str) -> dict: + """切换显示模式：clock1-3/calendar1-3/image0-3/tricolor/mono。""" + return await _call("set_mode", lambda dev: dev.set_mode(mode)) + + @mcp.tool() + async def toggle(name: str) -> dict: + """单字节切换：invert/font/rotate180/hour_format/clock_color。""" + return await _call("toggle", lambda dev: dev.toggle(name)) + + @mcp.tool() + async def upload_image(source: str, slot: int = 0, algo: str = "atkinson", + mono: bool = False, + threshold: float | None = None, + diffusion: float | None = None, + brightness: float | None = None, + contrast: float | None = None, + saturation: float | None = None, + rotate: float | None = None) -> dict: + """传图到指定槽位。source=服务器路径或 base64；algo∈ + none/floydsteinberg/atkinson/bayer/stucki/jarvis。""" + adjust = {k: v for k, v in { + "threshold": threshold, "diffusion": diffusion, + "brightness": brightness, "contrast": contrast, + "saturation": saturation, "rotate": rotate}.items() if v is not None} + img = decode_image_source(source) + return await _call("upload_image", lambda dev: dev.upload_image( + img, slot=slot, algo=algo, mono=mono, **adjust)) + + @mcp.tool() + async def render_template(name: str, payload: dict | None = None, + slot: int = 0, algo: str = "atkinson") -> dict: + """本地渲染模板并上传：schedule/businesscard/memo/course/qrcode/custom。""" + return await _call("render_template", lambda dev: dev.render_template( + name, payload, slot=slot, algo=algo)) + + @mcp.tool() + async def countdown(date: str, mode: str = "clock", + prefix: str | None = None) -> dict: + """倒计时；date 为 ISO YYYY-MM-DD；mode=clock/calendar。""" + d = datetime.date.fromisoformat(date) + return await _call("countdown", lambda dev: dev.countdown( + d, mode=mode, prefix=prefix)) + + @mcp.tool() + async def countdown_off() -> dict: + """关闭倒计时。""" + return await _call("countdown_off", lambda dev: dev.countdown_off()) + + @mcp.tool() + async def calendar_text(text: str) -> dict: + """设置日历中文文字并切到日历模式一。""" + return await _call("calendar_text", lambda dev: dev.calendar_text(text)) + + @mcp.tool() + async def set_sleep(on: bool, start_h: int, end_h: int) -> dict: + """设置休眠时段（0-23 时）。""" + return await _call("set_sleep", lambda dev: dev.sleep(on, start_h, end_h)) + + @mcp.tool() + async def clear_screen() -> dict: + """刷屏（EPD 清屏）。""" + return await _call("clear_screen", lambda dev: dev.clear()) + + @mcp.tool() + async def raw_send(data_hex: str, channel: str = "alt") -> dict: + """逃生舱：直发十六进制字节串；channel=rxtx/epd/alt。""" + data = bytes.fromhex(data_hex) + return await _call("raw_send", lambda dev: dev.raw(data, channel)) +``` + +注：`upload_image`/`countdown`/`raw_send` 在进入 `_call` 前的本地解析（decode/fromisoformat/fromhex）抛 `ValueError`——需被工具级兜底捕获。处理方式：这些工具体外不包 try 的话异常会炸 MCP 会话。把 `_call` 的 try 改为包住整个 lambda 构造不可行（解析发生在 `_call` 之外）。**实现时**把这三个工具改为： + +```python + @mcp.tool() + async def upload_image(source: str, slot: int = 0, algo: str = "atkinson", + mono: bool = False, **kw) -> dict: + """...""" + try: + adjust = {k: v for k, v in kw.items() if v is not None} + img = decode_image_source(source) + except ValueError as exc: + return _err("upload_image", "INVALID_PARAM", str(exc)) + return await _call("upload_image", lambda dev: dev.upload_image( + img, slot=slot, algo=algo, mono=mono, **adjust)) +``` + +`countdown`/`raw_send` 同理包 try（fromisoformat/fromhex 的 ValueError → INVALID_PARAM）。其余工具参数无需预解析，直接 `_call`。 + +⚠ `upload_image` 用 `**kw` 收 adjust 参数时，FastMCP 会从签名生成 schema——`**kw` 不进 schema。为保持 schema 显式，**保留显式参数签名**（threshold/diffusion/brightness/contrast/saturation/rotate 六个 `float | None = None`），仅把解析段包 try。以显式签名为准。 + +- [ ] **Step 4: 跑测试确认通过** + +Run: `.venv/bin/python -m pytest tests/test_mcp_tools.py -q` +Expected: 14 passed + +- [ ] **Step 5: 回归 + 提交** + +Run: `.venv/bin/python -m pytest tests/ -q` +Expected: 全绿 + +```bash +git add src/ppclock/mcp_tools.py tests/test_mcp_tools.py +git commit -m "feat(mcp): 13 个高层 MCP 工具 + 图源解码 + 统一输出契约" +``` + +--- + +### Task 3: mcp_server（装配入口 + HTTP token 鉴权 + 打包） + +**Files:** +- Create: `src/ppclock/mcp_server.py` +- Modify: `pyproject.toml`（extra + entry point） +- Test: `tests/test_mcp_server.py` + +**Interfaces:** +- Consumes: Task 1 `DeviceManager`、Task 2 `register_tools` +- Produces: + - `build_server(manager) -> FastMCP` + - `parse_args(argv=None) -> argparse.Namespace`（字段：`transport/host/port/mac/connect_timeout/idle_timeout`） + - `TokenAuthMiddleware(app, token)`（ASGI） + - `main(argv=None) -> None`（`ppclock-mcp` 入口） + - `TOKEN_ENV = "PPCLOCK_MCP_TOKEN"` + +- [ ] **Step 1: 写失败测试** + +```python +"""mcp_server 测试：参数解析 + 非回环强制 token + ASGI 鉴权中间件。""" +import asyncio + +import pytest + +from ppclock.mcp_server import (TOKEN_ENV, TokenAuthMiddleware, + parse_args) + + +class TestParseArgs: + def test_defaults(self): + a = parse_args([]) + assert a.transport == "stdio" + assert a.host == "127.0.0.1" + assert a.port == 8972 + assert a.mac is None + assert a.connect_timeout == 90.0 + assert a.idle_timeout == 300.0 + + def test_http_flags(self): + a = parse_args(["--transport", "http", "--host", "0.0.0.0", + "--port", "9000", "--mac", "AA:BB:CC:DD:EE:FF", + "--connect-timeout", "120", "--idle-timeout", "60"]) + assert a.transport == "http" and a.host == "0.0.0.0" + assert a.port == 9000 and a.mac == "AA:BB:CC:DD:EE:FF" + assert a.connect_timeout == 120.0 and a.idle_timeout == 60.0 + + +def run_middleware(token_set, auth_header): + """驱动 TokenAuthMiddleware，返回 (status, app_called)。""" + captured = {} + + async def app(scope, receive, send): + captured["called"] = True + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"ok"}) + + sent = [] + + async def send(msg): + sent.append(msg) + + headers = [] + if auth_header is not None: + headers.append((b"authorization", auth_header.encode())) + scope = {"type": "http", "headers": headers} + mw = TokenAuthMiddleware(app, token_set) + asyncio.run(mw(scope, None, send)) + status = next(m["status"] for m in sent if m["type"] == "http.response.start") + return status, captured.get("called", False) + + +class TestTokenAuth: + def test_no_header_401(self): + status, called = run_middleware("secret", None) + assert status == 401 and called is False + + def test_wrong_token_401(self): + status, called = run_middleware("secret", "Bearer nope") + assert status == 401 and called is False + + def test_right_token_passes(self): + status, called = run_middleware("secret", "Bearer secret") + assert status == 200 and called is True + + +class TestBindGuard: + def test_non_loopback_requires_token(self, monkeypatch): + from ppclock.mcp_server import main + monkeypatch.delenv(TOKEN_ENV, raising=False) + with pytest.raises(SystemExit): + main(["--transport", "http", "--host", "0.0.0.0"]) +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `.venv/bin/python -m pytest tests/test_mcp_server.py -q` +Expected: FAIL（`ModuleNotFoundError: No module named 'ppclock.mcp_server'`） + +- [ ] **Step 3: 实现 mcp_server.py + pyproject 打包** + +```python +"""ppclock-mcp —— MCP 服务器入口（stdio / streamable HTTP 双模）。 + +stdio：本机 agent 直挂（Claude Code/Desktop 等）。 +http ：局域网 agent 平台远程调用；绑非回环地址必须设 PPCLOCK_MCP_TOKEN。 +""" +from __future__ import annotations + +import argparse +import asyncio +import os + +from mcp.server.fastmcp import FastMCP + +from .device_manager import DeviceManager +from .mcp_tools import register_tools + +TOKEN_ENV = "PPCLOCK_MCP_TOKEN" + + +def build_server(manager: DeviceManager) -> FastMCP: + mcp = FastMCP("ppclock", stateless_http=True) + register_tools(mcp, manager) + return mcp + + +def parse_args(argv=None) -> argparse.Namespace: + p = argparse.ArgumentParser(prog="ppclock-mcp", + description="ppclock 墨水屏 MCP 服务器") + p.add_argument("--transport", choices=["stdio", "http"], default="stdio") + p.add_argument("--host", default="127.0.0.1") + p.add_argument("--port", type=int, default=8972) + p.add_argument("--mac", default=None, help="设备 MAC；缺省自动扫描 NRF- 前缀") + p.add_argument("--connect-timeout", type=float, default=90.0, + help="守候连接秒数（覆盖设备 30-60s 广播窗口）") + p.add_argument("--idle-timeout", type=float, default=300.0, + help="空闲断开秒数；0=每次调用独立连接") + return p.parse_args(argv) + + +class TokenAuthMiddleware: + """纯 ASGI 中间件：校验 Authorization: Bearer ，不符 401。""" + + def __init__(self, app, token: str): + self.app = app + self.token = token + + async def __call__(self, scope, receive, send): + if scope["type"] == "http": + headers = dict(scope.get("headers") or []) + auth = headers.get(b"authorization", b"").decode("latin1") + if auth != f"Bearer {self.token}": + await send({"type": "http.response.start", "status": 401, + "headers": [(b"content-type", b"text/plain")]}) + await send({"type": "http.response.body", + "body": b"unauthorized"}) + return + await self.app(scope, receive, send) + + +async def _run_http(mcp: FastMCP, host: str, port: int, token: str | None): + import uvicorn + app = mcp.streamable_http_app() + if token: + app = TokenAuthMiddleware(app, token) + config = uvicorn.Config(app, host=host, port=port, log_level="info") + await uvicorn.Server(config).serve() + + +def main(argv=None) -> None: + args = parse_args(argv) + manager = DeviceManager(mac=args.mac, + connect_timeout=args.connect_timeout, + idle_timeout=args.idle_timeout) + mcp = build_server(manager) + if args.transport == "stdio": + try: + mcp.run(transport="stdio") + finally: + asyncio.run(manager.close()) + return + token = os.environ.get(TOKEN_ENV) + loopback = args.host in ("127.0.0.1", "localhost", "::1") + if not loopback and not token: + raise SystemExit(f"绑定非回环地址 {args.host} 必须设置 {TOKEN_ENV}") + try: + asyncio.run(_run_http(mcp, args.host, args.port, token)) + finally: + asyncio.run(manager.close()) +``` + +pyproject.toml 修改（三处）： + +```toml +# [project.optional-dependencies] 下追加： +mcp = ["mcp>=1.10"] + +# [project.scripts] 下追加： +ppclock-mcp = "ppclock.mcp_server:main" +``` + +- [ ] **Step 4: 跑测试确认通过** + +Run: `.venv/bin/pip install -e ".[mcp]" -q && .venv/bin/python -m pytest tests/test_mcp_server.py -q` +Expected: 6 passed + +- [ ] **Step 5: stdio 冒烟（真实 MCP 握手，无硬件）** + +Run（应输出工具清单 JSON 后退出）: + +```bash +.venv/bin/python - <<'EOF' +import anyio +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client + +async def main(): + params = StdioServerParameters(command=".venv/bin/python", + args=["-m", "ppclock.mcp_server"]) + async with stdio_client(params) as (r, w): + async with ClientSession(r, w) as s: + await s.initialize() + tools = await s.list_tools() + names = sorted(t.name for t in tools.tools) + assert len(names) == 13, names + print(names) + +anyio.run(main) +EOF +``` + +Expected: 打印 13 个工具名。若无 `python -m` 入口（模块无 `__main__` 守卫），改用 `args=["-c", "from ppclock.mcp_server import main; main()"]` 并设 `env` 使 `PYTHONPATH=src`，或以 console script `.venv/bin/ppclock-mcp` 为 command。 + +- [ ] **Step 6: 回归 + 提交** + +Run: `.venv/bin/python -m pytest tests/ -q` +Expected: 全绿 + +```bash +git add src/ppclock/mcp_server.py tests/test_mcp_server.py pyproject.toml +git commit -m "feat(mcp): ppclock-mcp 入口——stdio/HTTP 双模 + token 鉴权 + 非回环守卫" +``` + +--- + +### Task 4: 文档与 0.2.0 + +**Files:** +- Create: `docs/sdk/MCP.md` +- Modify: `src/ppclock/__init__.py`（`__version__`）、`pyproject.toml`（version）、`CHANGELOG.md`、`README.md`、`llms.txt`、`docs/sdk/API.md` + +**Interfaces:** +- Consumes: Task 1-3 全部公开名 +- Produces: 文档与版本号；无新代码接口 + +- [ ] **Step 1: docs/sdk/MCP.md（工具参考 + agent 挂载示例）** + +内容骨架（逐项写全，勿留 TODO）： + +1. 一句话：ppclock-mcp 是跑在蓝牙所在机器上的单层 MCP 服务器，13 个工具覆盖对时/模式/传图/模板/倒计时/休眠等。 +2. 安装：`pip install -e ".[mcp]"`；要求 Python ≥3.10、本机蓝牙。 +3. 启动：stdio（`ppclock-mcp --mac XX`）与 http（`PPCLOCK_MCP_TOKEN=... ppclock-mcp --transport http --host 0.0.0.0 --port 8972`）双例。 +4. agent 挂载示例：Claude Code `claude mcp add ppclock -- ppclock-mcp --mac XX`（stdio）；streamable HTTP 端点 `http://:8972/mcp` + Bearer token。 +5. 13 工具参考表（名字/参数/返回，逐字对齐 mcp_tools.py docstring 与签名；`upload_image` 列出 6 个可选 adjust 参数）。 +6. 输出契约与错误码表（6 码 + hint 语义）。 +7. 连接行为说明：守候窗口/空闲超时/串行/重连一次（对应 `--connect-timeout`/`--idle-timeout`）。 +8. Windows 部署节：venv → `pip install -e ".[mcp]"` → 环境变量 → 启动 → 防火墙放行 8972（仅局域网）→ 不动 8971 bridge 与既有服务。 +9. 不暴露能力清单（OTA/激活/LUT/WiFi/轮播）及原因一句。 + +- [ ] **Step 2: 版本与既有文档同步** + +- `src/ppclock/__init__.py`：`__version__ = "0.2.0"` +- `pyproject.toml`：`version = "0.2.0"` +- `CHANGELOG.md`：新增 `## 0.2.0（2026-07-30）`——ppclock-mcp（13 工具/双模/鉴权/方案 C 连接管理）、DeviceManager 公开为子包能力。 +- `README.md`：SDK 用法后追加「MCP 服务器」小节（启动双例 + 指向 docs/sdk/MCP.md）。 +- `llms.txt`：目录行加 `ppclock-mcp`（mcp_server/device_manager/mcp_tools）；快速示例后加 MCP 挂载一句。 +- `docs/sdk/API.md`：子包速查表加三行（`ppclock.mcp_server`/`ppclock.device_manager`/`ppclock.mcp_tools`）。 + +- [ ] **Step 3: 回归 + 提交** + +Run: `.venv/bin/python -m pytest tests/ -q` +Expected: 全绿 + +```bash +git add -A +git commit -m "docs+release: MCP.md 工具参考与部署指南；0.2.0" +``` + +--- + +### Task 5: Windows 部署包与熠管家协作 + +**Files:** +- Create: `tools/mcp_windows_deploy.md`（熠管家执行清单） +- 外发：Worker Bridge `assist.requested` + +**Interfaces:** +- Consumes: Task 4 的 docs/sdk/MCP.md §Windows 部署 +- Produces: 可执行部署清单 + 熠管家 assist 请求记录 + +- [ ] **Step 1: 写 tools/mcp_windows_deploy.md** + +清单（每步含验证命令与预期输出）： + +1. 前置：Windows 机 Python ≥3.10（`python --version`）、蓝牙适配器在位、设备 NRF- 在附近。 +2. 取码：`git clone`（或压缩包拷贝）到 `C:\ppclock`；`cd C:\ppclock`。 +3. 环境：`python -m venv .venv; .venv\Scripts\pip install -e ".[mcp]"`。 +4. 冒烟（无设备）：`.venv\Scripts\ppclock-mcp --help` 打全参数表。 +5. stdio 实测：`claude mcp add ppclock -- C:\ppclock\.venv\Scripts\ppclock-mcp.exe --mac ` → 调 `scan_devices`/`set_time`/`upload_image`/`set_mode image0` 目视上屏。 +6. HTTP 实测：设 `PPCLOCK_MCP_TOKEN` → 启动 → 另一终端 `curl -H "Authorization: Bearer " http://127.0.0.1:8972/mcp` 初始化握手；无 token 请求预期 401。 +7. 约束核对：8971 bridge 与既有服务进程不动；8972 防火墙仅局域网放行。 +8. 回报格式：每步截图/文本输出 + 目视结果。 + +- [ ] **Step 2: 经 Worker Bridge 请求熠管家协助** + +```bash +W="$HOME/.local/bin/yi-worker-claude-current" +"$W" task.init ppclock-mcp-deploy '{"title":"ppclock-mcp Windows 部署实测","request_summary":"把 ppclock-mcp 部署到蓝牙所在 Windows 服务器并跑通 stdio+HTTP 实测（清单 tools/mcp_windows_deploy.md）","workspace":"/mnt/documents/Works/Eink/qianbian","native_session_id":"HOOK_INJECTED_EXACT_ID"}' +"$W" assist.requested ppclock-mcp-deploy '{"category":"action","question":"请按 tools/mcp_windows_deploy.md 在 Windows 服务器执行部署与实测，或给我该机的访问方式；约束：不动 8971 bridge 与既有服务","attempted":["本机 Linux 单元/协议测试全绿"],"needed":"Windows 执行环境或访问凭据","priority":"action_required"}' +``` + +（`native_session_id` 用 Hook 注入的原值；以上命令在 Task 5 执行时原样跑。） + +- [ ] **Step 3: 提交清单** + +```bash +git add tools/mcp_windows_deploy.md +git commit -m "docs: Windows 部署执行清单（熠管家协作）" +``` + +- [ ] **Step 4: 等复并执行联调** + +inbox 回复到达后按回复执行（拿到访问方式则远程部署；熠管家代办则核对回报与验证清单）。每条事件处理后 `event.ack`。此步产出实测记录追加到 `docs/superpowers/specs/2026-07-30-ppclock-mcp-design.md` 末尾「实测记录」节。 + +--- + +## Self-Review 记录 + +- Spec 覆盖：13 工具 ✅（Task 2）、双模+鉴权 ✅（Task 3）、方案 C ✅（Task 1）、打包 extra ✅（Task 3）、文档/版本 ✅（Task 4）、Windows+熠管家 ✅（Task 5）、Linux 兼容（bleak 抽象，无平台分支代码）✅。 +- 类型一致性：`DeviceManager.run(op)`/`status()` 签名在 Task 1 定义、Task 2 FakeManager 镜像；工具名 13 个 Task 2/3/4 一致；`TOKEN_ENV` Task 3 定义使用一致。 +- 已知风险（执行时验证）：FastMCP `stateless_http`/`streamable_http_app` 存在于 mcp≥1.10——Task 3 Step 2 失败则 `.venv/bin/pip install -U mcp` 并按实际 API 调整；`create_connected_server_and_client_session` 路径若变更，以 `python -c "import mcp.shared.memory; help(...)"` 为准。