qianbian/docs/superpowers/plans/2026-07-30-ppclock-mcp.md

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": <name>, "data": {...}}；失败 {"ok": false, "tool": <name>, "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: 写失败测试

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

"""设备连接管理器（方案 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 全绿

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: 写失败测试

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

"""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 之外）。实现时把这三个工具改为：

    @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: 全绿

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: 写失败测试

"""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 打包

"""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 <token>，不符 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 修改（三处）：

# [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 后退出）:

.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: 全绿

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://<host>: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: 全绿

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 <MAC> → 调 scan_devices/set_time/upload_image/set_mode image0 目视上屏。
6. HTTP 实测：设 PPCLOCK_MCP_TOKEN → 启动 → 另一终端 curl -H "Authorization: Bearer <token>" http://127.0.0.1:8972/mcp 初始化握手；无 token 请求预期 401。
7. 约束核对：8971 bridge 与既有服务进程不动；8972 防火墙仅局域网放行。
8. 回报格式：每步截图/文本输出 + 目视结果。

• Step 2: 经 Worker Bridge 请求熠管家协助

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: 提交清单

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(...)" 为准。