warehack.ing/mcbluetooth
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: warehack.ing:45c1e57dfe1cfb11a50c7405180efe2951e7f277
Branches Tags
warehack.ing:main
..
compare: warehack.ing:914db838d71bdd18d2b8530744cd5cec93f6ad1d
Branches Tags
warehack.ing:main

No commits in common. "45c1e57dfe1cfb11a50c7405180efe2951e7f277" and "914db838d71bdd18d2b8530744cd5cec93f6ad1d" have entirely different histories.
45c1e57dfe ... 914db838d7

18 changed files with 44 additions and 3520 deletions
Show Stats Expand all files Collapse all files

6
README.md
View File

@ -5,11 +5,11 @@
</p>
<p align="center">
<a href="https://mcbluetooth.supported.systems">Documentation</a> •
<a href="#installation">Installation</a> •
<a href="#quick-start">Quick Start</a> •
<a href="#features">Features</a> •
<a href="#tools-reference">Tools</a>
<a href="#tools-reference">Tools</a> •
<a href="#architecture">Architecture</a>
</p>
<p align="center">
@ -509,7 +509,7 @@ mcbluetooth/
## Related Projects
- [mcbluetooth-esp32](https://git.supported.systems/MCP/mcbluetooth-esp32) — ESP32 test harness for automated E2E Bluetooth testing with mcbluetooth
- [mcbluetooth-esp32](https://github.com/supported-systems/mcbluetooth-esp32) — ESP32 test harness for automated E2E Bluetooth testing with mcbluetooth
- [FastMCP](https://gofastmcp.com) — The MCP framework powering this server
- [BlueZ](http://www.bluez.org/) — The official Linux Bluetooth stack
- [dbus-fast](https://github.com/bluetooth-devices/dbus-fast) — Fast D-Bus client for Python

12
docs-site/Dockerfile
View File

@ -1,12 +0,0 @@
# Build stage
FROM node:22-alpine AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build
# Serve stage
FROM caddy:2-alpine
COPY --from=builder /app/dist /srv
CMD ["caddy", "file-server", "--root", "/srv", "--listen", ":80"]

2
docs-site/astro.config.mjs
View File

@ -6,7 +6,7 @@ import icon from 'astro-icon';
// https://astro.build/config
export default defineConfig({
// Site URL for sitemap generation
site: 'https://mcbluetooth.supported.systems',
site: 'https://mcbluetooth.example.com',
// Disable telemetry
telemetry: false,

13
docs-site/docker-compose.yml
View File

@ -1,13 +0,0 @@
services:
docs:
build: .
restart: unless-stopped
networks:
- caddy
labels:
caddy: ${DOCS_DOMAIN:-mcbluetooth.supported.systems}
caddy.reverse_proxy: "{{upstreams 80}}"
networks:
caddy:
external: true

63
docs-site/src/content/docs/guides/ble.md
View File

@ -132,7 +132,7 @@ bt_ble_write ... with_response=false
## Notifications
BLE notifications let devices push data when values change — essential for real-time sensor data like heart rate, temperature, or button presses.
Subscribe to value changes:
### Enable Notifications
@ -140,67 +140,8 @@ BLE notifications let devices push data when values change — essential for rea
bt_ble_notify adapter="hci0" address="..." char_uuid="00002a37-..." enable=true
```
Returns:
```json
{
"status": "notifications_enabled",
"uuid": "00002a37-0000-1000-8000-00805f9b34fb",
"uuid_short": "0x2A37",
"resource_uri": "bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications",
"history_uri": "bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications/history"
}
```
### Reading Notification Values
Notification values are automatically buffered (up to 100) and accessible via MCP resources:
<Aside type="tip">
The `resource_uri` and `history_uri` returned by `bt_ble_notify` can be used directly with `ReadMcpResource` — no need to construct the URI manually.
</Aside>
```
# Latest value and stats
ReadMcpResource uri="bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications"
# Buffered history
ReadMcpResource uri="bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications/history"
```
**Latest notification resource:**
```json
{
"address": "AA:BB:CC:DD:EE:FF",
"characteristic_uuid": "00002a37-...",
"notifying": true,
"latest": {
"timestamp": "2026-02-09T20:05:39.091206+00:00",
"value_hex": "1648",
"value_bytes": [22, 72]
},
"buffer_count": 15,
"total_received": 47
}
```
### Check Active Subscriptions
```
bt_ble_notification_status
```
Returns all active notification subscriptions with buffer stats.
### Clear Notification Buffer
```
bt_ble_clear_notification_buffer adapter="hci0" address="..." char_uuid="..."
```
Clears buffered values while keeping the subscription active.
<Aside type="note">
Notification buffers persist for the MCP server session lifetime. If the server restarts, subscriptions must be re-enabled.
After enabling notifications, the device will send updates when values change. Currently, mcbluetooth enables the notification mode but doesn't provide a callback mechanism — use protocol capture to see the actual notifications.
</Aside>
### Disable Notifications

82
docs-site/src/content/docs/reference/ble-tools.md
View File

@ -199,7 +199,7 @@ bt_ble_write adapter="hci0" address="..." char_uuid="..." value="01" with_respon
## bt_ble_notify
Enable or disable notifications for a characteristic. When enabled, notification values are automatically buffered and accessible via MCP resources.
Enable or disable notifications for a characteristic.
**Parameters:**
@ -210,21 +210,10 @@ Enable or disable notifications for a characteristic. When enabled, notification
| `char_uuid` | string | Yes | Characteristic UUID |
| `enable` | boolean | Yes | true to enable, false to disable |
**Returns (enable=true):**
**Returns:**
```json
{
"status": "notifications_enabled",
"uuid": "00002a37-0000-1000-8000-00805f9b34fb",
"uuid_short": "0x2A37",
"resource_uri": "bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications",
"history_uri": "bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications/history"
}
```
**Returns (enable=false):**
```json
{
"status": "notifications_disabled",
"uuid": "00002a37-0000-1000-8000-00805f9b34fb"
}
```
@ -234,77 +223,14 @@ Enable or disable notifications for a characteristic. When enabled, notification
# Enable heart rate notifications
bt_ble_notify adapter="hci0" address="..." char_uuid="00002a37-0000-1000-8000-00805f9b34fb" enable=true
# Read notification values via MCP resource
ReadMcpResource uri="bluetooth://ble/.../notifications"
# Disable notifications
bt_ble_notify adapter="hci0" address="..." char_uuid="..." enable=false
```
**Notes:**
- Characteristic must have `notify` flag
- Notifications are buffered (up to 100 values) in a circular buffer
- Access buffered values via `resource_uri` or `history_uri`
---
## bt_ble_notification_status
List all active BLE notification subscriptions.
**Parameters:** None
**Returns:**
```json
{
"count": 2,
"subscriptions": [
{
"address": "AA:BB:CC:DD:EE:FF",
"char_uuid": "00002a37-0000-1000-8000-00805f9b34fb",
"uuid_short": "0x2A37",
"notifying": true,
"buffer_count": 15,
"total_received": 47,
"resource_uri": "bluetooth://ble/.../notifications",
"history_uri": "bluetooth://ble/.../notifications/history"
}
]
}
```
**Example:**
```
bt_ble_notification_status
```
---
## bt_ble_clear_notification_buffer
Clear the notification buffer for a characteristic while keeping the subscription active.
**Parameters:**
| Name | Type | Required | Description |
|------|------|----------|-------------|
| `adapter` | string | Yes | Adapter name |
| `address` | string | Yes | Device MAC address |
| `char_uuid` | string | Yes | Characteristic UUID |
**Returns:**
```json
{
"status": "buffer_cleared",
"uuid": "00002a37-0000-1000-8000-00805f9b34fb",
"cleared_count": 15
}
```
**Example:**
```
bt_ble_clear_notification_buffer adapter="hci0" address="AA:BB:CC:DD:EE:FF" char_uuid="00002a37-..."
```
- Notifications are delivered via GATT protocol
- Use protocol capture to see notification data
---

85
docs-site/src/content/docs/reference/resources.md
View File

@ -16,9 +16,6 @@ mcbluetooth exposes live Bluetooth state through MCP resources. These provide re
| `bluetooth://trusted` | Trusted devices |
| `bluetooth://adapter/{name}` | Specific adapter details |
| `bluetooth://device/{address}` | Specific device details |
| `bluetooth://ble/notifications` | All active BLE notification subscriptions |
| `bluetooth://ble/{address}/{uuid}/notifications` | Latest notification value and stats |
| `bluetooth://ble/{address}/{uuid}/notifications/history` | Buffered notification history |
## Adapter Resources
@ -198,85 +195,3 @@ Resources can be read using the standard MCP resource protocol:
- Connecting/disconnecting
- Sending files
- Changing settings
## BLE Notification Resources
These resources provide access to buffered BLE GATT notification values.
### bluetooth://ble/notifications
Lists all active notification subscriptions.
**Response:**
```json
{
"count": 1,
"subscriptions": [
{
"address": "AA:BB:CC:DD:EE:FF",
"char_uuid": "00002a37-0000-1000-8000-00805f9b34fb",
"char_path": "/org/bluez/hci0/dev_.../service.../char...",
"notifying": true,
"buffer_count": 15,
"total_received": 47,
"uuid_short": "0x2A37"
}
]
}
```
### bluetooth://ble/{address}/{uuid}/notifications
Get latest notification value and buffer statistics.
**URI:** `bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-0000-1000-8000-00805f9b34fb/notifications`
**Response:**
```json
{
"address": "AA:BB:CC:DD:EE:FF",
"characteristic_uuid": "00002a37-0000-1000-8000-00805f9b34fb",
"characteristic_uuid_short": "0x2A37",
"notifying": true,
"latest": {
"timestamp": "2026-02-09T20:05:39.091206+00:00",
"value_hex": "1648",
"value_bytes": [22, 72]
},
"buffer_count": 15,
"total_received": 47
}
```
### bluetooth://ble/{address}/{uuid}/notifications/history
Get buffered notification history (up to 100 values).
**URI:** `bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications/history?count=10`
**Response:**
```json
{
"address": "AA:BB:CC:DD:EE:FF",
"characteristic_uuid": "00002a37-...",
"count": 10,
"values": [
{
"timestamp": "2026-02-09T20:05:39.091206+00:00",
"value_hex": "1648",
"value_bytes": [22, 72]
},
{
"timestamp": "2026-02-09T20:05:38.091206+00:00",
"value_hex": "1647",
"value_bytes": [22, 71]
}
]
}
```
**Notes:**
- Enable notifications with `bt_ble_notify` first
- Notifications are buffered in a circular buffer (max 100 values)
- Buffer persists for the MCP server session lifetime
- Use `bt_ble_clear_notification_buffer` to clear without disabling

5
pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "mcbluetooth"
version = "2026.02.09"
version = "2026.02.04"
description = "Comprehensive BlueZ MCP server - expose the full Linux Bluetooth stack to LLMs"
readme = "README.md"
requires-python = ">=3.11"
@ -43,7 +43,6 @@ dev = [
mcbluetooth = "mcbluetooth:main"
[project.urls]
Documentation = "https://mcbluetooth.supported.systems"
Repository = "https://git.supported.systems/MCP/mcbluetooth"
[tool.hatch.build.targets.wheel]
@ -69,8 +68,6 @@ ignore = ["E501"]
[tool.ruff.lint.per-file-ignores]
# dbus-fast uses D-Bus type signatures ("o", "h", "a{sv}") as annotations
"src/mcbluetooth/hfp_ag.py" = ["F821", "F722"]
"src/mcbluetooth/gatt_server.py" = ["F821", "F722"]
"src/mcbluetooth/spp.py" = ["F821", "F722"]
[tool.pytest.ini_options]
asyncio_mode = "auto"

144
src/mcbluetooth/ble_subscriptions.py
View File

@ -1,144 +0,0 @@
"""MCP subscription manager for BLE notifications.
This module bridges BlueZ D-Bus notification signals to MCP resource subscriptions.
When a BLE device sends a GATT notification, subscribed MCP clients receive a
`notifications/resources/updated` message for the corresponding resource URI.
"""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from fastmcp import FastMCP
from mcbluetooth.dbus_client import BLENotifyManager
logger = logging.getLogger(__name__)
def make_notification_uri(address: str, char_uuid: str) -> str:
"""Construct the MCP resource URI for a notification subscription.
Args:
address: Device Bluetooth address (e.g., "AA:BB:CC:DD:EE:FF")
char_uuid: Characteristic UUID (full form)
Returns:
Resource URI like "bluetooth://ble/AA:BB:CC:DD:EE:FF/00002a37-.../notifications"
"""
# Normalize address (uppercase, colons)
addr_normalized = address.upper()
# Keep UUID lowercase for consistency
uuid_normalized = char_uuid.lower()
return f"bluetooth://ble/{addr_normalized}/{uuid_normalized}/notifications"
def parse_notification_uri(uri: str) -> tuple[str, str] | None:
"""Parse a notification resource URI to extract address and char_uuid.
Args:
uri: Resource URI
Returns:
Tuple of (address, char_uuid) or None if not a valid notification URI
"""
prefix = "bluetooth://ble/"
suffix = "/notifications"
if not uri.startswith(prefix) or not uri.endswith(suffix):
return None
# Extract the middle part: "AA:BB:CC:DD:EE:FF/uuid"
middle = uri[len(prefix) : -len(suffix)]
parts = middle.split("/")
if len(parts) != 2:
return None
address, char_uuid = parts
return address, char_uuid
class BLESubscriptionManager:
"""Manages MCP subscriptions for BLE notification resources.
This class:
- Tracks which resource URIs have been subscribed to
- Sends resource_updated notifications when D-Bus signals arrive
- Handles automatic cleanup
"""
def __init__(self, mcp: FastMCP, notify_manager: BLENotifyManager):
self._mcp = mcp
self._notify_manager = notify_manager
self._subscribed_uris: set[str] = set()
self._event_loop: asyncio.AbstractEventLoop | None = None
# Register callback to receive notification events
notify_manager.on_notification(self._on_ble_notification)
def _on_ble_notification(self, address: str, char_uuid: str, value: bytes) -> None:
"""Callback when a BLE notification is received.
This is called synchronously from the D-Bus signal handler.
We schedule the async MCP notification on the event loop.
"""
uri = make_notification_uri(address, char_uuid)
# Check if anyone is subscribed
if uri not in self._subscribed_uris:
return
# Schedule the resource update notification
if self._event_loop is None:
try:
self._event_loop = asyncio.get_running_loop()
except RuntimeError:
logger.warning("No event loop available for notification dispatch")
return
# Fire-and-forget the notification
self._event_loop.call_soon_threadsafe(
lambda: asyncio.create_task(self._send_resource_updated(uri))
)
async def _send_resource_updated(self, uri: str) -> None:
"""Send a resource_updated notification to MCP clients."""
try:
# Access the underlying MCP server to send notifications
# FastMCP exposes this through _mcp_server
server = getattr(self._mcp, "_mcp_server", None)
if server is None:
logger.debug("MCP server not available for resource notifications")
return
# The MCP protocol uses notifications/resources/updated
# We need to send this through the server's notification mechanism
await server.request_context.session.send_resource_updated(uri)
logger.debug(f"Sent resource_updated for {uri}")
except Exception as e:
# Don't crash on notification failures
logger.debug(f"Failed to send resource_updated: {e}")
def subscribe(self, uri: str) -> None:
"""Mark a URI as subscribed (for tracking purposes)."""
self._subscribed_uris.add(uri)
def unsubscribe(self, uri: str) -> None:
"""Mark a URI as unsubscribed."""
self._subscribed_uris.discard(uri)
def is_subscribed(self, uri: str) -> bool:
"""Check if a URI is currently subscribed."""
return uri in self._subscribed_uris
def list_subscriptions(self) -> list[str]:
"""List all subscribed URIs."""
return list(self._subscribed_uris)
def clear_subscriptions(self) -> None:
"""Clear all subscriptions."""
self._subscribed_uris.clear()

239
src/mcbluetooth/dbus_client.py
View File

@ -18,10 +18,7 @@ Object paths follow this pattern:
from __future__ import annotations
from collections import deque
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from dbus_fast import BusType, Variant
@ -137,219 +134,6 @@ class GattCharacteristic:
notifying: bool = False
@dataclass
class NotificationValue:
"""A single BLE notification value with timestamp."""
timestamp: datetime
value: bytes
def to_dict(self) -> dict[str, Any]:
"""Convert to JSON-serializable dict."""
return {
"timestamp": self.timestamp.isoformat(),
"value_hex": self.value.hex(),
"value_bytes": list(self.value),
}
@dataclass
class NotificationBuffer:
"""Circular buffer for BLE notification values."""
address: str
char_uuid: str
char_path: str
max_size: int = 100
values: deque[NotificationValue] = field(default_factory=lambda: deque(maxlen=100))
total_received: int = 0
notifying: bool = True
def __post_init__(self) -> None:
# Ensure deque has correct maxlen
if not isinstance(self.values, deque) or self.values.maxlen != self.max_size:
self.values = deque(maxlen=self.max_size)
def add(self, value: bytes) -> NotificationValue:
"""Add a notification value to the buffer."""
notification = NotificationValue(
timestamp=datetime.now(UTC),
value=value,
)
self.values.append(notification)
self.total_received += 1
return notification
@property
def latest(self) -> NotificationValue | None:
"""Get the most recent notification value."""
return self.values[-1] if self.values else None
def get_history(self, count: int = 10) -> list[NotificationValue]:
"""Get the most recent N notification values."""
return list(self.values)[-count:]
# Type alias for notification callbacks
NotificationCallback = Callable[[str, str, bytes], None] # (address, char_uuid, value)
class BLENotifyManager:
"""Manages BLE GATT notification subscriptions and buffers.
This class handles:
- D-Bus signal subscriptions for PropertiesChanged on characteristics
- Circular buffers for notification values per characteristic
- Callbacks for notifying higher layers of new values
"""
def __init__(self, client: BlueZClient):
self._client = client
self._buffers: dict[str, NotificationBuffer] = {} # char_path -> buffer
self._signal_handlers: dict[str, Any] = {} # char_path -> handler removal func
self._callbacks: list[NotificationCallback] = []
def on_notification(self, callback: NotificationCallback) -> None:
"""Register a callback for notification events.
The callback receives (address, char_uuid, value) when a notification arrives.
"""
self._callbacks.append(callback)
def remove_callback(self, callback: NotificationCallback) -> None:
"""Remove a previously registered callback."""
if callback in self._callbacks:
self._callbacks.remove(callback)
async def subscribe(self, char_path: str, address: str, char_uuid: str) -> NotificationBuffer:
"""Subscribe to D-Bus signals for a characteristic.
Args:
char_path: D-Bus object path of the characteristic
address: Device Bluetooth address
char_uuid: Characteristic UUID
Returns:
The notification buffer for this characteristic
"""
# Create or return existing buffer
if char_path in self._buffers:
self._buffers[char_path].notifying = True
return self._buffers[char_path]
buffer = NotificationBuffer(
address=address,
char_uuid=char_uuid,
char_path=char_path,
)
self._buffers[char_path] = buffer
# Subscribe to PropertiesChanged signal
await self._subscribe_to_signal(char_path)
return buffer
async def _subscribe_to_signal(self, char_path: str) -> None:
"""Subscribe to D-Bus PropertiesChanged signal for a characteristic."""
await self._client._ensure_connected()
assert self._client._bus is not None
try:
introspection = await self._client._bus.introspect(BLUEZ_SERVICE, char_path)
proxy = self._client._bus.get_proxy_object(BLUEZ_SERVICE, char_path, introspection)
props_iface = proxy.get_interface(DBUS_PROPS_IFACE)
def on_properties_changed(
interface: str,
changed: dict[str, Any],
invalidated: list[str],
) -> None:
"""Handle PropertiesChanged signal."""
if interface != BLUEZ_GATT_CHAR_IFACE:
return
if "Value" in changed:
value = changed["Value"]
# Unwrap Variant if needed
if isinstance(value, Variant):
value = value.value
# Convert to bytes
raw_bytes = bytes(value) if isinstance(value, (list, bytearray)) else value
buffer = self._buffers.get(char_path)
if buffer:
buffer.add(raw_bytes)
# Fire callbacks
for cb in self._callbacks:
try:
cb(buffer.address, buffer.char_uuid, raw_bytes)
except Exception:
pass # Don't let callback errors break signal handling
# Connect signal handler
props_iface.on_properties_changed(on_properties_changed)
self._signal_handlers[char_path] = (props_iface, on_properties_changed)
except Exception as e:
# Clean up buffer if signal subscription fails
if char_path in self._buffers:
del self._buffers[char_path]
raise RuntimeError(f"Failed to subscribe to notifications: {e}") from e
async def unsubscribe(self, char_path: str) -> None:
"""Unsubscribe from D-Bus signals for a characteristic.
Note: This doesn't stop notifications on the device - use BlueZClient.stop_notify()
for that. This just stops receiving signals in this manager.
"""
if char_path in self._signal_handlers:
props_iface, handler = self._signal_handlers.pop(char_path)
props_iface.off_properties_changed(handler)
if char_path in self._buffers:
self._buffers[char_path].notifying = False
def get_buffer(self, char_path: str) -> NotificationBuffer | None:
"""Get the notification buffer for a characteristic."""
return self._buffers.get(char_path)
def get_buffer_by_address_uuid(self, address: str, char_uuid: str) -> NotificationBuffer | None:
"""Find buffer by device address and characteristic UUID."""
address_upper = address.upper()
char_uuid_lower = char_uuid.lower()
for buffer in self._buffers.values():
if (
buffer.address.upper() == address_upper
and buffer.char_uuid.lower() == char_uuid_lower
):
return buffer
return None
def list_active_subscriptions(self) -> list[dict[str, Any]]:
"""List all active notification subscriptions."""
return [
{
"address": buf.address,
"char_uuid": buf.char_uuid,
"char_path": buf.char_path,
"notifying": buf.notifying,
"buffer_count": len(buf.values),
"total_received": buf.total_received,
}
for buf in self._buffers.values()
]
def clear_buffer(self, char_path: str) -> None:
"""Clear the notification buffer for a characteristic."""
if char_path in self._buffers:
self._buffers[char_path].values.clear()
def remove_buffer(self, char_path: str) -> None:
"""Remove a buffer entirely (for cleanup)."""
if char_path in self._buffers:
del self._buffers[char_path]
class BlueZClient:
"""Async client for BlueZ D-Bus API."""
@ -393,7 +177,9 @@ class BlueZClient:
for path, interfaces in objects.items():
result[path] = {}
for iface, props in interfaces.items():
result[path][iface] = {k: unwrap_variant(v) for k, v in props.items()}
result[path][iface] = {
k: unwrap_variant(v) for k, v in props.items()
}
return result
async def _get_interface(self, path: str, interface: str) -> ProxyInterface:
@ -482,7 +268,9 @@ class BlueZClient:
"""Set adapter discoverable state."""
path = f"/org/bluez/{adapter}"
if timeout > 0:
await self._set_property(path, BLUEZ_ADAPTER_IFACE, "DiscoverableTimeout", timeout)
await self._set_property(
path, BLUEZ_ADAPTER_IFACE, "DiscoverableTimeout", timeout
)
await self._set_property(path, BLUEZ_ADAPTER_IFACE, "Discoverable", discoverable)
async def set_adapter_pairable(self, adapter: str, pairable: bool, timeout: int = 0) -> None:
@ -734,10 +522,7 @@ class BlueZClient:
if service_uuid:
for path, interfaces in objects.items():
if BLUEZ_GATT_SERVICE_IFACE in interfaces:
if (
interfaces[BLUEZ_GATT_SERVICE_IFACE].get("UUID", "").lower()
== service_uuid.lower()
):
if interfaces[BLUEZ_GATT_SERVICE_IFACE].get("UUID", "").lower() == service_uuid.lower():
service_paths.add(path)
characteristics = []
@ -810,7 +595,6 @@ class BlueZClient:
# Global client instance
_client: BlueZClient | None = None
_notify_manager: BLENotifyManager | None = None
async def get_client() -> BlueZClient:
@ -820,12 +604,3 @@ async def get_client() -> BlueZClient:
_client = BlueZClient()
await _client.connect()
return _client
async def get_notify_manager() -> BLENotifyManager:
"""Get or create the global BLE notify manager."""
global _notify_manager
if _notify_manager is None:
client = await get_client()
_notify_manager = BLENotifyManager(client)
return _notify_manager

756
src/mcbluetooth/gatt_server.py
View File

@ -1,756 +0,0 @@
"""BLE GATT Server implementation for BlueZ.
Registers as a BLE peripheral via BlueZ's GattManager1 D-Bus API, allowing
mcbluetooth to act as a GATT server. Remote BLE central devices can discover
services, read/write characteristics, and subscribe to notifications.
D-Bus hierarchy exported to BlueZ:
/mcbluetooth/gatt ObjectManager
/mcbluetooth/gatt/serviceN GattService1
/mcbluetooth/gatt/serviceN/charM GattCharacteristic1
/mcbluetooth/gatt/serviceN/charM/descK GattDescriptor1
/mcbluetooth/gatt/adv0 LEAdvertisement1
Reference patterns: agent.py (ServiceInterface, pending requests),
hfp_ag.py (dedicated bus, module singleton, ProfileManager registration).
"""
import asyncio
import logging
from collections import deque
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from dbus_fast import BusType, Message, MessageType, Variant
from dbus_fast.aio import MessageBus
from dbus_fast.service import PropertyAccess, ServiceInterface, dbus_property, method
log = logging.getLogger(__name__)
BLUEZ_SERVICE = "org.bluez"
GATT_MANAGER_IFACE = "org.bluez.GattManager1"
LE_ADV_MANAGER_IFACE = "org.bluez.LEAdvertisingManager1"
APP_BASE_PATH = "/mcbluetooth/gatt"
ADV_PATH = "/mcbluetooth/gatt/adv0"
# ==================== Data Classes ====================
@dataclass
class WriteEvent:
"""A write received from a remote BLE central device."""
index: int
timestamp: str
char_id: str
char_uuid: str
value: bytes
value_hex: str
value_string: str | None
device: str | None
def to_dict(self) -> dict[str, Any]:
d: dict[str, Any] = {
"index": self.index,
"timestamp": self.timestamp,
"char_id": self.char_id,
"char_uuid": self.char_uuid,
"value_hex": self.value_hex,
"length": len(self.value),
}
if self.value_string is not None:
d["value_string"] = self.value_string
if self.device:
d["device"] = self.device
return d
@dataclass
class ServerCharacteristic:
"""Internal state for a server-side GATT characteristic."""
char_id: str
path: str
uuid: str
flags: list[str]
value: bytes = b""
notifying: bool = False
service_id: str = ""
desc_count: int = 0
dbus_obj: Any = field(default=None, repr=False)
@dataclass
class ServerDescriptor:
"""Internal state for a server-side GATT descriptor."""
desc_id: str
path: str
uuid: str
flags: list[str]
value: bytes = b""
char_id: str = ""
dbus_obj: Any = field(default=None, repr=False)
@dataclass
class ServerService:
"""Internal state for a server-side GATT service."""
service_id: str
path: str
uuid: str
primary: bool = True
char_count: int = 0
characteristics: dict[str, ServerCharacteristic] = field(default_factory=dict)
dbus_obj: Any = field(default=None, repr=False)
# ==================== D-Bus Interfaces ====================
class GattApplication(ServiceInterface):
"""ObjectManager for the GATT application root.
BlueZ calls GetManagedObjects() on this path to discover all
services, characteristics, and descriptors we're hosting.
"""
def __init__(self, manager: "GattServerManager"):
super().__init__("org.freedesktop.DBus.ObjectManager")
self._manager = manager
@method()
def GetManagedObjects(self) -> "a{oa{sa{sv}}}": # noqa: F821
return self._manager._build_managed_objects()
class GattServiceIface(ServiceInterface):
"""org.bluez.GattService1 — exported at each service path.
Properties must be exposed via @dbus_property so BlueZ can validate
the service object through the standard Properties interface.
"""
def __init__(self, uuid: str, primary: bool):
super().__init__("org.bluez.GattService1")
self._uuid = uuid
self._primary = primary
@dbus_property(PropertyAccess.READ)
def UUID(self) -> "s": # noqa: F821
return self._uuid
@dbus_property(PropertyAccess.READ)
def Primary(self) -> "b": # noqa: F821
return self._primary
class GattCharacteristicIface(ServiceInterface):
"""org.bluez.GattCharacteristic1 — handles reads/writes from remotes.
ReadValue returns the stored value. WriteValue stores the value AND
records a WriteEvent for LLM polling (like agent.py's pending_requests).
Properties exposed via @dbus_property for BlueZ introspection.
"""
def __init__(self, char: ServerCharacteristic, manager: "GattServerManager"):
super().__init__("org.bluez.GattCharacteristic1")
self._char = char
self._manager = manager
@dbus_property(PropertyAccess.READ)
def UUID(self) -> "s": # noqa: F821
return self._char.uuid
@dbus_property(PropertyAccess.READ)
def Service(self) -> "o": # noqa: F821
return f"{APP_BASE_PATH}/{self._char.service_id}"
@dbus_property(PropertyAccess.READ)
def Flags(self) -> "as": # noqa: F821
return self._char.flags
@method()
def ReadValue(self, options: "a{sv}") -> "ay": # noqa: F821
offset = 0
opt_offset = options.get("offset")
if opt_offset is not None:
offset = opt_offset.value if hasattr(opt_offset, "value") else int(opt_offset)
return list(self._char.value[offset:])
@method()
def WriteValue(self, value: "ay", options: "a{sv}") -> None: # noqa: F821
new_bytes = bytes(value)
offset = 0
opt_offset = options.get("offset")
if opt_offset is not None:
offset = opt_offset.value if hasattr(opt_offset, "value") else int(opt_offset)
if offset > 0:
self._char.value = self._char.value[:offset] + new_bytes
else:
self._char.value = new_bytes
device = None
opt_device = options.get("device")
if opt_device is not None:
device = opt_device.value if hasattr(opt_device, "value") else str(opt_device)
self._manager._record_write(self._char.char_id, self._char.uuid, self._char.value, device)
@method()
def StartNotify(self) -> None:
self._char.notifying = True
log.info("GATT server: StartNotify on %s (%s)", self._char.char_id, self._char.uuid)
@method()
def StopNotify(self) -> None:
self._char.notifying = False
log.info("GATT server: StopNotify on %s (%s)", self._char.char_id, self._char.uuid)
class GattDescriptorIface(ServiceInterface):
"""org.bluez.GattDescriptor1 — handles reads/writes on descriptors.
Properties exposed via @dbus_property for BlueZ introspection.
"""
def __init__(self, desc: ServerDescriptor):
super().__init__("org.bluez.GattDescriptor1")
self._desc = desc
@dbus_property(PropertyAccess.READ)
def UUID(self) -> "s": # noqa: F821
return self._desc.uuid
@dbus_property(PropertyAccess.READ)
def Characteristic(self) -> "o": # noqa: F821
return f"{APP_BASE_PATH}/{self._desc.char_id}"
@dbus_property(PropertyAccess.READ)
def Flags(self) -> "as": # noqa: F821
return self._desc.flags
@method()
def ReadValue(self, options: "a{sv}") -> "ay": # noqa: F821
offset = 0
opt_offset = options.get("offset")
if opt_offset is not None:
offset = opt_offset.value if hasattr(opt_offset, "value") else int(opt_offset)
return list(self._desc.value[offset:])
@method()
def WriteValue(self, value: "ay", options: "a{sv}") -> None: # noqa: F821
self._desc.value = bytes(value)
class LEAdvertisementIface(ServiceInterface):
"""org.bluez.LEAdvertisement1 — BLE advertising data.
BlueZ reads advertisement properties via Properties.GetAll, so these
must be exposed as @dbus_property (unlike GATT objects which use
GetManagedObjects).
"""
def __init__(self, adv_type: str, local_name: str, service_uuids: list[str]):
super().__init__("org.bluez.LEAdvertisement1")
self._type = adv_type
self._local_name = local_name
self._service_uuids = service_uuids
@method()
def Release(self) -> None:
log.info("GATT server: advertisement released by BlueZ")
@dbus_property(PropertyAccess.READ)
def Type(self) -> "s": # noqa: F821
return self._type
@dbus_property(PropertyAccess.READ)
def LocalName(self) -> "s": # noqa: F821
return self._local_name
@dbus_property(PropertyAccess.READ)
def ServiceUUIDs(self) -> "as": # noqa: F821
return self._service_uuids
# ==================== GATT Server Manager ====================
class GattServerManager:
"""Manages the GATT server lifecycle and state.
Coordinates D-Bus object creation, BlueZ registration,
advertising, and write event buffering.
"""
def __init__(self):
self._bus: MessageBus | None = None
self._app: GattApplication | None = None
self._adv: LEAdvertisementIface | None = None
self._services: dict[str, ServerService] = {}
self._characteristics: dict[str, ServerCharacteristic] = {}
self._descriptors: dict[str, ServerDescriptor] = {}
self._write_events: deque[WriteEvent] = deque(maxlen=200)
self._write_index: int = 0
self._service_count: int = 0
self._registered: bool = False
self._advertising: bool = False
self._write_callback: Any = None # async callable(char_id, value, device)
# ---- Service construction ----
def add_service(self, uuid: str, primary: bool = True) -> str:
"""Add a GATT service. Returns the service_id (e.g. 'service0')."""
service_id = f"service{self._service_count}"
self._service_count += 1
path = f"{APP_BASE_PATH}/{service_id}"
svc = ServerService(
service_id=service_id,
path=path,
uuid=uuid,
primary=primary,
)
svc.dbus_obj = GattServiceIface(uuid, primary)
self._services[service_id] = svc
log.info("GATT server: added service %s (UUID=%s)", service_id, uuid)
return service_id
def add_characteristic(
self,
service_id: str,
uuid: str,
flags: list[str],
value: bytes = b"",
) -> str:
"""Add a characteristic to a service. Returns the char_id."""
svc = self._services.get(service_id)
if not svc:
raise ValueError(f"Service '{service_id}' not found")
char_idx = svc.char_count
svc.char_count += 1
char_id = f"{service_id}/char{char_idx}"
path = f"{APP_BASE_PATH}/{char_id}"
char = ServerCharacteristic(
char_id=char_id,
path=path,
uuid=uuid,
flags=flags,
value=value,
service_id=service_id,
)
char.dbus_obj = GattCharacteristicIface(char, self)
svc.characteristics[char_id] = char
self._characteristics[char_id] = char
log.info(
"GATT server: added characteristic %s (UUID=%s, flags=%s)",
char_id,
uuid,
flags,
)
return char_id
def add_descriptor(
self,
char_id: str,
uuid: str,
flags: list[str],
value: bytes = b"",
) -> str:
"""Add a descriptor to a characteristic. Returns the desc_id."""
char = self._characteristics.get(char_id)
if not char:
raise ValueError(f"Characteristic '{char_id}' not found")
desc_idx = char.desc_count
char.desc_count += 1
desc_id = f"{char_id}/desc{desc_idx}"
path = f"{APP_BASE_PATH}/{desc_id}"
desc = ServerDescriptor(
desc_id=desc_id,
path=path,
uuid=uuid,
flags=flags,
value=value,
char_id=char_id,
)
desc.dbus_obj = GattDescriptorIface(desc)
self._descriptors[desc_id] = desc
log.info("GATT server: added descriptor %s (UUID=%s)", desc_id, uuid)
return desc_id
def clear(self) -> None:
"""Remove all services, characteristics, and descriptors."""
if self._registered:
raise RuntimeError("Cannot clear while registered — unregister first")
self._services.clear()
self._characteristics.clear()
self._descriptors.clear()
self._service_count = 0
# ---- Registration with BlueZ ----
async def register(self, adapter: str) -> None:
"""Register the GATT application with BlueZ GattManager1."""
if self._registered:
return
if not self._services:
raise ValueError("No services defined — add at least one service first")
# Dedicated bus connection (like hfp_ag.py)
if not self._bus:
self._bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
# Export ObjectManager at application root
self._app = GattApplication(self)
self._bus.export(APP_BASE_PATH, self._app)
# Export all service/characteristic/descriptor D-Bus objects
for svc in self._services.values():
self._bus.export(svc.path, svc.dbus_obj)
for char in svc.characteristics.values():
self._bus.export(char.path, char.dbus_obj)
for desc in self._descriptors.values():
self._bus.export(desc.path, desc.dbus_obj)
# Register with BlueZ
adapter_path = f"/org/bluez/{adapter}"
introspection = await self._bus.introspect(BLUEZ_SERVICE, adapter_path)
proxy = self._bus.get_proxy_object(BLUEZ_SERVICE, adapter_path, introspection)
gatt_mgr = proxy.get_interface(GATT_MANAGER_IFACE)
try:
await gatt_mgr.call_register_application(APP_BASE_PATH, {})
self._registered = True
log.info("GATT server: registered with BlueZ (%s)", adapter)
except Exception as e:
if "Already Exists" in str(e):
log.info("GATT server: stale registration — re-registering")
try:
await gatt_mgr.call_unregister_application(APP_BASE_PATH)
except Exception:
pass
await gatt_mgr.call_register_application(APP_BASE_PATH, {})
self._registered = True
log.info("GATT server: re-registered with BlueZ (%s)", adapter)
else:
raise
async def unregister(self, adapter: str) -> None:
"""Unregister the GATT application from BlueZ."""
if self._advertising:
await self.set_advertising(adapter, False)
if self._bus and self._registered:
try:
adapter_path = f"/org/bluez/{adapter}"
introspection = await self._bus.introspect(BLUEZ_SERVICE, adapter_path)
proxy = self._bus.get_proxy_object(BLUEZ_SERVICE, adapter_path, introspection)
gatt_mgr = proxy.get_interface(GATT_MANAGER_IFACE)
await gatt_mgr.call_unregister_application(APP_BASE_PATH)
except Exception:
pass
self._registered = False
# Disconnect bus (cleans up all exports)
if self._bus:
self._bus.disconnect()
self._bus = None
self._app = None
self._adv = None
log.info("GATT server: unregistered")
# ---- Advertising ----
async def set_advertising(
self,
adapter: str,
enable: bool,
name: str = "mcbluetooth",
service_uuids: list[str] | None = None,
) -> None:
"""Start or stop BLE advertising."""
if not self._bus:
raise RuntimeError("Bus not connected — register first")
adapter_path = f"/org/bluez/{adapter}"
introspection = await self._bus.introspect(BLUEZ_SERVICE, adapter_path)
proxy = self._bus.get_proxy_object(BLUEZ_SERVICE, adapter_path, introspection)
adv_mgr = proxy.get_interface(LE_ADV_MANAGER_IFACE)
if enable:
if self._advertising:
return
uuids = service_uuids or [svc.uuid for svc in self._services.values()]
self._adv = LEAdvertisementIface("peripheral", name, uuids)
self._bus.export(ADV_PATH, self._adv)
try:
await adv_mgr.call_register_advertisement(ADV_PATH, {})
self._advertising = True
log.info("GATT server: advertising started (name=%s)", name)
except Exception as e:
if "Already Exists" in str(e):
try:
await adv_mgr.call_unregister_advertisement(ADV_PATH)
except Exception:
pass
await adv_mgr.call_register_advertisement(ADV_PATH, {})
self._advertising = True
log.info("GATT server: advertising restarted (name=%s)", name)
else:
raise
else:
if not self._advertising:
return
try:
await adv_mgr.call_unregister_advertisement(ADV_PATH)
except Exception:
pass
self._advertising = False
self._adv = None
log.info("GATT server: advertising stopped")
# ---- Value management ----
def set_value(self, char_id: str, value: bytes) -> None:
"""Set characteristic value and auto-notify if subscribed."""
char = self._characteristics.get(char_id)
if not char:
raise ValueError(f"Characteristic '{char_id}' not found")
char.value = value
if char.notifying:
self._emit_notification(char)
def get_value(self, char_id: str) -> bytes:
"""Get current characteristic value."""
char = self._characteristics.get(char_id)
if not char:
raise ValueError(f"Characteristic '{char_id}' not found")
return char.value
def notify(self, char_id: str) -> bool:
"""Explicitly send notification for current value."""
char = self._characteristics.get(char_id)
if not char:
raise ValueError(f"Characteristic '{char_id}' not found")
if not char.notifying:
return False
self._emit_notification(char)
return True
def _emit_notification(self, char: ServerCharacteristic) -> None:
"""Emit PropertiesChanged signal to trigger BLE notification."""
if not self._bus:
return
msg = Message(
message_type=MessageType.SIGNAL,
path=char.path,
interface="org.freedesktop.DBus.Properties",
member="PropertiesChanged",
signature="sa{sv}as",
body=[
"org.bluez.GattCharacteristic1",
{"Value": Variant("ay", list(char.value))},
[],
],
)
self._bus.send_message(msg)
log.debug("GATT server: notification sent on %s (%d bytes)", char.char_id, len(char.value))
# ---- Write event monitoring ----
def _record_write(self, char_id: str, char_uuid: str, value: bytes, device: str | None) -> None:
"""Record a write event and fire callback if registered."""
event = WriteEvent(
index=self._write_index,
timestamp=datetime.now(UTC).isoformat(),
char_id=char_id,
char_uuid=char_uuid,
value=value,
value_hex=value.hex(),
value_string=_try_decode(value),
device=device,
)
self._write_events.append(event)
self._write_index += 1
log.debug(
"GATT server: write on %s from %s (%d bytes)",
char_id,
device or "unknown",
len(value),
)
# Fire async callback (e.g. ELM327 emulator auto-responder)
if self._write_callback:
try:
loop = asyncio.get_running_loop()
loop.create_task(self._write_callback(char_id, value, device))
except RuntimeError:
pass
def get_write_events(
self,
since_index: int = 0,
char_id: str | None = None,
limit: int = 50,
) -> list[dict[str, Any]]:
"""Get write events, optionally filtered."""
events = [
e
for e in self._write_events
if e.index >= since_index and (char_id is None or e.char_id == char_id)
]
return [e.to_dict() for e in events[:limit]]
def clear_write_events(self) -> int:
"""Clear all write events. Returns count cleared."""
count = len(self._write_events)
self._write_events.clear()
return count
# ---- GetManagedObjects builder ----
def _build_managed_objects(self) -> dict[str, dict[str, dict[str, Variant]]]:
"""Build the response for ObjectManager.GetManagedObjects().
Returns the complete D-Bus object hierarchy that BlueZ uses
to discover our GATT services, characteristics, and descriptors.
"""
objects: dict[str, dict[str, dict[str, Variant]]] = {}
for svc in self._services.values():
char_paths = [c.path for c in svc.characteristics.values()]
objects[svc.path] = {
"org.bluez.GattService1": {
"UUID": Variant("s", svc.uuid),
"Primary": Variant("b", svc.primary),
"Characteristics": Variant("ao", char_paths),
}
}
for char in svc.characteristics.values():
desc_paths = [
d.path for d in self._descriptors.values() if d.char_id == char.char_id
]
objects[char.path] = {
"org.bluez.GattCharacteristic1": {
"UUID": Variant("s", char.uuid),
"Service": Variant("o", svc.path),
"Flags": Variant("as", char.flags),
"Descriptors": Variant("ao", desc_paths),
}
}
for desc in self._descriptors.values():
char = self._characteristics.get(desc.char_id)
if char:
objects[desc.path] = {
"org.bluez.GattDescriptor1": {
"UUID": Variant("s", desc.uuid),
"Characteristic": Variant("o", char.path),
"Flags": Variant("as", desc.flags),
}
}
return objects
# ---- Status ----
def get_status(self) -> dict[str, Any]:
"""Get overall GATT server status."""
services = []
for svc in self._services.values():
chars = []
for char in svc.characteristics.values():
descs = [
{
"desc_id": d.desc_id,
"uuid": d.uuid,
"flags": d.flags,
}
for d in self._descriptors.values()
if d.char_id == char.char_id
]
chars.append(
{
"char_id": char.char_id,
"uuid": char.uuid,
"flags": char.flags,
"value_length": len(char.value),
"notifying": char.notifying,
"descriptors": descs,
}
)
services.append(
{
"service_id": svc.service_id,
"uuid": svc.uuid,
"primary": svc.primary,
"characteristics": chars,
}
)
return {
"registered": self._registered,
"advertising": self._advertising,
"services": services,
"write_event_count": len(self._write_events),
"write_event_total": self._write_index,
}
# ==================== Module-level lifecycle ====================
_manager: GattServerManager | None = None
def get_gatt_server() -> GattServerManager:
"""Get or create the global GATT server manager (singleton)."""
global _manager
if _manager is None:
_manager = GattServerManager()
return _manager
async def shutdown_gatt_server(adapter: str) -> None:
"""Shut down the GATT server completely."""
global _manager
if _manager:
try:
await _manager.unregister(adapter)
except Exception:
pass
_manager = None
# ==================== Helpers ====================
def _try_decode(value: bytes) -> str | None:
"""Try to decode bytes as printable UTF-8."""
try:
decoded = value.decode("utf-8")
if all(c.isprintable() or c in "\r\n\t" for c in decoded):
return decoded
return None
except (UnicodeDecodeError, ValueError):
return None

210
src/mcbluetooth/resources.py
View File

@ -3,7 +3,6 @@
Resources provide a live, queryable view of Bluetooth state:
- Adapters: Available Bluetooth controllers
- Devices: Filtered by state (visible, paired, connected)
- BLE Notifications: Real-time GATT notification values
Unlike tools which perform actions, resources are read-only snapshots
that clients can poll or subscribe to for state changes.
@ -16,8 +15,7 @@ from dataclasses import asdict
from fastmcp import FastMCP
from mcbluetooth.dbus_client import get_client, get_notify_manager
from mcbluetooth.gatt_server import get_gatt_server
from mcbluetooth.dbus_client import get_client
def register_resources(mcp: FastMCP) -> None:
@ -198,209 +196,3 @@ def register_resources(mcp: FastMCP) -> None:
# DeviceInfo already has manufacturer_data/service_data as hex strings
return json.dumps(asdict(d), indent=2)
return json.dumps({"error": f"Device '{address}' not found"})
# ==================== BLE Notification Resources ====================
def _format_uuid_short(uuid: str) -> str:
"""Format UUID for display - short form for standard UUIDs."""
if uuid.lower().endswith("-0000-1000-8000-00805f9b34fb"):
short = uuid[:8].lstrip("0") or "0"
return f"0x{short.upper()}"
return uuid
@mcp.resource(
"bluetooth://ble/{address}/{char_uuid}/notifications",
name="BLE Notifications",
description=(
"Current state of a BLE GATT characteristic notification subscription. "
"Returns the latest notification value and buffer statistics. "
"Subscribe to this resource to receive updates when new notifications arrive."
),
mime_type="application/json",
)
async def resource_ble_notifications(address: str, char_uuid: str) -> str:
"""Get current notification state for a characteristic."""
notify_manager = await get_notify_manager()
buffer = notify_manager.get_buffer_by_address_uuid(address, char_uuid)
if not buffer:
return json.dumps(
{
"error": f"No active notification subscription for {address}/{char_uuid}",
"hint": "Use bt_ble_notify to enable notifications first",
}
)
result = {
"address": buffer.address,
"characteristic_uuid": buffer.char_uuid,
"characteristic_uuid_short": _format_uuid_short(buffer.char_uuid),
"notifying": buffer.notifying,
"latest": buffer.latest.to_dict() if buffer.latest else None,
"buffer_count": len(buffer.values),
"total_received": buffer.total_received,
}
return json.dumps(result, indent=2)
@mcp.resource(
"bluetooth://ble/{address}/{char_uuid}/notifications/history",
name="BLE Notification History",
description=(
"Buffered history of BLE GATT notification values. "
"Returns the most recent notification values with timestamps. "
"Default returns last 10 values; use count parameter for more."
),
mime_type="application/json",
)
async def resource_ble_notification_history(
address: str, char_uuid: str, count: int = 10
) -> str:
"""Get notification history for a characteristic."""
notify_manager = await get_notify_manager()
buffer = notify_manager.get_buffer_by_address_uuid(address, char_uuid)
if not buffer:
return json.dumps(
{
"error": f"No active notification subscription for {address}/{char_uuid}",
"hint": "Use bt_ble_notify to enable notifications first",
}
)
# Ensure count is reasonable
count = min(max(1, count), buffer.max_size)
history = buffer.get_history(count)
result = {
"address": buffer.address,
"characteristic_uuid": buffer.char_uuid,
"count": len(history),
"total_available": len(buffer.values),
"total_received": buffer.total_received,
"values": [v.to_dict() for v in history],
}
return json.dumps(result, indent=2)
@mcp.resource(
"bluetooth://ble/notifications",
name="Active BLE Subscriptions",
description=(
"List all active BLE notification subscriptions with their buffer statistics. "
"Useful for monitoring which characteristics are being tracked."
),
mime_type="application/json",
)
async def resource_ble_all_notifications() -> str:
"""List all active notification subscriptions."""
notify_manager = await get_notify_manager()
subscriptions = notify_manager.list_active_subscriptions()
# Enhance with short UUID format
for sub in subscriptions:
sub["uuid_short"] = _format_uuid_short(sub["char_uuid"])
return json.dumps(
{
"count": len(subscriptions),
"subscriptions": subscriptions,
},
indent=2,
)
# ==================== SPP Resources ====================
@mcp.resource(
"bluetooth://spp/connections",
name="SPP Connections",
description=(
"Active SPP (Serial Port Profile) connections with role, "
"duration, and byte counters for each peer."
),
mime_type="application/json",
)
async def resource_spp_connections() -> str:
"""Get active SPP connections."""
from mcbluetooth.spp import get_spp
profile = await get_spp()
if not profile:
return json.dumps({"registered": False, "connections": []})
status = profile.get_status()
return json.dumps(
{
"registered": status["registered"],
"uuid": status["uuid"],
"connections": status["connections"],
},
indent=2,
)
@mcp.resource(
"bluetooth://spp/data",
name="SPP Received Data",
description=(
"Recent data received from SPP peers. Returns the last 50 "
"data events with timestamps, addresses, and hex/string values."
),
mime_type="application/json",
)
async def resource_spp_data() -> str:
"""Get recent SPP received data events."""
from mcbluetooth.spp import get_spp
profile = await get_spp()
if not profile:
return json.dumps({"count": 0, "events": [], "hint": "SPP not enabled"})
events = profile.get_recv_events(since_index=0, limit=50)
return json.dumps(
{
"count": len(events),
"total": profile._recv_index,
"events": events,
},
indent=2,
)
# ==================== GATT Server Resources ====================
@mcp.resource(
"bluetooth://gatt/server",
name="GATT Server State",
description=(
"Current state of the BLE GATT server (peripheral mode). "
"Shows registered services, characteristics, advertising status, "
"and write event statistics."
),
mime_type="application/json",
)
async def resource_gatt_server() -> str:
"""Get GATT server state."""
mgr = get_gatt_server()
return json.dumps(mgr.get_status(), indent=2)
@mcp.resource(
"bluetooth://gatt/server/writes",
name="GATT Server Write Events",
description=(
"Write events received from remote BLE central devices. "
"Shows the most recent writes to server characteristics "
"with timestamps, values, and source device info."
),
mime_type="application/json",
)
async def resource_gatt_server_writes() -> str:
"""Get recent write events from remote clients."""
mgr = get_gatt_server()
events = mgr.get_write_events(since_index=0, limit=50)
return json.dumps(
{
"count": len(events),
"total": mgr._write_index,
"events": events,
},
indent=2,
)

58
src/mcbluetooth/server.py
View File

@ -3,18 +3,7 @@
from fastmcp import FastMCP
from mcbluetooth import resources
from mcbluetooth.tools import (
adapter,
audio,
ble,
bt_elm327_emu,
device,
gatt_server,
hfp,
monitor,
obex,
spp,
)
from mcbluetooth.tools import adapter, audio, ble, device, hfp, monitor, obex
mcp = FastMCP(
name="mcbluetooth",
@ -36,52 +25,10 @@ This server provides comprehensive control over the Linux Bluetooth stack:
- bluetooth://adapter/{name} - Specific adapter details
- bluetooth://device/{address} - Specific device details
### BLE Notification Resources
- bluetooth://ble/notifications - List all active notification subscriptions
- bluetooth://ble/{address}/{char_uuid}/notifications - Current notification state and latest value
- bluetooth://ble/{address}/{char_uuid}/notifications/history - Buffered notification history
To capture BLE notifications:
1. Connect to the device: bt_connect(adapter, address)
2. Enable notifications: bt_ble_notify(adapter, address, char_uuid, enable=True)
3. Read notifications: Use the bluetooth://ble/{address}/{uuid}/notifications resource
4. Subscribe to the resource for real-time updates (client-side)
### GATT Server Resources (peripheral mode)
- bluetooth://gatt/server - GATT server state (registered, services, advertising)
- bluetooth://gatt/server/writes - Write events from remote clients
To act as a BLE peripheral (GATT server):
1. Add services: bt_gatt_server_add_service(uuid)
2. Add characteristics: bt_gatt_server_add_characteristic(service_id, uuid, flags)
3. Register: bt_gatt_server_register(adapter)
4. Advertise: bt_gatt_server_advertise(adapter, enable=True, name="MyDevice")
5. Monitor writes: bt_gatt_server_read_writes(since_index=0)
6. Respond: bt_gatt_server_set_value(char_id, value) — auto-notifies subscribers
### ELM327 OBD-II Emulator
For quick OBD-II BLE adapter emulation (Nordic UART Service):
1. bt_elm327_emu_start(adapter, name="OBDII") — sets up NUS, advertises
2. bt_elm327_emu_set_pid(0x0C, "0BB8", "hex") — set RPM to 750
3. bt_elm327_emu_read_commands() — see what the OBD app sent
4. bt_elm327_emu_stop(adapter) — clean up
## Tools
All tools require an explicit 'adapter' parameter (e.g., "hci0").
Use bt_list_adapters() to discover available adapters.
### SPP (Serial Port Profile) — Classic Bluetooth Serial
- bluetooth://spp/connections — Active SPP connections
- bluetooth://spp/data — Recent received data events
For classic Bluetooth serial (RFCOMM):
1. bt_spp_enable() — register profile
2. Server: make discoverable, remote connects automatically
3. Client: bt_spp_connect(adapter, address)
4. Send: bt_spp_send(address, "ATZ\\r\\n", "string")
5. Receive: bt_spp_recv(since_index=0) — cursor-based polling
6. Disconnect: bt_spp_disconnect(address)
For pairing, use pairing_mode parameter:
- "elicit": Use MCP elicitation to request PIN from user (preferred)
- "interactive": Return awaiting status, then call bt_pair_confirm
@ -98,11 +45,8 @@ device.register_tools(mcp)
audio.register_tools(mcp)
hfp.register_tools(mcp)
ble.register_tools(mcp)
gatt_server.register_tools(mcp)
bt_elm327_emu.register_tools(mcp)
monitor.register_tools(mcp)
obex.register_tools(mcp)
spp.register_tools(mcp)
def main():

419
src/mcbluetooth/spp.py
View File

@ -1,419 +0,0 @@
"""SPP (Serial Port Profile) implementation for BlueZ.
Registers as an SPP endpoint via BlueZ ProfileManager1, providing raw
bidirectional serial byte streams over RFCOMM. When a remote device
connects, BlueZ hands us the RFCOMM file descriptor through the Profile1
D-Bus interface — the same pattern used by hfp_ag.py, minus the AT
command protocol layer.
Use cases: serial terminal to Arduino/ESP32, GPS receivers (NMEA),
legacy sensors, Bluetooth modems, any line-oriented protocol over
classic Bluetooth.
D-Bus flow:
1. RegisterProfile(SPP_UUID) with ProfileManager1
2. Remote connects → BlueZ calls NewConnection(device, fd, props)
3. We dup the fd, wrap in async streams, read/write raw bytes
4. RequestDisconnection or EOF → cleanup
"""
import asyncio
import logging
import os
import socket
import time
from collections import deque
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from dbus_fast import BusType, Variant
from dbus_fast.aio import MessageBus
from dbus_fast.service import ServiceInterface, method
log = logging.getLogger(__name__)
# Bluetooth socket constants — Python often compiled without bluetooth.h
_AF_BLUETOOTH = getattr(socket, "AF_BLUETOOTH", 31)
_BTPROTO_RFCOMM = getattr(socket, "BTPROTO_RFCOMM", 3)
# D-Bus constants
BLUEZ_SERVICE = "org.bluez"
PROFILE_MANAGER_IFACE = "org.bluez.ProfileManager1"
SPP_UUID = "00001101-0000-1000-8000-00805f9b34fb"
SPP_PROFILE_PATH = "/mcbluetooth/spp"
def _try_decode(value: bytes) -> str | None:
"""Try to decode bytes as printable UTF-8."""
try:
decoded = value.decode("utf-8")
if all(c.isprintable() or c in "\r\n\t" for c in decoded):
return decoded
return None
except (UnicodeDecodeError, ValueError):
return None
def _path_to_address(device_path: str) -> str:
parts = device_path.split("/")
if len(parts) >= 5 and parts[-1].startswith("dev_"):
return parts[-1][4:].replace("_", ":")
return device_path
@dataclass
class SPPDataEvent:
"""A chunk of data received from a remote SPP peer."""
index: int
timestamp: str
address: str
value: bytes
value_hex: str
value_string: str | None
def to_dict(self) -> dict[str, Any]:
d: dict[str, Any] = {
"index": self.index,
"timestamp": self.timestamp,
"address": self.address,
"value_hex": self.value_hex,
"length": len(self.value),
}
if self.value_string is not None:
d["value_string"] = self.value_string
return d
@dataclass
class SPPConnection:
"""Per-peer connection state for an SPP session."""
device_path: str
address: str
role: str # "server" (they connected to us) or "client" (we connected to them)
fd: int
connected_at: float = field(default_factory=time.monotonic)
sock: socket.socket | None = None
reader: asyncio.StreamReader | None = None
writer: asyncio.StreamWriter | None = None
bytes_sent: int = 0
bytes_received: int = 0
_read_task: asyncio.Task | None = field(default=None, repr=False)
class SPPProfile(ServiceInterface):
"""D-Bus Profile1 service for SPP (Serial Port Profile)."""
def __init__(self) -> None:
super().__init__("org.bluez.Profile1")
self.connections: dict[str, SPPConnection] = {} # keyed by device_path
self._recv_events: deque[SPPDataEvent] = deque(maxlen=500)
self._recv_index: int = 0
self._recv_callback: Any = None # async callable(address, data)
@method()
def Release(self) -> None:
log.info("SPP profile released")
for conn in list(self.connections.values()):
self._cleanup_connection(conn)
self.connections.clear()
@method()
def NewConnection(self, device: "o", fd: "h", properties: "a{sv}") -> None:
address = _path_to_address(device)
log.debug("SPP NewConnection: device=%s fd=%r props=%s", device, fd, properties)
if fd is None or (isinstance(fd, int) and fd < 0):
log.error("SPP: invalid fd received: %r", fd)
return
log.info("SPP: NewConnection from %s (fd=%d)", address, fd)
# Duplicate the fd so we own it independent of dbus-fast
try:
new_fd = os.dup(fd)
log.debug("os.dup(%d) -> %d", fd, new_fd)
except OSError:
log.exception("SPP: os.dup(%d) failed for %s", fd, address)
return
conn = SPPConnection(
device_path=device,
address=address,
role="server",
fd=new_fd,
)
self.connections[device] = conn
loop = asyncio.get_event_loop()
conn._read_task = loop.create_task(self._handle_connection(conn))
log.debug("SPP NewConnection done, read task created for %s", address)
@method()
def RequestDisconnection(self, device: "o") -> None:
address = _path_to_address(device)
log.info("SPP: disconnect requested for %s", address)
conn = self.connections.pop(device, None)
if conn:
self._cleanup_connection(conn)
def _cleanup_connection(self, conn: SPPConnection) -> None:
if conn._read_task and not conn._read_task.done():
conn._read_task.cancel()
if conn.writer:
try:
conn.writer.close()
except Exception:
pass
if conn.sock:
try:
conn.sock.close()
except Exception:
pass
elif conn.fd >= 0:
try:
os.close(conn.fd)
except Exception:
pass
async def _handle_connection(self, conn: SPPConnection) -> None:
"""Read loop for raw bytes from the remote SPP peer."""
try:
log.debug("SPP _handle_connection start: addr=%s fd=%d", conn.address, conn.fd)
# socket.fromfd() dups the fd internally — close our intermediate copy
conn.sock = socket.fromfd(
conn.fd, _AF_BLUETOOTH, socket.SOCK_STREAM,
_BTPROTO_RFCOMM,
)
log.debug("socket.fromfd OK: fileno=%d", conn.sock.fileno())
try:
os.close(conn.fd)
except OSError:
pass
conn.fd = -1 # transferred to socket
conn.sock.setblocking(False)
conn.reader, conn.writer = await asyncio.open_connection(sock=conn.sock)
log.debug("SPP async streams ready for %s, entering read loop", conn.address)
while True:
data = await conn.reader.read(4096)
if not data:
log.debug("SPP: EOF from %s (clean disconnect)", conn.address)
break
log.debug("SPP recv %d bytes from %s: %r", len(data), conn.address, data[:80])
conn.bytes_received += len(data)
event = SPPDataEvent(
index=self._recv_index,
timestamp=datetime.now(UTC).isoformat(),
address=conn.address,
value=data,
value_hex=data.hex(),
value_string=_try_decode(data),
)
self._recv_events.append(event)
self._recv_index += 1
# Fire async callback if registered (auto-responders)
if self._recv_callback:
try:
loop = asyncio.get_running_loop()
loop.create_task(self._recv_callback(conn.address, data))
except RuntimeError:
pass
except (ConnectionResetError, BrokenPipeError, OSError) as e:
log.debug("SPP connection error for %s: %s: %s", conn.address, type(e).__name__, e)
except asyncio.CancelledError:
log.debug("SPP task cancelled for %s", conn.address)
except Exception:
log.exception("SPP UNEXPECTED error for %s", conn.address)
finally:
log.debug("SPP cleanup for %s", conn.address)
self.connections.pop(conn.device_path, None)
self._cleanup_connection(conn)
async def send(self, address: str, data: bytes) -> bool:
"""Send raw bytes to a connected SPP peer."""
conn = self._get_connection(address)
if not conn or not conn.writer or conn.writer.is_closing():
return False
conn.writer.write(data)
await conn.writer.drain()
conn.bytes_sent += len(data)
log.debug("SPP sent %d bytes to %s", len(data), address)
return True
def add_client_connection(self, device_path: str, address: str, fd: int) -> None:
"""Register a client-initiated connection (we connected to them).
Called after Device1.ConnectProfile triggers NewConnection on our
Profile1 handler. In client mode, BlueZ still delivers the fd via
NewConnection — we just tag the role as "client" before the read
loop starts. However, BlueZ may have already fired NewConnection
by the time ConnectProfile returns, so this is a fallback to
retag if needed.
"""
conn = self.connections.get(device_path)
if conn:
conn.role = "client"
def _get_connection(self, address: str) -> SPPConnection | None:
for conn in self.connections.values():
if conn.address.upper() == address.upper():
return conn
return None
def get_recv_events(
self,
since_index: int = 0,
address: str | None = None,
limit: int = 50,
) -> list[dict[str, Any]]:
"""Get received data events, optionally filtered."""
events = [
e
for e in self._recv_events
if e.index >= since_index
and (address is None or e.address.upper() == address.upper())
]
return [e.to_dict() for e in events[:limit]]
def clear_recv_events(self) -> int:
"""Clear all received data events. Returns count cleared."""
count = len(self._recv_events)
self._recv_events.clear()
return count
def get_status(self) -> dict[str, Any]:
"""Get overall SPP status."""
conns = []
now = time.monotonic()
for conn in self.connections.values():
conns.append({
"address": conn.address,
"role": conn.role,
"duration_seconds": round(now - conn.connected_at, 1),
"bytes_sent": conn.bytes_sent,
"bytes_received": conn.bytes_received,
})
return {
"registered": _profile_registered,
"uuid": _registered_uuid or SPP_UUID,
"connections": conns,
"recv_buffer_count": len(self._recv_events),
"recv_buffer_total": self._recv_index,
}
# ==================== Module-level lifecycle ====================
_profile: SPPProfile | None = None
_profile_bus: MessageBus | None = None
_profile_registered: bool = False
_registered_uuid: str | None = None
async def enable_spp(
uuid: str = SPP_UUID,
channel: int = 0,
name: str = "mcbluetooth SPP",
) -> SPPProfile:
"""Register the SPP profile with BlueZ.
Args:
uuid: Service UUID. Default is standard SPP UUID. Use custom UUIDs
for Arduino/ESP32 devices that advertise non-standard RFCOMM.
channel: RFCOMM channel number (0 = auto-assign).
name: Profile display name.
"""
global _profile, _profile_bus, _profile_registered, _registered_uuid
if _profile_registered and _profile:
return _profile
if _profile is None:
_profile = SPPProfile()
if _profile_bus is None:
_profile_bus = await MessageBus(
bus_type=BusType.SYSTEM,
negotiate_unix_fd=True, # Required: BlueZ passes RFCOMM fd via D-Bus
).connect()
log.debug("SPP D-Bus connected: negotiate_unix_fd=%s unique_name=%s",
_profile_bus._negotiate_unix_fd, _profile_bus.unique_name)
_profile_bus.export(SPP_PROFILE_PATH, _profile)
# Register with ProfileManager1
introspection = await _profile_bus.introspect(BLUEZ_SERVICE, "/org/bluez")
proxy = _profile_bus.get_proxy_object(BLUEZ_SERVICE, "/org/bluez", introspection)
profile_mgr = proxy.get_interface(PROFILE_MANAGER_IFACE)
options: dict[str, Variant] = {
"Name": Variant("s", name),
"Role": Variant("s", "client-server"),
}
if channel > 0:
options["Channel"] = Variant("q", channel)
try:
await profile_mgr.call_register_profile(
SPP_PROFILE_PATH,
uuid,
options,
)
_profile_registered = True
_registered_uuid = uuid
log.info("SPP profile registered with BlueZ (uuid=%s, channel=%s)", uuid, channel or "auto")
except Exception as e:
if "Already Exists" in str(e):
log.info("SPP profile stale — unregistering and re-registering")
try:
await profile_mgr.call_unregister_profile(SPP_PROFILE_PATH)
except Exception:
pass
await profile_mgr.call_register_profile(
SPP_PROFILE_PATH,
uuid,
options,
)
_profile_registered = True
_registered_uuid = uuid
log.info("SPP profile re-registered with BlueZ")
else:
raise
return _profile
async def disable_spp() -> None:
"""Unregister the SPP profile and close all connections."""
global _profile, _profile_bus, _profile_registered, _registered_uuid
if _profile_bus and _profile_registered:
try:
introspection = await _profile_bus.introspect(BLUEZ_SERVICE, "/org/bluez")
proxy = _profile_bus.get_proxy_object(BLUEZ_SERVICE, "/org/bluez", introspection)
profile_mgr = proxy.get_interface(PROFILE_MANAGER_IFACE)
await profile_mgr.call_unregister_profile(SPP_PROFILE_PATH)
except Exception:
pass
_profile_registered = False
_registered_uuid = None
if _profile:
_profile.Release()
_profile = None
if _profile_bus:
_profile_bus.disconnect()
_profile_bus = None
async def get_spp() -> SPPProfile | None:
"""Get the current SPP profile instance (None if not enabled)."""
return _profile

154
src/mcbluetooth/tools/ble.py
View File

@ -7,8 +7,7 @@ from typing import Any
from fastmcp import FastMCP
from mcbluetooth.ble_subscriptions import make_notification_uri
from mcbluetooth.dbus_client import get_client, get_notify_manager
from mcbluetooth.dbus_client import get_client
# Common BLE service UUIDs
BATTERY_SERVICE_UUID = "0000180f-0000-1000-8000-00805f9b34fb"
@ -58,7 +57,9 @@ def register_tools(mcp: FastMCP) -> None:
# Set BLE-specific filter
uuids = [service_filter] if service_filter else None
await client.set_discovery_filter(adapter, uuids=uuids, transport="le", duplicate_data=True)
await client.set_discovery_filter(
adapter, uuids=uuids, transport="le", duplicate_data=True
)
await client.start_discovery(adapter)
try:
@ -83,22 +84,21 @@ def register_tools(mcp: FastMCP) -> None:
for k, v in d.manufacturer_data.items()
}
svc_data_hex = {
k: v.hex() if isinstance(v, bytes) else str(v) for k, v in d.service_data.items()
k: v.hex() if isinstance(v, bytes) else str(v)
for k, v in d.service_data.items()
}
result.append(
{
"address": d.address,
"name": d.name or "(unknown)",
"rssi": d.rssi,
"paired": d.paired,
"connected": d.connected,
"uuids": [_format_uuid(u) for u in d.uuids],
"manufacturer_data": mfr_data_hex,
"service_data": svc_data_hex,
"appearance": d.appearance,
}
)
result.append({
"address": d.address,
"name": d.name or "(unknown)",
"rssi": d.rssi,
"paired": d.paired,
"connected": d.connected,
"uuids": [_format_uuid(u) for u in d.uuids],
"manufacturer_data": mfr_data_hex,
"service_data": svc_data_hex,
"appearance": d.appearance,
})
return result
@ -318,9 +318,8 @@ def register_tools(mcp: FastMCP) -> None:
When enabled, the device will send updates when the characteristic
value changes (e.g., heart rate measurements, sensor data).
Notification values are buffered (up to 100) and accessible via:
- Resource: bluetooth://ble/{address}/{char_uuid}/notifications
- Resource: bluetooth://ble/{address}/{char_uuid}/notifications/history
Note: To receive actual notifications, you would need to set up
a callback - this tool just enables/disables the notification mode.
Args:
adapter: Adapter name (e.g., "hci0")
@ -329,7 +328,7 @@ def register_tools(mcp: FastMCP) -> None:
enable: True to enable notifications, False to disable
Returns:
Notification status with resource_uri for subscription
Notification status
"""
client = await get_client()
@ -348,35 +347,14 @@ def register_tools(mcp: FastMCP) -> None:
try:
if enable:
# Start BlueZ notifications
await client.start_notify(target_char.path)
# Subscribe to D-Bus signals for buffering
notify_manager = await get_notify_manager()
await notify_manager.subscribe(target_char.path, address, target_char.uuid)
# Build resource URI for client subscription
resource_uri = make_notification_uri(address, target_char.uuid)
return {
"status": "notifications_enabled",
"uuid": char_uuid,
"uuid_short": _format_uuid(char_uuid),
"resource_uri": resource_uri,
"history_uri": f"{resource_uri}/history",
}
else:
# Stop BlueZ notifications
await client.stop_notify(target_char.path)
# Unsubscribe from D-Bus signals
notify_manager = await get_notify_manager()
await notify_manager.unsubscribe(target_char.path)
return {
"status": "notifications_disabled",
"uuid": char_uuid,
}
return {
"status": "notifications_enabled" if enable else "notifications_disabled",
"uuid": char_uuid,
}
except Exception as e:
return {"error": str(e)}
@ -423,87 +401,3 @@ def register_tools(mcp: FastMCP) -> None:
return {"error": "Invalid battery value"}
except Exception as e:
return {"error": str(e)}
@mcp.tool()
async def bt_ble_notification_status() -> dict[str, Any]:
"""List all active BLE notification subscriptions.
Shows which characteristics currently have notifications enabled
and are being buffered. Each subscription shows buffer statistics
and the resource URIs for accessing notification data.
Returns:
List of active notification subscriptions with buffer stats
"""
notify_manager = await get_notify_manager()
subscriptions = notify_manager.list_active_subscriptions()
# Enhance with resource URIs and short UUIDs
result = []
for sub in subscriptions:
resource_uri = make_notification_uri(sub["address"], sub["char_uuid"])
result.append(
{
"address": sub["address"],
"char_uuid": sub["char_uuid"],
"uuid_short": _format_uuid(sub["char_uuid"]),
"notifying": sub["notifying"],
"buffer_count": sub["buffer_count"],
"total_received": sub["total_received"],
"resource_uri": resource_uri,
"history_uri": f"{resource_uri}/history",
}
)
return {
"count": len(result),
"subscriptions": result,
}
@mcp.tool()
async def bt_ble_clear_notification_buffer(
adapter: str,
address: str,
char_uuid: str,
) -> dict[str, Any]:
"""Clear the notification buffer for a characteristic.
Removes all buffered notification values while keeping
the subscription active. Useful for starting fresh or
freeing memory.
Args:
adapter: Adapter name (e.g., "hci0")
address: Device Bluetooth address
char_uuid: Characteristic UUID
Returns:
Status of the clear operation
"""
client = await get_client()
# Find the characteristic to get its path
chars = await client.list_gatt_characteristics(adapter, address)
target_char = None
for c in chars:
if c.uuid.lower() == char_uuid.lower():
target_char = c
break
if not target_char:
return {"error": f"Characteristic {char_uuid} not found"}
notify_manager = await get_notify_manager()
buffer = notify_manager.get_buffer(target_char.path)
if not buffer:
return {"error": "No active notification subscription for this characteristic"}
cleared_count = len(buffer.values)
notify_manager.clear_buffer(target_char.path)
return {
"status": "buffer_cleared",
"uuid": char_uuid,
"cleared_count": cleared_count,
}

681
src/mcbluetooth/tools/bt_elm327_emu.py
View File

@ -1,681 +0,0 @@
"""ELM327 OBD-II BLE Emulator.
Emulates an ELM327 v1.5 adapter over BLE using Nordic UART Service (NUS).
OBD-II apps (Torque, Car Scanner, etc.) connect via BLE, send AT commands
and OBD-II PID queries, and receive simulated responses.
The emulator handles the full ELM327 protocol:
- AT command set (ATZ, ATE0, ATH1, ATSP, etc.)
- Mode 01 (live data) with configurable PID values
- Mode 03 (stored DTCs)
- Mode 09 (VIN query)
- Supported PID bitmaps auto-generated from configured PIDs
NUS UUIDs (same as real BLE OBD adapters):
Service: 6E400001-B5A3-F393-E0A9-E50E24DCCA9E
RX (write): 6E400002-... (client writes commands here)
TX (notify): 6E400003-... (server sends responses here)
"""
from __future__ import annotations
import logging
from collections import deque
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any
from fastmcp import FastMCP
from mcbluetooth.gatt_server import get_gatt_server
log = logging.getLogger(__name__)
# Nordic UART Service UUIDs
NUS_SERVICE_UUID = "6e400001-b5a3-f393-e0a9-e50e24dcca9e"
NUS_RX_CHAR_UUID = "6e400002-b5a3-f393-e0a9-e50e24dcca9e" # Client writes here
NUS_TX_CHAR_UUID = "6e400003-b5a3-f393-e0a9-e50e24dcca9e" # Server notifies here
# Default ECU header (primary ECU response address)
ECU_HEADER = "7E8"
# Common OBD-II PIDs with default values
DEFAULT_PIDS: dict[int, tuple[bytes, str]] = {
0x05: (bytes([70]), "Coolant temp (°C, offset -40) → 30°C"),
0x0C: (bytes([0x0B, 0xB8]), "Engine RPM (× ¼) → 750 RPM"),
0x0D: (bytes([0]), "Vehicle speed (km/h) → 0"),
0x0F: (bytes([55]), "Intake air temp (°C, offset -40) → 15°C"),
0x11: (bytes([25]), "Throttle position (%) → ~10%"),
0x1F: (bytes([0x00, 0x00]), "Runtime since start (sec) → 0"),
0x2F: (bytes([51]), "Fuel tank level (%) → 20%"),
0x46: (bytes([60]), "Ambient air temp (°C, offset -40) → 20°C"),
}
@dataclass
class CommandRecord:
"""A command received from the OBD-II client."""
index: int
timestamp: str
raw: str
parsed: str
response: str
class ELM327Emulator:
"""ELM327 AT command and OBD-II protocol handler."""
def __init__(self):
self.device_id = "ELM327 v1.5"
self.protocol = "6" # ISO 15765-4 CAN (11/500)
self.echo = False
self.headers = True
self.spaces = False
self.linefeed = False
self.pid_values: dict[int, bytes] = {}
self.dtc_codes: list[str] = []
self.vin = "1MCBT00TESTING12345"
self._commands: deque[CommandRecord] = deque(maxlen=200)
self._cmd_index = 0
self._started = False
self._rx_char_id: str | None = None
self._tx_char_id: str | None = None
self._adapter: str | None = None
self._buf = "" # Accumulates partial writes
def set_defaults(self) -> None:
"""Load default PID values for a realistic idle vehicle."""
for pid, (value, _desc) in DEFAULT_PIDS.items():
self.pid_values[pid] = value
def handle_data(self, data: bytes) -> str | None:
"""Process incoming bytes, return response when a complete command is received.
ELM327 commands are terminated with CR (\\r). Accumulates partial
writes until a complete command is received.
"""
try:
text = data.decode("utf-8", errors="replace")
except Exception:
return None
self._buf += text
# Check for command terminator
if "\r" not in self._buf:
return None
# Split on CR, process first complete command
line, self._buf = self._buf.split("\r", 1)
line = line.strip()
if not line:
return None
response = self._process_command(line)
# Record command
record = CommandRecord(
index=self._cmd_index,
timestamp=datetime.now(UTC).isoformat(),
raw=line,
parsed=line.upper().strip(),
response=response.replace("\r", "\\r").replace("\n", "\\n"),
)
self._commands.append(record)
self._cmd_index += 1
return response
def _process_command(self, line: str) -> str:
"""Parse and handle an AT or OBD command."""
cmd = line.strip().upper()
if cmd.startswith("AT"):
return self._handle_at(cmd)
# OBD-II query
return self._handle_obd(cmd)
def _handle_at(self, cmd: str) -> str:
"""Handle ELM327 AT commands."""
cr = "\r\n" if self.linefeed else "\r"
if cmd == "ATZ":
self.echo = False
self.headers = True
self.spaces = False
return f"{cr}{self.device_id}{cr}{cr}>"
if cmd == "ATI":
return f"{self.device_id}{cr}{cr}>"
if cmd == "AT@1":
return f"mcbluetooth OBD-II Emulator{cr}{cr}>"
if cmd == "ATE0":
self.echo = False
return f"OK{cr}{cr}>"
if cmd == "ATE1":
self.echo = True
return f"OK{cr}{cr}>"
if cmd == "ATH0":
self.headers = False
return f"OK{cr}{cr}>"
if cmd == "ATH1":
self.headers = True
return f"OK{cr}{cr}>"
if cmd == "ATS0":
self.spaces = False
return f"OK{cr}{cr}>"
if cmd == "ATS1":
self.spaces = True
return f"OK{cr}{cr}>"
if cmd.startswith("ATL"):
self.linefeed = cmd.endswith("1")
cr = "\r\n" if self.linefeed else "\r"
return f"OK{cr}{cr}>"
if cmd == "ATDPN":
return f"A{self.protocol}{cr}{cr}>"
if cmd == "ATDP":
protocols = {
"0": "AUTO",
"6": "ISO 15765-4 (CAN 11/500)",
"7": "ISO 15765-4 (CAN 29/500)",
"8": "ISO 15765-4 (CAN 11/250)",
"9": "ISO 15765-4 (CAN 29/250)",
}
name = protocols.get(self.protocol, "AUTO")
return f"{name}{cr}{cr}>"
if cmd == "ATRV":
return f"12.6V{cr}{cr}>"
# Commands that just get OK
ok_prefixes = (
"ATSP",
"ATST",
"ATAT",
"ATM",
"ATCAF",
"ATD",
"ATCE",
"ATCF",
"ATCM",
"ATSH",
"ATFC",
"ATWM",
"ATMA",
"ATAR",
"ATPC",
)
for prefix in ok_prefixes:
if cmd.startswith(prefix):
return f"OK{cr}{cr}>"
if cmd == "AT":
return f"OK{cr}{cr}>"
return f"?{cr}{cr}>"
def _handle_obd(self, cmd: str) -> str:
"""Handle OBD-II service queries."""
cr = "\r\n" if self.linefeed else "\r"
cmd = cmd.replace(" ", "")
# Modes 03/04 don't require a PID — pad to 4 chars
if len(cmd) == 2:
cmd += "00"
elif len(cmd) < 4:
return f"?{cr}{cr}>"
try:
mode = int(cmd[:2], 16)
pid = int(cmd[2:4], 16)
except ValueError:
return f"?{cr}{cr}>"
if mode == 0x01:
return self._handle_mode01(pid) + f"{cr}>"
if mode == 0x03:
return self._handle_mode03() + f"{cr}>"
if mode == 0x04:
self.dtc_codes.clear()
return f"44{cr}{cr}>"
if mode == 0x09:
return self._handle_mode09(pid) + f"{cr}>"
return f"NO DATA{cr}{cr}>"
def _handle_mode01(self, pid: int) -> str:
"""Handle Mode 01 (current data) queries."""
cr = "\r\n" if self.linefeed else "\r"
# Supported PID bitmaps
if pid in (0x00, 0x20, 0x40, 0x60, 0x80, 0xA0, 0xC0, 0xE0):
bitmap = self._build_pid_bitmap(pid)
return self._format_response(0x41, pid, bitmap)
# Configured PID value
if pid in self.pid_values:
return self._format_response(0x41, pid, self.pid_values[pid])
return f"NO DATA{cr}"
def _handle_mode03(self) -> str:
"""Handle Mode 03 (stored DTCs) query."""
if not self.dtc_codes:
return self._format_response(0x43, 0x00, bytes([0x00, 0x00, 0x00, 0x00]))
# Pack DTCs into response (2 bytes each)
dtc_bytes = bytearray()
for code in self.dtc_codes[:3]: # Max 3 per frame
dtc_bytes.extend(self._encode_dtc(code))
# Pad to even number of DTCs
while len(dtc_bytes) % 4 != 0:
dtc_bytes.extend(b"\x00\x00")
return self._format_response(0x43, len(self.dtc_codes), bytes(dtc_bytes))
def _handle_mode09(self, pid: int) -> str:
"""Handle Mode 09 (vehicle info) queries."""
cr = "\r\n" if self.linefeed else "\r"
if pid == 0x00:
# Supported Mode 09 PIDs: 02 (VIN)
return self._format_response(0x49, 0x00, bytes([0x40, 0x00, 0x00, 0x00]))
if pid == 0x02:
# VIN (17 chars, multi-frame)
vin_bytes = self.vin.encode("ascii")[:17].ljust(17, b"0")
if self.headers:
lines = []
# First frame: 10 14 49 02 01 VIN[0:3]
lines.append(
f"{ECU_HEADER} 10 14 49 02 01 " + " ".join(f"{b:02X}" for b in vin_bytes[:3])
)
# Consecutive frames
offset = 3
seq = 1
while offset < len(vin_bytes):
chunk = vin_bytes[offset : offset + 7]
frame = f"{ECU_HEADER} 2{seq:01X} " + " ".join(f"{b:02X}" for b in chunk)
lines.append(frame)
offset += 7
seq = (seq + 1) & 0x0F
return cr.join(lines) + cr
else:
return f"4902{vin_bytes.hex().upper()}{cr}"
return f"NO DATA{cr}"
def _format_response(self, mode_resp: int, pid: int, data: bytes) -> str:
"""Format an OBD-II response with optional headers."""
cr = "\r\n" if self.linefeed else "\r"
length = 2 + len(data) # mode + pid + data
if self.headers:
sep = " " if self.spaces else ""
data_hex = sep.join(f"{b:02X}" for b in data)
return f"{ECU_HEADER}{sep}{length:02X}{sep}{mode_resp:02X}{sep}{pid:02X}{sep}{data_hex}{cr}"
else:
data_hex = "".join(f"{b:02X}" for b in data)
return f"{mode_resp:02X}{pid:02X}{data_hex}{cr}"
def _build_pid_bitmap(self, base_pid: int) -> bytes:
"""Build a 4-byte bitmap of supported PIDs for a range."""
bitmap = 0
all_pids = set(self.pid_values.keys())
for i in range(32):
actual_pid = base_pid + i + 1
if actual_pid in all_pids:
bitmap |= 1 << (31 - i)
# Set bit 32 (= bit 0) if the next range has supported PIDs
next_base = base_pid + 0x20
if next_base <= 0xE0:
for p in all_pids:
if next_base < p <= next_base + 0x20:
bitmap |= 1
break
return bitmap.to_bytes(4, "big")
@staticmethod
def _encode_dtc(code: str) -> bytes:
"""Encode a DTC string like 'P0301' into 2 bytes."""
prefixes = {"P": 0, "C": 1, "B": 2, "U": 3}
prefix = prefixes.get(code[0].upper(), 0)
number = int(code[1:], 16)
value = (prefix << 14) | number
return value.to_bytes(2, "big")
def get_commands(
self,
since_index: int = 0,
limit: int = 50,
) -> list[dict[str, Any]]:
"""Get recorded commands."""
records = [
{
"index": r.index,
"timestamp": r.timestamp,
"command": r.raw,
"response_preview": r.response[:80],
}
for r in self._commands
if r.index >= since_index
]
return records[:limit]
# ==================== Module-level emulator ====================
_emulator: ELM327Emulator | None = None
def _get_emulator() -> ELM327Emulator:
"""Get the global emulator instance."""
global _emulator
if _emulator is None:
_emulator = ELM327Emulator()
return _emulator
async def _on_write(char_id: str, value: bytes, device: str | None) -> None:
"""Callback fired when a remote client writes to the RX characteristic."""
emu = _get_emulator()
if not emu._started or char_id != emu._rx_char_id:
return
response = emu.handle_data(value)
if response and emu._tx_char_id:
mgr = get_gatt_server()
mgr.set_value(emu._tx_char_id, response.encode("utf-8"))
# ==================== Tool registration ====================
def register_tools(mcp: FastMCP) -> None:
"""Register ELM327 emulator tools with the MCP server."""
@mcp.tool()
async def bt_elm327_emu_start(
adapter: str,
name: str = "OBDII",
load_defaults: bool = True,
) -> dict[str, Any]:
"""Start the ELM327 OBD-II BLE emulator.
Sets up a Nordic UART Service (NUS) GATT server, registers with
BlueZ, and starts advertising. OBD-II apps can then discover and
connect to this emulated adapter.
The emulator auto-responds to AT commands and OBD-II PID queries.
Configure PID values with bt_elm327_emu_set_pid().
Args:
adapter: Bluetooth adapter (e.g., "hci0").
name: BLE device name shown to scanners (default "OBDII").
load_defaults: Pre-load default PID values for a realistic
idle vehicle (RPM 750, coolant 30°C, etc.).
Returns:
Status with NUS characteristic IDs and default PID info.
"""
emu = _get_emulator()
if emu._started:
return {"status": "error", "error": "Emulator already running"}
try:
if load_defaults:
emu.set_defaults()
mgr = get_gatt_server()
# Build NUS GATT hierarchy
svc_id = mgr.add_service(NUS_SERVICE_UUID)
rx_id = mgr.add_characteristic(
svc_id,
NUS_RX_CHAR_UUID,
flags=["write", "write-without-response"],
)
tx_id = mgr.add_characteristic(
svc_id,
NUS_TX_CHAR_UUID,
flags=["notify"],
value=f"{emu.device_id}\r\n>".encode(),
)
emu._rx_char_id = rx_id
emu._tx_char_id = tx_id
emu._adapter = adapter
# Register write callback for auto-response
mgr._write_callback = _on_write
# Register + advertise
await mgr.register(adapter)
await mgr.set_advertising(
adapter,
True,
name=name,
service_uuids=[NUS_SERVICE_UUID],
)
emu._started = True
pid_info = (
{pid: desc for pid, (_val, desc) in DEFAULT_PIDS.items()} if load_defaults else {}
)
return {
"status": "ok",
"emulator": "ELM327 v1.5",
"adapter": adapter,
"name": name,
"rx_char_id": rx_id,
"tx_char_id": tx_id,
"default_pids": pid_info,
"vin": emu.vin,
}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_elm327_emu_stop(adapter: str) -> dict[str, Any]:
"""Stop the ELM327 OBD-II BLE emulator.
Stops advertising, unregisters GATT services, and cleans up.
Args:
adapter: Bluetooth adapter (e.g., "hci0").
Returns:
Status confirming shutdown.
"""
global _emulator
emu = _get_emulator()
if not emu._started:
return {"status": "ok", "was_running": False}
try:
mgr = get_gatt_server()
mgr._write_callback = None
await mgr.unregister(adapter)
mgr.clear()
except Exception:
pass
emu._started = False
_emulator = None
return {"status": "ok", "was_running": True}
@mcp.tool()
async def bt_elm327_emu_status() -> dict[str, Any]:
"""Get ELM327 emulator status.
Returns:
- running: Whether the emulator is active
- pids: Configured PID values with descriptions
- dtcs: Active diagnostic trouble codes
- command_count: Number of commands received
- settings: Current ELM327 settings (echo, headers, etc.)
"""
emu = _get_emulator()
pids = {}
for pid, value in emu.pid_values.items():
desc = ""
if pid in DEFAULT_PIDS:
_, desc = DEFAULT_PIDS[pid]
pids[f"0x{pid:02X}"] = {
"value_hex": value.hex(),
"value_bytes": list(value),
"description": desc,
}
return {
"status": "ok",
"running": emu._started,
"device_id": emu.device_id,
"adapter": emu._adapter,
"protocol": emu.protocol,
"pids": pids,
"dtcs": emu.dtc_codes,
"vin": emu.vin,
"command_count": emu._cmd_index,
"settings": {
"echo": emu.echo,
"headers": emu.headers,
"spaces": emu.spaces,
"linefeed": emu.linefeed,
},
}
@mcp.tool()
async def bt_elm327_emu_set_pid(
pid: int,
value: str,
value_type: str = "hex",
) -> dict[str, Any]:
"""Set an OBD-II PID response value.
When an OBD app queries this PID (Mode 01), the emulator returns
the configured value. Values follow OBD-II encoding conventions.
Common PIDs:
0x05: Coolant temp (1 byte, value - 40 = °C)
0x0C: Engine RPM (2 bytes, value / 4 = RPM)
0x0D: Vehicle speed (1 byte = km/h)
0x0F: Intake air temp (1 byte, value - 40 = °C)
0x11: Throttle position (1 byte, value / 2.55 = %)
0x2F: Fuel tank level (1 byte, value / 2.55 = %)
0x46: Ambient air temp (1 byte, value - 40 = °C)
Args:
pid: PID number (e.g., 0x0C for RPM).
value: Encoded value. For RPM of 3000: "0BB8" (hex) or "3000" (int).
value_type: "hex" (raw bytes), "int" (little-endian integer),
or "string" (UTF-8).
Returns:
Status with the configured PID value.
"""
emu = _get_emulator()
try:
if value_type == "hex":
data = bytes.fromhex(value)
elif value_type == "int":
int_val = int(value)
byte_len = max(1, (int_val.bit_length() + 7) // 8)
data = int_val.to_bytes(byte_len, "big") # OBD uses big-endian
elif value_type == "string":
data = value.encode("utf-8")
else:
return {"status": "error", "error": f"Unknown value_type: {value_type}"}
emu.pid_values[pid] = data
return {
"status": "ok",
"pid": f"0x{pid:02X}",
"value_hex": data.hex(),
"value_bytes": list(data),
}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_elm327_emu_set_dtcs(dtc_codes: list[str]) -> dict[str, Any]:
"""Set diagnostic trouble codes for Mode 03 queries.
Args:
dtc_codes: List of DTC strings (e.g., ["P0301", "P0420", "C0035"]).
P = Powertrain, C = Chassis, B = Body, U = Network.
Returns:
Status with configured DTCs.
"""
emu = _get_emulator()
emu.dtc_codes = [c.upper() for c in dtc_codes]
return {
"status": "ok",
"dtcs": emu.dtc_codes,
"count": len(emu.dtc_codes),
}
@mcp.tool()
async def bt_elm327_emu_set_vin(vin: str) -> dict[str, Any]:
"""Set the Vehicle Identification Number for Mode 09 queries.
Args:
vin: 17-character VIN string.
Returns:
Status with configured VIN.
"""
emu = _get_emulator()
if len(vin) != 17:
return {"status": "error", "error": f"VIN must be 17 characters (got {len(vin)})"}
emu.vin = vin.upper()
return {"status": "ok", "vin": emu.vin}
@mcp.tool()
async def bt_elm327_emu_read_commands(
since_index: int = 0,
limit: int = 50,
) -> dict[str, Any]:
"""Read commands received from the OBD-II client.
Monitor what the connected app is requesting. Useful for debugging
protocol issues or understanding app behavior.
Args:
since_index: Only return commands with index >= this value.
limit: Maximum commands to return.
Returns:
List of received commands with timestamps and responses.
"""
emu = _get_emulator()
cmds = emu.get_commands(since_index, limit)
next_idx = max((c["index"] for c in cmds), default=since_index) + 1 if cmds else since_index
return {
"status": "ok",
"commands": cmds,
"count": len(cmds),
"next_index": next_idx,
}

392
src/mcbluetooth/tools/gatt_server.py
View File

@ -1,392 +0,0 @@
"""GATT Server tools for Bluetooth MCP server.
These tools let Linux act as a BLE peripheral (GATT server) that
advertises services, accepts connections, and handles reads/writes
from remote central devices. Use cases include: emulating sensors,
building test harnesses, mocking BLE devices, protocol fuzzing.
Typical flow:
1. bt_gatt_server_add_service(uuid) → service0
2. bt_gatt_server_add_characteristic(service0, ...) → service0/char0
3. bt_gatt_server_register(adapter="hci0") → registered
4. bt_gatt_server_advertise(adapter="hci0", enable=True, name="MyDevice")
5. Remote connects, writes → bt_gatt_server_read_writes()
6. Respond via bt_gatt_server_set_value() → auto-notifies
"""
from __future__ import annotations
from typing import Any
from fastmcp import FastMCP
from mcbluetooth.gatt_server import get_gatt_server, shutdown_gatt_server
def _parse_value(value: str, value_type: str) -> bytes:
"""Convert a string value to bytes based on value_type."""
if value_type == "hex":
return bytes.fromhex(value)
elif value_type == "string":
return value.encode("utf-8")
elif value_type == "int":
int_val = int(value)
byte_len = max(1, (int_val.bit_length() + 7) // 8)
return int_val.to_bytes(byte_len, "little")
else:
raise ValueError(f"Unknown value_type: {value_type}")
def register_tools(mcp: FastMCP) -> None:
"""Register GATT server tools with the MCP server."""
# ---- Lifecycle ----
@mcp.tool()
async def bt_gatt_server_register(adapter: str) -> dict[str, Any]:
"""Register the GATT application with BlueZ.
Exports all added services/characteristics to D-Bus and calls
GattManager1.RegisterApplication. Must add at least one service
before registering.
After registration, remote BLE centrals can discover and interact
with the hosted services (once advertising is started).
Args:
adapter: Adapter name (e.g., "hci0").
Returns:
Registration status with service count.
"""
try:
mgr = get_gatt_server()
await mgr.register(adapter)
status = mgr.get_status()
return {
"status": "ok",
"registered": True,
"service_count": len(status["services"]),
}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_gatt_server_unregister(adapter: str) -> dict[str, Any]:
"""Unregister the GATT application from BlueZ.
Stops advertising (if active) and removes the application
registration. Services remain in memory and can be re-registered.
Args:
adapter: Adapter name (e.g., "hci0").
Returns:
Status confirming unregistration.
"""
try:
await shutdown_gatt_server(adapter)
return {"status": "ok", "registered": False}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_gatt_server_status() -> dict[str, Any]:
"""Get GATT server status.
Returns:
- registered: Whether the application is registered with BlueZ
- advertising: Whether BLE advertising is active
- services: List of services with their characteristics
- write_event_count: Number of buffered write events
"""
mgr = get_gatt_server()
return {"status": "ok", **mgr.get_status()}
# ---- Service construction ----
@mcp.tool()
async def bt_gatt_server_add_service(
uuid: str,
primary: bool = True,
) -> dict[str, Any]:
"""Add a GATT service to the server.
Services must be added BEFORE registering with BlueZ.
If already registered, unregister first, add services, then re-register.
Args:
uuid: Service UUID (e.g., "180F" for Battery Service, or full
128-bit "6E400001-B5A3-F393-E0A9-E50E24DCCA9E" for custom).
primary: Whether this is a primary service (default True).
Returns:
The assigned service_id (e.g., "service0") for adding characteristics.
"""
try:
mgr = get_gatt_server()
service_id = mgr.add_service(uuid, primary)
return {"status": "ok", "service_id": service_id, "uuid": uuid}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_gatt_server_add_characteristic(
service_id: str,
uuid: str,
flags: list[str],
value: str | None = None,
value_type: str = "hex",
) -> dict[str, Any]:
"""Add a characteristic to a GATT service.
Args:
service_id: Service to add to (e.g., "service0").
uuid: Characteristic UUID.
flags: List of flags/permissions. Common values:
"read", "write", "write-without-response", "notify", "indicate",
"encrypt-read", "encrypt-write".
value: Initial value (optional). Format depends on value_type.
value_type: How to interpret value — "hex", "string", or "int".
Returns:
The assigned char_id (e.g., "service0/char0") for setting values.
"""
try:
mgr = get_gatt_server()
initial = _parse_value(value, value_type) if value else b""
char_id = mgr.add_characteristic(service_id, uuid, flags, initial)
return {
"status": "ok",
"char_id": char_id,
"uuid": uuid,
"flags": flags,
}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_gatt_server_add_descriptor(
char_id: str,
uuid: str,
flags: list[str],
value: str | None = None,
value_type: str = "hex",
) -> dict[str, Any]:
"""Add a descriptor to a GATT characteristic.
Note: CCCD (0x2902) for notify/indicate is auto-managed by BlueZ —
don't add it manually.
Args:
char_id: Characteristic to add to (e.g., "service0/char0").
uuid: Descriptor UUID (e.g., "2901" for Characteristic User Description).
flags: List of flags (e.g., ["read"], ["read", "write"]).
value: Initial value (optional).
value_type: How to interpret value — "hex", "string", or "int".
Returns:
The assigned desc_id.
"""
try:
mgr = get_gatt_server()
initial = _parse_value(value, value_type) if value else b""
desc_id = mgr.add_descriptor(char_id, uuid, flags, initial)
return {"status": "ok", "desc_id": desc_id, "uuid": uuid}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_gatt_server_clear() -> dict[str, Any]:
"""Remove all services, characteristics, and descriptors.
Must unregister first. Use this to rebuild the service hierarchy
from scratch.
Returns:
Status confirming services were cleared.
"""
try:
mgr = get_gatt_server()
mgr.clear()
return {"status": "ok", "cleared": True}
except Exception as exc:
return {"status": "error", "error": str(exc)}
# ---- Value management ----
@mcp.tool()
async def bt_gatt_server_set_value(
char_id: str,
value: str,
value_type: str = "hex",
) -> dict[str, Any]:
"""Set a characteristic value on the GATT server.
If a remote client has subscribed to notifications on this
characteristic, the new value is automatically pushed via
BLE notification.
Args:
char_id: Characteristic ID (e.g., "service0/char0").
value: Value to set.
value_type: How to interpret value — "hex" (e.g., "0102ff"),
"string" (UTF-8 text), or "int" (decimal number).
Returns:
Status with value details and whether notification was sent.
"""
try:
mgr = get_gatt_server()
data = _parse_value(value, value_type)
char = mgr._characteristics.get(char_id)
was_notifying = char.notifying if char else False
mgr.set_value(char_id, data)
return {
"status": "ok",
"char_id": char_id,
"value_hex": data.hex(),
"length": len(data),
"notified": was_notifying,
}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_gatt_server_get_value(char_id: str) -> dict[str, Any]:
"""Read a characteristic value from the server side.
This reads the value stored on OUR server, not from a remote device.
Use bt_ble_read() for reading from remote GATT servers.
Args:
char_id: Characteristic ID (e.g., "service0/char0").
Returns:
The current value as hex, bytes, and string (if decodable).
"""
try:
mgr = get_gatt_server()
value = mgr.get_value(char_id)
result: dict[str, Any] = {
"status": "ok",
"char_id": char_id,
"value_hex": value.hex(),
"value_bytes": list(value),
"length": len(value),
}
try:
result["value_string"] = value.decode("utf-8")
except UnicodeDecodeError:
pass
return result
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_gatt_server_notify(char_id: str) -> dict[str, Any]:
"""Explicitly send a BLE notification for the current value.
Normally, bt_gatt_server_set_value auto-notifies. Use this to
re-send the current value without changing it.
Args:
char_id: Characteristic ID (e.g., "service0/char0").
Returns:
Status indicating whether notification was sent.
"""
try:
mgr = get_gatt_server()
sent = mgr.notify(char_id)
if sent:
return {"status": "ok", "notified": True}
return {
"status": "ok",
"notified": False,
"reason": "No client subscribed to notifications",
}
except Exception as exc:
return {"status": "error", "error": str(exc)}
# ---- Advertising ----
@mcp.tool()
async def bt_gatt_server_advertise(
adapter: str,
enable: bool,
name: str = "mcbluetooth",
service_uuids: list[str] | None = None,
) -> dict[str, Any]:
"""Start or stop BLE advertising.
When advertising, the device becomes discoverable to BLE scanners
and can accept connections from central devices.
Must register the GATT application first.
Args:
adapter: Adapter name (e.g., "hci0").
enable: True to start advertising, False to stop.
name: Local name shown to scanners (default "mcbluetooth").
service_uuids: UUIDs to include in advertisement. If not specified,
uses all registered service UUIDs.
Returns:
Advertising status.
"""
try:
mgr = get_gatt_server()
await mgr.set_advertising(adapter, enable, name=name, service_uuids=service_uuids)
return {
"status": "ok",
"advertising": enable,
"name": name if enable else None,
}
except Exception as exc:
return {"status": "error", "error": str(exc)}
# ---- Write event monitoring ----
@mcp.tool()
async def bt_gatt_server_read_writes(
since_index: int = 0,
char_id: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
"""Read write events from remote BLE clients.
When a remote central device writes to a server characteristic,
the write is buffered here. Poll this to see incoming commands/data.
Args:
since_index: Only return events with index >= this value.
Start with 0, then use the highest returned index + 1.
char_id: Filter to writes on this characteristic (optional).
limit: Maximum events to return (default 50).
Returns:
List of write events with value, timestamp, and source device.
"""
mgr = get_gatt_server()
events = mgr.get_write_events(since_index, char_id, limit)
next_index = (
max((e["index"] for e in events), default=since_index) + 1 if events else since_index
)
return {
"status": "ok",
"events": events,
"count": len(events),
"next_index": next_index,
}
@mcp.tool()
async def bt_gatt_server_clear_writes() -> dict[str, Any]:
"""Clear the write event buffer.
Returns:
Number of events cleared.
"""
mgr = get_gatt_server()
count = mgr.clear_write_events()
return {"status": "ok", "cleared_count": count}

243
src/mcbluetooth/tools/spp.py
View File

@ -1,243 +0,0 @@
"""SPP (Serial Port Profile) tools for Bluetooth MCP server.
Classic Bluetooth serial communication over RFCOMM. SPP provides raw
bidirectional byte streams — the classic BT equivalent of BLE's Nordic
UART Service.
Typical flow:
Server mode:
1. bt_spp_enable() → Register SPP profile
2. Make adapter discoverable → Remote device connects
3. bt_spp_send(address, "Hello\r\n") → Send data
4. bt_spp_recv() → Poll received data
Client mode:
1. bt_spp_enable() → Register SPP profile
2. bt_spp_connect(adapter, address) → Connect to remote SPP
3. bt_spp_send / bt_spp_recv → Bidirectional I/O
"""
from __future__ import annotations
from typing import Any, Literal
from fastmcp import FastMCP
from mcbluetooth.dbus_client import get_client
from mcbluetooth.spp import (
SPP_UUID,
disable_spp,
enable_spp,
get_spp,
)
def register_tools(mcp: FastMCP) -> None:
"""Register SPP tools with the MCP server."""
@mcp.tool()
async def bt_spp_enable(
uuid: str = SPP_UUID,
channel: int = 0,
name: str = "mcbluetooth SPP",
) -> dict[str, Any]:
"""Enable SPP (Serial Port Profile) for classic Bluetooth serial.
Registers an RFCOMM serial profile with BlueZ. After enabling,
remote devices can connect for bidirectional serial communication,
or use bt_spp_connect() to initiate outbound connections.
Args:
uuid: Service UUID. Default is standard SPP (0x1101).
Use custom UUIDs for Arduino/ESP32 with non-standard RFCOMM.
channel: RFCOMM channel (0 = auto-assign, recommended).
name: Profile display name visible during service discovery.
Returns:
Registration status with active UUID.
"""
try:
await enable_spp(uuid=uuid, channel=channel, name=name)
return {"status": "ok", "uuid": uuid, "channel": channel or "auto", "name": name}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_spp_disable() -> dict[str, Any]:
"""Disable SPP and close all serial connections.
Unregisters the SPP profile from BlueZ and terminates any active
RFCOMM sessions.
Returns:
Status confirming profile removal.
"""
try:
await disable_spp()
return {"status": "ok", "disabled": True}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_spp_status() -> dict[str, Any]:
"""Get SPP status: registration state, connections, and buffer stats.
Returns:
- registered: Whether the SPP profile is active
- uuid: The UUID the profile was registered with
- connections: Active peers with role, duration, byte counters
- recv_buffer_count: Buffered received data events
- recv_buffer_total: Total events received since enable
"""
profile = await get_spp()
if not profile:
return {"status": "ok", "registered": False, "connections": []}
return {"status": "ok", **profile.get_status()}
@mcp.tool()
async def bt_spp_connect(
adapter: str,
address: str,
uuid: str = SPP_UUID,
) -> dict[str, Any]:
"""Connect to a remote device's SPP service (client mode).
Initiates an outbound RFCOMM connection. The SPP profile must be
enabled first (bt_spp_enable). BlueZ will deliver the RFCOMM fd
through our Profile1 handler automatically.
The device should already be paired and trusted.
Args:
adapter: Bluetooth adapter (e.g. "hci0").
address: Remote device Bluetooth address.
uuid: SPP service UUID on the remote device.
Returns:
Connection status.
"""
profile = await get_spp()
if not profile:
return {"status": "error", "error": "SPP not enabled — call bt_spp_enable() first"}
try:
client = await get_client()
await client.connect_profile(adapter, address, uuid)
# BlueZ fires NewConnection on our Profile1 handler — tag as client
device_path = f"/org/bluez/{adapter}/dev_{address.upper().replace(':', '_')}"
profile.add_client_connection(device_path, address, -1)
return {"status": "ok", "address": address, "role": "client"}
except Exception as exc:
return {"status": "error", "error": str(exc)}
@mcp.tool()
async def bt_spp_disconnect(address: str) -> dict[str, Any]:
"""Disconnect a specific SPP peer.
Closes the RFCOMM socket and removes the connection.
Args:
address: Bluetooth address of the peer to disconnect.
Returns:
Disconnect status.
"""
profile = await get_spp()
if not profile:
return {"status": "error", "error": "SPP not enabled"}
conn = profile._get_connection(address)
if not conn:
return {"status": "error", "error": f"No SPP connection to {address}"}
profile.connections.pop(conn.device_path, None)
profile._cleanup_connection(conn)
return {"status": "ok", "address": address, "disconnected": True}
@mcp.tool()
async def bt_spp_send(
address: str,
data: str,
data_type: Literal["string", "hex", "line"] = "string",
) -> dict[str, Any]:
"""Send data to a connected SPP peer.
Args:
address: Bluetooth address of the peer.
data: The data to send.
data_type: How to interpret the data parameter:
- "string": Send as-is (UTF-8 encoded)
- "hex": Parse as hex string (e.g. "48656c6c6f")
- "line": Append CR+LF for line-oriented protocols
(AT commands, NMEA, etc.)
Returns:
Send status with byte count.
"""
profile = await get_spp()
if not profile:
return {"status": "error", "error": "SPP not enabled"}
if data_type == "hex":
try:
raw = bytes.fromhex(data)
except ValueError as exc:
return {"status": "error", "error": f"Invalid hex: {exc}"}
elif data_type == "line":
raw = (data + "\r\n").encode("utf-8")
else:
raw = data.encode("utf-8")
ok = await profile.send(address, raw)
if ok:
return {"status": "ok", "bytes_sent": len(raw)}
return {"status": "error", "error": f"No active SPP connection to {address}"}
@mcp.tool()
async def bt_spp_recv(
since_index: int = 0,
address: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
"""Read received data from SPP connections (cursor-based polling).
Returns buffered data events since the given index. Use the highest
returned index + 1 as since_index for the next poll to avoid
duplicates.
Args:
since_index: Return events with index >= this value.
Start at 0 for first call, then use last index + 1.
address: Filter by peer address (optional).
limit: Maximum number of events to return.
Returns:
List of data events with index, timestamp, address, and values.
"""
profile = await get_spp()
if not profile:
return {"status": "ok", "events": [], "hint": "SPP not enabled"}
events = profile.get_recv_events(since_index=since_index, address=address, limit=limit)
return {
"status": "ok",
"count": len(events),
"events": events,
"next_index": events[-1]["index"] + 1 if events else since_index,
}
@mcp.tool()
async def bt_spp_clear_recv() -> dict[str, Any]:
"""Clear the SPP receive buffer.
Removes all buffered received data events. The index counter
continues from where it left off (not reset to 0).
Returns:
Count of events cleared.
"""
profile = await get_spp()
if not profile:
return {"status": "ok", "cleared": 0}
count = profile.clear_recv_events()
return {"status": "ok", "cleared": count}