把蓝牙玩具接入 AI —— 从协议逆向到 MCP 自主控制

你对 AI 说:"调到震动5档" / "来一下吮吸3秒"
   → AI 自己决定调用 MCP 工具 set_vibration(5)
   → 你电脑上的 MCP 服务收到
   → 用 Windows 蓝牙广播一段厂商数据
   → 玩具收到并锁存该档位

flowchart LR
    U["你对 AI 说：调到震动5档"] --> AI["AI 智能体 RikkaHub"]
    AI -->|"MCP 调用 set_vibration"| S["电脑 MCP 服务<br/>muse_server.py"]
    S --> C["muse_control.py 查指令表"]
    C -->|"winsdk BLE 广播"| T["玩具：锁存 5 档"]

company_id = 0xFF
000000  6DB643xxxxxx427C   XX YY ZZ
└填充┘ └──── prefix(8B) ──┘ └指令3B┘
        随玩具"地址"白化得到    档位+校验(CRC)

uv venv --python 3.12
uv pip install winsdk

flowchart TD
    A["手机连玩具，官方App按各档位"] --> B["电脑 adb logcat 抓日志"]
    B --> C["搜 startAdvertising 提取广播字节"]
    C --> D["拆分：前8字节=prefix，后3字节=指令"]
    D --> E["填进 muse_control.py 的 MODE1/MODE2"]
    E --> F["winsdk 广播验证，玩具动=成功"]

# 电脑装好 platform-tools(adb),手机开 USB 调试并授权
adb logcat -c                     # 清空日志
adb logcat -v time > log.txt      # 开始录制(另开一个窗口)

I/联网 蓝牙指令：31 : 6d b6 43 ce 97 fe 42 7c d4 1f 5d
W/tag startAdvertising: 6db643ce97fe427cd41f5d

#!/usr/bin/env python3
"""MuSe/LoveSpouse 广播型玩具控制器 —— Windows 原生(winsdk)。"""
from __future__ import annotations
import sys, time
try:
    sys.stdout.reconfigure(encoding="utf-8")   # 防 GBK 控制台中文崩溃
except Exception:
    pass
import winsdk.windows.devices.bluetooth.advertisement as _adv
import winsdk.windows.storage.streams as _streams
from winsdk.windows.devices.bluetooth.advertisement import (
    BluetoothLEAdvertisementPublisherStatus as _PubStatus,
)

PREFIX_HEX = "0000006db643ce97fe427c"   # 000000 + 你玩具的 prefix(8字节)
COMPANY_ID = 0xFF                        # 若不响应,试 0xFFF0(见"坑")
# 下标 0=停,1~9=强度。换成你自己抓到的字节:
MODE1 = ["d5964c","d41f5d","d7846f","d60d7e","d1b20a",
         "d03b1b","d3a029","d22938","dddec0","dc57d1"]   # 震动
MODE2 = ["a5113f","a4982e","a7031c","a68a0d","a13579",
         "a0bc68","a3275a","a2ae4b","ad59b3","acd0a2"]   # 吮吸/伸缩

class MuseController:
    def __init__(self, prefix_hex=PREFIX_HEX, company_id=COMPANY_ID):
        self._prefix = prefix_hex.replace(" ", "").lower()
        self._company_id = company_id
        self.level1 = 0; self.level2 = 0
    def mode1(self, level, duration=0.6):
        lv = max(0, min(9, int(level))); self._send(MODE1[lv], duration); self.level1 = lv
    def mode2(self, level, duration=0.6):
        lv = max(0, min(9, int(level))); self._send(MODE2[lv], duration); self.level2 = lv
    def stop(self, duration=0.6):
        self._send(MODE1[0], duration); self._send(MODE2[0], duration)
        self.level1 = 0; self.level2 = 0
    def _send(self, cmd_hex, duration):
        payload = bytearray.fromhex(self._prefix + cmd_hex)
        pub = _adv.BluetoothLEAdvertisementPublisher()
        md = _adv.BluetoothLEManufacturerData(); md.company_id = self._company_id
        w = _streams.DataWriter(); w.write_bytes(payload); md.data = w.detach_buffer()
        pub.advertisement.manufacturer_data.append(md)
        pub.start()
        while pub.status != _PubStatus.STARTED: time.sleep(0.01)
        time.sleep(max(0.05, duration)); pub.stop()

if __name__ == "__main__":
    c = MuseController()
    a = sys.argv[1:]
    if not a: print("用法: m1|m2 <0-9> | stop"); raise SystemExit
    if a[0]=="m1": c.mode1(int(a[1]) if len(a)>1 else 1)
    elif a[0]=="m2": c.mode2(int(a[1]) if len(a)>1 else 1)
    elif a[0]=="stop": c.stop()

.venv/Scripts/python muse_control.py m1 3   # 震动3档,会锁存
.venv/Scripts/python muse_control.py stop

uv pip install mcp uvicorn

#!/usr/bin/env python3
import asyncio, sys
from concurrent.futures import ThreadPoolExecutor
try: sys.stdout.reconfigure(encoding="utf-8")
except Exception: pass
from mcp.server.fastmcp import FastMCP
from muse_control import MuseController

HOST, PORT = "0.0.0.0", 8765
_ctl = MuseController()
_ex = ThreadPoolExecutor(max_workers=1)   # 所有广播在同一线程串行,COM状态稳定
mcp = FastMCP("muse-toy", host=HOST, port=PORT)

async def _run(fn, *args):
    await asyncio.get_running_loop().run_in_executor(_ex, fn, *args)

@mcp.tool()
async def set_vibration(level: int) -> str:
    """设置震动强度并保持。level 0~9:0=关闭,9=最强。"""
    lv = max(0, min(9, int(level))); await _run(_ctl.mode1, lv)
    return f"震动 -> {lv} 档"

@mcp.tool()
async def set_suction(level: int) -> str:
    """设置吮吸/伸缩强度并保持。level 0~9。"""
    lv = max(0, min(9, int(level))); await _run(_ctl.mode2, lv)
    return f"吮吸/伸缩 -> {lv} 档"

@mcp.tool()
async def pulse(target: str, level: int, seconds: float = 3.0) -> str:
    """定时刺激后自动停。target=vibration|suction, level 1~9, seconds<=30。"""
    lv = max(0, min(9, int(level))); secs = max(0.2, min(30.0, float(seconds)))
    setter = _ctl.mode1 if target.lower().startswith(("vib","震")) else _ctl.mode2
    await _run(setter, lv); await asyncio.sleep(secs); await _run(setter, 0)
    return f"{target} {lv}档 持续{secs:g}s 已停"

@mcp.tool()
async def stop() -> str:
    """停止所有马达。"""
    await _run(_ctl.stop); return "已全部停止"

@mcp.tool()
async def get_status() -> str:
    """查询当前档位。"""
    return f"震动={_ctl.level1}, 吮吸/伸缩={_ctl.level2}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

.venv/Scripts/python muse_server.py
# 端点: http://<本机IP>:8765/mcp

flowchart LR
    P["手机 RikkaHub（同一WiFi）"] -->|"http 局域网"| PC["电脑：MCP服务 + 蓝牙"]
    PC -.->|"BLE 广播"| Toy["玩具"]
    P2["手机（外网/4G）"] -.->|"可选：走 FRP"| VPS["VPS（仅隧道转发）"]
    VPS -.-> PC