ble-toy-mcp

# Tool List > For **MuSe / LoveSpouse / Leten** and other "broadcast-type" Bluetooth toys (common Apps: MuSe, LoveSpouse, Leten, etc.). > Goal: Use a Windows computer to let AI agents (e.g., RikkaHub) control toy gears autonomously through **MCP**. No need to connect to the toy, no need for jailbreak/root, and no need to capture HCI Bluetooth packets. ⚠️ **Pre-declaration**: This is for controlling devices **you own** and for personal learning and interoperability research. The protocol **has no authentication**, and any device that can send broadcasts can control toys within range; please do not expose the control endpoint to the public network (see "Security" section). ## 0.成果预览 ``` You tell AI: "Adjust to vibration level 5" / "Give a 3-second suction" → AI decides to call MCP tool set_vibration(5) → MCP service on your computer receives it → Sends a Bluetooth broadcast with custom data using Windows Bluetooth → Toy receives and locks in the gear ``` ```mermaid flowchart LR U["You tell AI: Adjust to vibration level 5"] --> AI["AI Agent RikkaHub"] AI -->|"MCP calls set_vibration"| S["Computer MCP Service<br/>muse_server.py"] S --> C["muse_control.py checks command table"] C -->|"winsdk BLE broadcast"| T["Toy: Lock in level 5"] ``` You will get: - `muse_control.py` — Command-line/library for direct toy control - `muse_server.py` — MCP service for any MCP client (RikkaHub/Claude/Cherry Studio…) ## 1. Principle: It's "broadcast-type", not "connection-type" Ordinary Bluetooth devices (headphones, fitness trackers) require pairing, establishing a connection, and GATT read/write. **These toys are different: They are passive receivers that only listen and do not connect.** They continuously scan surrounding BLE broadcast packets and execute commands when they find a broadcast with a specific **manufacturer data** that matches their own signature. Analogy: Like a TV remote control sending infrared signals — the transmitter only broadcasts, and the receiver acts on it without establishing a connection. A broadcast packet (11-byte manufacturer data) looks like this: ``` company_id = 0xFF 000000 6DB643xxxxxx427C XX YY ZZ └padding┘ └──── prefix(8B) ──┘ └command 3B┘ Derived from toy "address" and command ``` - **prefix** is the **identity signature of each toy** (derived from the toy's logical address). Other toys have different signatures, so your broadcasts will only be responded to by your toy. - The **last 3 bytes** encode "which motor + level + checksum", which is also different for each toy (including CRC). > This explains why no pairing is needed, no toy MAC is used, no GATT UUID is required, and no HCI packets need to be captured — because no connection is established. ## 2. Preparation | Required | Description | |---|---| | A **Windows 10/11 computer** | Necessary for Bluetooth LE antenna (WinRT API) | | Toy + Official App | For capturing real commands for your toy | | Android phone + **USB debugging** | For capturing App-printed commands using `adb logcat` | | Python 3.10~3.12 | Install `winsdk`; recommend using `uv` for a standalone environment | Setup environment (recommended `uv` to avoid polluting system Python): ```bash uv venv --python 3.12 uv pip install winsdk ``` ## 3. Key Step: Obtain Your Toy's Command Table ```mermaid flowchart TD A["Connect toy to phone, official App adjust levels"] --> B["Computer adb logcat capture logs"] B --> C["Search startAdvertising and extract broadcast bytes"] C --> D["Split: first 8 bytes = prefix, last 3 bytes = command"] D --> E["Fill into muse_control.py's MODE1/MODE2"] E --> F["winsdk broadcast verification, toy moves = success"] ``` All public implementations only hardcode one demo toy. You must obtain **your toy's** prefix and command table. Three ways, **prioritize the first**: > 💡 **No account/login/UserID information needed.** Command bytes are tied to **toy hardware**, not accounts — you're just reading your phone's logs and capturing your toy's bytes. Zero account and zero server sessions required. ### ① Directly capture using adb logcat (recommended) Many Apps print broadcast bytes directly to logs. Steps: ```bash # Install platform-tools (adb) on computer, enable USB debugging on phone, and authorize adb logcat -c # Clear logs adb logcat -v time > log.txt # Start recording (open another window) ``` Then connect to the toy using the phone App and adjust each level from low to high (stop for 2-3 seconds each). Search for broadcast characteristics in `log.txt`. For example, MuSe logs: ``` I/联网 蓝牙指令：31 : 6d b6 43 ce 97 fe 42 7c d4 1f 5d W/tag startAdvertising: 6db643ce97fe427cd41f5d ``` > Search keywords: `startAdvertising`, `advertis`, `manufacturer`, `6db643` (prefix feature), `蓝牙指令`, `ble`, `cmd`. > Each 11-byte packet has the first 8 bytes as prefix (same for all levels) and last 3 bytes as commands that change with levels — organize them into a table. ### ② Public command table (try your luck) GitHub `arz321/MuSe-Protocol`'s `MuSe_bleCommand.txt` lists complete 256 command tables for 6 toys. If your prefix matches one, use it directly. ### ③ Server/Reverse APK (advanced) - Family backends are often `*.zlmicro.com`, and Apps pull tables from `getproductdetail` (need packet capture and session). - Or use jadx to decompile APK, find "address → prefix + CRC" whitening algorithm in `BleIpaoUtils`, and calculate for any address. ## 4. Use Python for Broadcast Control (`muse_control.py`) Fill your captured commands into `MODE1`/`MODE2` (index 0=stop, 1~9=intensity). Below is a MuSe tested table (yours may differ): ```python #!/usr/bin/env python3 """MuSe/LoveSpouse Broadcast-type Toy Controller — Windows native (winsdk).""" from __future__ import annotations import sys, time try: sys.stdout.reconfigure(encoding="utf-8") # Prevent GBK console crash 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 + Your toy's prefix (8 bytes) COMPANY_ID = 0xFF # If not responding, try 0xFFF0 (see "Pitfalls") # Index 0=stop, 1~9=intensity. Replace with your captured bytes: MODE1 = ["d5964c","d41f5d","d7846f","d60d7e","d1b20a", "d03b1b","d3a029","d22938","dddec0","dc57d1"] # Vibration MODE2 = ["a5113f","a4982e","a7031c","a68a0d","a13579", "a0bc68","a3275a","a2ae4b","ad59b3","acd0a2"] # Suction/extension 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("Usage: 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() ``` Test (toy powered on, App closed, within a few meters of the computer): ```bash .venv/Scripts/python muse_control.py m1 3 # Vibration level 3, will lock in .venv/Scripts/python muse_control.py stop ``` Toy moves = success. It will **lock in** at that level until you send the next command. ## 5. Package as MCP Service (`muse_server.py`) Let AI call it. Use official MCP SDK's FastMCP, via 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) # All broadcasts in one thread, stable COM state 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: """Set vibration intensity and hold. level 0~9: 0=off, 9=strongest.""" lv = max(0, min(9, int(level))); await _run(_ctl.mode1, lv) return f"Vibration -> {lv}" @mcp.tool() async def set_suction(level: int) -> str: """Set suction/extension intensity and hold. level 0~9.""" lv = max(0, min(9, int(level))); await _run(_ctl.mode2, lv) return f"Suction/extension -> {lv}" @mcp.tool() async def pulse(target: str, level: int, seconds: float = 3.0) -> str: """Timed stimulation then auto-stop. 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} for {secs:g}s stopped" @mcp.tool() async def stop() -> str: """Stop all motors.""" await _run(_ctl.stop); return "All stopped" @mcp.tool() async def get_status() -> str: """Query current levels.""" return f"Vibration={_ctl.level1}, Suction/extension={_ctl.level2}" if __name__ == "__main__": mcp.run(transport="streamable-http") ``` Start: ```bash .venv/Scripts/python muse_server.py # Endpoint: http://<your IP>:8765/mcp ``` ## 6. Integrate with RikkaHub (or any MCP client) ```mermaid flowchart LR P["Phone RikkaHub (same WiFi)"] -->|"http LAN"| PC["Computer: MCP service + Bluetooth"] PC -.->|"BLE broadcast"| Toy["Toy"] P2["Phone (external network/4G)"] -.->|"Optional: via FRP"| VPS["VPS (only tunnel forwarding)"] VPS -.-> PC ``` RikkaHub → Settings → MCP → 「+」→ Select **Streamable HTTP**: - Name: `Toy` - URL: `http://<your computer's LAN IP>:8765/mcp` (phone and computer on the same WiFi) After saving, status changes to **Connected**, and 5 tools are displayed. **Long press assistant → edit → enable MCP**. Then tell AI "Adjust to vibration level 5", and it will automatically call `set_vibration(5)`. > Replace with Claude Desktop / Cherry Studio, etc., by filling in this URL. ## 7. Make it Resident (startup) BLE broadcast **must run in the logged-in desktop session** (cannot be installed as a system service/Docker). Simplest way: 1. Create a self-restarting batch `run_server.bat`: ```bat @echo off cd /d <project directory> :loop .venv\Scripts\python.exe muse_server.py timeout /t 3 >nul goto loop ``` 2. Create `launch_hidden.vbs` (hidden window launch): ```vbs CreateObject("WScript.Shell").Run "<project directory>\run_server.bat", 0, False ``` 3. Put `launch_hidden.vbs` in **startup folder** (`Win+R` → `shell:startup`) → log in and automatically run. If LAN IP changes: In the **router**, do **DHCP address reservation** (bind MAC → fixed IP), and URL remains permanent. ## 8. Common Pitfalls - **Toy not responding**: ①Toy not powered on/too far; ②Official App still connected and competing for broadcasts; ③Change `COMPANY_ID` from `0xFF` to `0xFFF0` (different implementation frameworks, ESP32 version uses the latter). - **Console Chinese error `UnicodeEncodeError`**: Script adds `sys.stdout.reconfigure(encoding="utf-8")`; or run `set PYTHONUTF8=1` before running. - **`python`/`pip` points to weird paths`: Windows Store placeholder hijacks; use standalone venv's full path `.venv\Scripts\python.exe` to bypass. - **Phone cannot connect to MCP**: Confirm phone's WiFi proxy not enabled; use IP instead of `xxx.local` (many phones/routers do not resolve mDNS); confirm firewall allows port: ``` netsh advfirewall firewall add rule name=muse-mcp dir=in action=allow protocol=TCP localport=8765 ``` ## 9. Security - The protocol **has no authentication**, and anyone within the broadcast range can control it. **Only use it within your home local area network, and avoid exposing it to the public internet using FRP/port forwarding.** - If you need to access it from outside the network: Use a VPS + FRP to create a tunnel (the VPS does not send Bluetooth, only forwards), and add a token to MCP (verify the `Authorization` header, RikkaHub supports custom headers). ## Appendix A: Advanced Direction - Reverse Engineering APK for a "Universal Algorithm" (⚠️ Directional guidance, not personally verified) > The main path ① (logcat) is sufficient, **unless**: The app does not print bytes (logcat can't catch it), or you want to **not capture packets and directly calculate the instructions of any toy by address**. > The following is just a direction, **the author has not personally verified it**, and does not guarantee success. If you're lazy, you can ignore this appendix. **Why it's worth it**: Currently, all public projects only hardcode one toy; reverse engineering the "universal algorithm" of "address → prefix + CRC" is a unique contribution. **Rough idea:** 1. **Get the APK**: Use `adb shell pm path <package name>` to find the base.apk, and then use `adb pull` to download it. 2. **Decompile**: Open it with jadx-gui (if the APK is large, give enough memory); or use apktool to extract smali. 3. **Locate key classes**: R8/ProGuard heavily obfuscates, and class and method names may become `a`/`b`/`c`, so don't expect to search by name. Instead, use **fixed string reverse query**: prefix constants like `6d b6 43` / `42 7c`, log tags (like `Bluetooth instructions` / `startAdvertising`), `ManufacturerData`, `BluetoothLeAdvertiser`, etc., and follow cross-references to find the packaging logic (i.e., `BleIpaoUtils` / `doubleCmdMap` mentioned earlier). 4. **Two segments to reverse**: - Toy address → **prefix middle 3 bytes** (whitening) - (address + instruction code) → **last 2 bytes** (疑似 CRC-16) - The first byte (channel + gear) is a fixed table, easy to recognize; the difficulty lies in whitening and CRC polynomial/initial value. 5. **Verification**: Rewrite the algorithm with Python, feed it a known toy address, and see if it can calculate the bytes you actually captured - if they match, you're done. **Pitfalls / Tips:** - The key algorithm **may be in native `.so`**, which requires Ghidra/IDA, more hardcore. - "Whitening" is likely BLE's **PN9 data whitening** + some **CRC-16**, can guess in this direction. - Again, **not personally verified**, this is guidance for those who want to dig deeper, not a step-by-step guide to success. ## 10. Acknowledgements / References This tutorial stands on the shoulders of these open-source reverse engineering works: - `arz321/MuSe-Protocol` — MuSe instruction table + toy library - `RevenantFreddy/pylovespouse` — winsdk broadcast Python implementation - `HackShiitake/LoveSpouse-Vibration-Controller` — Windows controller - `IngeniousKink/LVS-Gateway` — ESP32 gateway (instruction semantic annotation) - `51enuxu/sosexy-ble-control` — similar 12-byte protocol reference *The method is universal, and data is self-provided. Control your own devices, have fun, and be aware of security.*