Ble Toy

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

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

---

0. 成果预览

你对 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 档"]

最终你得到:

• 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):

uv venv --python 3.12
uv pip install winsdk

---

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

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 会把要广播的字节原样打进日志。做法:

# 电脑装好 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=强度)。下面是暮瑟实测可用的表(你的可能不同,以自己抓到的为准):

#!/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、放电脑旁几米内):

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

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

---

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

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:

   @echo off
   cd /d <项目目录>
   :loop
   .venv\Scripts\python.exe muse_server.py
   timeout /t 3 >nul
   goto loop
   

2. 建 launch_hidden.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 字节协议参考

---

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