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

> 面向 **MuSe / LoveSpouse / Leten** 这一类"广播型"蓝牙玩具(常见 App:暮瑟、LoveSpouse、Leten 等)。
> 目标:用一台 Windows 电脑,让 AI 智能体(如 RikkaHub)通过 **MCP** 自主控制玩具档位。全程不连接玩具、不需要越狱/root、不需要抓 HCI 蓝牙包。

⚠️ **前置声明**:仅用于控制**你自己拥有**的设备、用于个人学习与互操作研究。该协议**无任何鉴权**,任何能发广播的设备都能控制范围内的玩具;请勿把控制端点暴露到公网(详见"安全"一节)。

---

## 0. 成果预览

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

```mermaid
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 档"]
```

最终你得到:
- `muse_control.py` —— 命令行/库,直接控制玩具
- `muse_server.py` —— MCP 服务,任何 MCP 客户端(RikkaHub/Claude/Cherry Studio…)都能接

---

## 1. 原理:它是"广播型",不是"连接型"

普通蓝牙设备(耳机、手环)要先配对、建连接、走 GATT 读写。
**这类玩具不一样:它是个"只听不连"的被动接收器。** 它不停扫描周围的 BLE 广播包,一旦发现某个广播的"厂商数据(manufacturer data)"符合**它自己的签名**,就照着执行。

类比:像电视遥控器对空气发红外——发射端只管广播,接收端自己接,双方不建立连接。

一条广播包(11 字节厂商数据)长这样:

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

- **prefix** 是**每只玩具的身份签名**(由玩具的逻辑地址白化推导)。别的玩具签名不同,所以你的广播只有你这只会响应。
- **后 3 字节**编码"哪个马达 + 第几档 + 校验",每只玩具也不同(含 CRC)。

> 这也解释了:为什么不用配对、不用玩具 MAC、不用 GATT UUID、也不用抓 HCI 包——因为根本不建立连接。

---

## 2. 前置准备

| 需要 | 说明 |
|---|---|
| 一台 **Windows 10/11 电脑** | 必须,广播要靠它的蓝牙 LE 天线(WinRT API) |
| 玩具 + 官方 App | 用来抓你这只玩具的真实指令 |
| 安卓手机 + **USB 调试** | 用 `adb logcat` 抓 App 打印的指令(最省事的一条路) |
| Python 3.10~3.12 | 装 `winsdk`;推荐用 `uv` 建独立环境 |

装环境(推荐 uv,避免污染系统 Python):
```bash
uv venv --python 3.12
uv pip install winsdk
```

---

## 3. 关键一步:拿到"你这只玩具的指令表"

```mermaid
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 广播验证，玩具动=成功"]
```

所有公开实现都只硬编码了一只演示玩具。你必须拿到**自己这只**的 prefix + 指令。三条路,**优先第 ①**:

> 💡 **不需要任何账号 / 登录 / UserID 信息。** 指令字节是跟着**玩具硬件**走的,不是跟账号走的——你只是在读自己手机的日志、抓自己玩具的字节。全程零账号、零服务器会话。(只有下面进阶备选 ③ 才会用到登录会话,绝大多数人用不到。)

### ① adb logcat 直接抓(最省事,推荐)
很多这类 App 会把要广播的字节**原样打进日志**。做法:
```bash
# 电脑装好 platform-tools(adb),手机开 USB 调试并授权
adb logcat -c                     # 清空日志
adb logcat -v time > log.txt      # 开始录制(另开一个窗口)
```
然后在**手机 App 里连上玩具,把每个模式/每个档位从低到高按一遍**(每档停 2~3 秒)。
按完在 `log.txt` 里搜广播特征。以暮瑟为例,日志里直接有:
```
I/联网 蓝牙指令：31 : 6d b6 43 ce 97 fe 42 7c d4 1f 5d
W/tag startAdvertising: 6db643ce97fe427cd41f5d
```
> 搜索关键词建议:`startAdvertising`、`advertis`、`manufacturer`、`6db643`(prefix 特征)、`蓝牙指令`、`ble`、`cmd`。
> 每条 11 字节里,前 8 字节是 prefix(全档相同),后 3 字节随档位变——把它们按"模式+档位"整理成表即可。

### ② 公开指令表(碰运气)
GitHub `arz321/MuSe-Protocol` 里的 `MuSe_bleCommand.txt` 收录了 6 只玩具的完整 256 指令表。若你的 prefix 恰好和其中一只一致,直接抄表即可。

### ③ 服务器 / 逆向 APK(进阶)
- 该家族后端多为 `*.zlmicro.com`,App 会按玩具 barcode 从 `getproductdetail` 拉表(需抓包 + 会话)。
- 或用 jadx 反编译 APK,找 `BleIpaoUtils` 里"地址→prefix+CRC"的白化算法,给任意地址现算。

---

## 4. 用 Python 广播控制(`muse_control.py`)

把你抓到的指令填进 `MODE1`/`MODE2`(下标 0=停,1~9=强度)。下面是暮瑟实测可用的表(你的可能不同,以自己抓到的为准):

```python
#!/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()
```

测试(玩具开机、关掉 App、放电脑旁几米内):
```bash
.venv/Scripts/python muse_control.py m1 3   # 震动3档,会锁存
.venv/Scripts/python muse_control.py stop
```
玩具动了 = 成功。它会**锁存**在该档位,直到你发下一条,所以发一小下就够。

---

## 5. 包成 MCP 服务(`muse_server.py`)

让 AI 能调用。用官方 MCP SDK 的 FastMCP,走 Streamable HTTP:
```bash
uv pip install mcp uvicorn
```
```python
#!/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")
```
启动:
```bash
.venv/Scripts/python muse_server.py
# 端点: http://<本机IP>:8765/mcp
```

---

## 6. 接入 RikkaHub(或任意 MCP 客户端)

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

RikkaHub → 设置 → MCP → 「+」→ 选 **Streamable HTTP**:
- 名称:`玩具`
- URL:`http://<电脑局域网IP>:8765/mcp`(手机与电脑同一 WiFi)

保存后状态变 **Connected**,会同步出 5 个工具。再 **长按助手 → 编辑 → 启用该 MCP**。之后对 AI 说"调到震动5档",它就会自己调 `set_vibration(5)`。

> 换成 Claude Desktop / Cherry Studio 等也一样,填这个 URL 即可。

---

## 7. 让它常驻(开机自启)

BLE 广播**必须在登录后的桌面会话**里跑(装成系统服务/Docker 拿不到蓝牙)。最简单:
1. 建一个自重启批处理 `run_server.bat`:
   ```bat
   @echo off
   cd /d <项目目录>
   :loop
   .venv\Scripts\python.exe muse_server.py
   timeout /t 3 >nul
   goto loop
   ```
2. 建 `launch_hidden.vbs`(隐藏窗口启动):
   ```vbs
   CreateObject("WScript.Shell").Run "<项目目录>\run_server.bat", 0, False
   ```
3. 把 `launch_hidden.vbs` 放进**启动文件夹**(`Win+R` → `shell:startup`)→ 登录即自动运行。

局域网 IP 会变的话:在**路由器**里给这台电脑做 **DHCP 地址保留**(绑 MAC → 固定 IP),URL 就永久不变。

---

## 8. 常见坑

- **玩具不响应**:①玩具没开机/太远;②官方 App 还连着在抢广播,先关 App;③把 `COMPANY_ID` 从 `0xFF` 换成 `0xFFF0`(不同实现的厂商 ID 框架不同,ESP32 版用后者)。
- **控制台中文报错 `UnicodeEncodeError`**:脚本已加 `sys.stdout.reconfigure(encoding="utf-8")`;或运行前 `set PYTHONUTF8=1`。
- **`python`/`pip` 指向诡异路径**:Windows 商店占位符劫持,用独立 venv 的全路径 `.venv\Scripts\python.exe` 即可绕开。
- **手机连不上 MCP**:先确认手机没开着废弃的 WiFi 代理;用 IP 而非 `xxx.local`(很多手机/路由不解析 mDNS);确认防火墙放行端口:
  ```
  netsh advfirewall firewall add rule name=muse-mcp dir=in action=allow protocol=TCP localport=8765
  ```

## 9. 安全

- 该协议**无鉴权**,广播范围内谁都能控。**只在家里局域网用,别用 FRP/端口转发暴露到公网。**
- 若确需外网:用 VPS + FRP 做隧道(VPS 不发蓝牙,只转发),并给 MCP 加一个 token(校验 `Authorization` header,RikkaHub 支持自定义 header)。

## 附录 A:进阶方向 —— 逆向 APK 求"通用算法"(⚠️ 方向指引,未亲测)

> 主路径 ①(logcat)已经够用,**除非**:App 不打印字节(logcat 抓不到),或你想**不抓包、按地址直接算出任意玩具的指令**。
> 下面只是方向,**作者没亲自做通**,不保证顺利。懒得折腾就忽略本附录。

**为什么值得**:目前所有公开项目都只硬编码了一只玩具;把"地址 → prefix + CRC"的**通用算法**逆出来,是全网独一份。

**大致思路:**
1. **拿 APK**:`adb shell pm path <包名>` 找到 base.apk,`adb pull` 下来。
2. **反编译**:jadx-gui 打开(APK 大就给足内存);或 apktool 出 smali。
3. **定位关键类**:R8/ProGuard 重度混淆,类名方法名会变成 `a`/`b`/`c`,别指望按名字搜。改用**固定字符串反查**:prefix 里恒定的 `6d b6 43` / `42 7c`、日志 tag(如 `蓝牙指令`/`startAdvertising`)、`ManufacturerData`、`BluetoothLeAdvertiser` 等,顺着交叉引用找到组包逻辑(即之前提到的 `BleIpaoUtils`/`doubleCmdMap` 这类)。
4. **要逆的两段**:
   - 玩具地址 → **prefix 中间 3 字节**(白化)
   - (地址 + 指令码) → **后 2 字节**(疑似 CRC-16)
   - 第 1 字节(通道+档位)是固定表,好认;难在白化和 CRC 的多项式/初值。
5. **验证**:用 Python 重写算法,喂已知的一只玩具地址,看能否算出你实抓到的字节——对得上就成了。

**坑 / 提示:**
- 关键算法**可能在 native `.so`** 里,那得上 Ghidra/IDA,更硬核。
- "白化"很可能是 BLE 的 **PN9 数据白化** + 某种 **CRC-16**,可往这个方向猜。
- 再次强调:**未亲测**,是给想深挖的人指路,不是保证能照抄成功的步骤。

## 10. 致谢 / 参考

本教程站在这些开源逆向工作的肩膀上:
- `arz321/MuSe-Protocol` —— MuSe 指令表 + 玩具库
- `RevenantFreddy/pylovespouse` —— winsdk 广播的 Python 实现
- `HackShiitake/LoveSpouse-Vibration-Controller` —— Windows 控制器
- `IngeniousKink/LVS-Gateway` —— ESP32 网关(指令语义注释)
- `51enuxu/sosexy-ble-control` —— 同类 12 字节协议参考

---

*方法通用,数据自备。控制你自己的设备,玩得开心也注意安全。*