0
# GATT Client Operations
1
2
Bleak's `BleakClient` provides comprehensive GATT client functionality for connecting to BLE devices and performing operations on services, characteristics, and descriptors. It supports async context management and provides a unified API across all platforms.
3
4
## Capabilities
5
6
### Client Initialization and Configuration
7
8
Initialize GATT clients with device specifications, connection parameters, and platform-specific options.
9
10
```python { .api }
11
class BleakClient:
12
def __init__(
13
self,
14
address_or_ble_device: Union[BLEDevice, str],
15
disconnected_callback: Optional[Callable[[BleakClient], None]] = None,
16
services: Optional[Iterable[str]] = None,
17
*,
18
timeout: float = 10.0,
19
pair: bool = False,
20
winrt: WinRTClientArgs = {},
21
backend: Optional[type[BaseBleakClient]] = None,
22
**kwargs: Any,
23
) -> None:
24
"""
25
Initialize GATT client.
26
27
Args:
28
address_or_ble_device: BLEDevice from scanner or Bluetooth address string
29
disconnected_callback: Function called when device disconnects
30
services: Optional list of service UUIDs to resolve
31
timeout: Connection timeout in seconds
32
pair: Attempt pairing before connecting (not supported on macOS)
33
winrt: Windows-specific client arguments
34
backend: Custom backend implementation
35
"""
36
```
37
38
### Connection Management
39
40
Establish and manage GATT connections with async context manager support for automatic lifecycle management.
41
42
```python { .api }
43
async def connect(self, **kwargs: Any) -> None:
44
"""
45
Connect to the GATT server.
46
47
Args:
48
**kwargs: Backward compatibility arguments (should not be used)
49
"""
50
51
async def disconnect(self) -> None:
52
"""Disconnect from the GATT server."""
53
54
async def __aenter__(self) -> Self:
55
"""Async context manager entry - connects to device."""
56
57
async def __aexit__(
58
self,
59
exc_type: Optional[type[BaseException]],
60
exc_val: Optional[BaseException],
61
exc_tb: Optional[TracebackType],
62
) -> None:
63
"""Async context manager exit - disconnects from device."""
64
```
65
66
### Device Information Access
67
68
Access basic device information and connection status.
69
70
```python { .api }
71
@property
72
def name(self) -> str:
73
"""Human-readable name for the peripheral device."""
74
75
@property
76
def address(self) -> str:
77
"""Bluetooth address of this device (UUID on macOS)."""
78
79
@property
80
def mtu_size(self) -> int:
81
"""Negotiated MTU size in bytes for the active connection."""
82
83
@property
84
def is_connected(self) -> bool:
85
"""Check connection status between client and GATT server."""
86
```
87
88
### Pairing and Security
89
90
Manage device pairing and security operations.
91
92
```python { .api }
93
async def pair(self, *args: Any, **kwargs: Any) -> None:
94
"""
95
Pair with the GATT server.
96
97
Note: Not available on macOS. User will be prompted automatically
98
when accessing characteristics requiring authentication.
99
100
Args:
101
*args, **kwargs: Backend-specific pairing arguments
102
"""
103
104
async def unpair(self) -> None:
105
"""
106
Unpair from the GATT server.
107
108
Note: Only available on Windows and Linux. Unpairing also disconnects.
109
"""
110
```
111
112
### Service Discovery
113
114
Access the device's GATT service hierarchy after connection.
115
116
```python { .api }
117
@property
118
def services(self) -> BleakGATTServiceCollection:
119
"""
120
Collection of GATT services available on the device.
121
122
Returns:
123
BleakGATTServiceCollection containing all discovered services
124
125
Raises:
126
BleakError: If service discovery has not been performed
127
"""
128
```
129
130
### Characteristic Operations
131
132
Read from and write to GATT characteristics with support for different write modes.
133
134
```python { .api }
135
async def read_gatt_char(
136
self,
137
char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
138
**kwargs: Any,
139
) -> bytearray:
140
"""
141
Read data from a GATT characteristic.
142
143
Args:
144
char_specifier: Characteristic specified by handle, UUID, or object
145
**kwargs: Backend-specific read arguments
146
147
Returns:
148
Bytearray containing the read data
149
150
Raises:
151
BleakCharacteristicNotFoundError: If characteristic not found
152
"""
153
154
async def write_gatt_char(
155
self,
156
char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
157
data: Buffer,
158
response: Optional[bool] = None,
159
) -> None:
160
"""
161
Write data to a GATT characteristic.
162
163
Args:
164
char_specifier: Characteristic specified by handle, UUID, or object
165
data: Data to write (supports buffer protocol)
166
response: True for write-with-response, False for write-without-response,
167
None for automatic selection based on characteristic properties
168
169
Raises:
170
BleakCharacteristicNotFoundError: If characteristic not found
171
"""
172
```
173
174
### Notification and Indication Handling
175
176
Subscribe to and manage notifications and indications from GATT characteristics.
177
178
```python { .api }
179
async def start_notify(
180
self,
181
char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
182
callback: Callable[
183
[BleakGATTCharacteristic, bytearray], Union[None, Awaitable[None]]
184
],
185
*,
186
cb: CBStartNotifyArgs = {},
187
**kwargs: Any,
188
) -> None:
189
"""
190
Activate notifications/indications on a characteristic.
191
192
Args:
193
char_specifier: Characteristic specified by handle, UUID, or object
194
callback: Function called when notifications are received
195
cb: CoreBluetooth-specific notification arguments
196
**kwargs: Backend-specific notification arguments
197
198
Raises:
199
BleakCharacteristicNotFoundError: If characteristic not found
200
BleakError: If not connected
201
"""
202
203
async def stop_notify(
204
self, char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID]
205
) -> None:
206
"""
207
Deactivate notifications/indications on a characteristic.
208
209
Args:
210
char_specifier: Characteristic specified by handle, UUID, or object
211
212
Raises:
213
BleakCharacteristicNotFoundError: If characteristic not found
214
"""
215
```
216
217
### Descriptor Operations
218
219
Read from and write to GATT descriptors for advanced characteristic configuration.
220
221
```python { .api }
222
async def read_gatt_descriptor(
223
self,
224
desc_specifier: Union[BleakGATTDescriptor, int],
225
**kwargs: Any,
226
) -> bytearray:
227
"""
228
Read data from a GATT descriptor.
229
230
Args:
231
desc_specifier: Descriptor specified by handle or object
232
**kwargs: Backend-specific read arguments
233
234
Returns:
235
Bytearray containing the read data
236
237
Raises:
238
BleakError: If descriptor not found
239
"""
240
241
async def write_gatt_descriptor(
242
self,
243
desc_specifier: Union[BleakGATTDescriptor, int],
244
data: Buffer,
245
) -> None:
246
"""
247
Write data to a GATT descriptor.
248
249
Args:
250
desc_specifier: Descriptor specified by handle or object
251
data: Data to write (supports buffer protocol)
252
253
Raises:
254
BleakError: If descriptor not found
255
"""
256
```
257
258
## Usage Examples
259
260
### Basic Connection and Reading
261
262
```python
263
import asyncio
264
from bleak import BleakClient
265
266
async def connect_and_read():
267
address = "00:11:22:33:44:55" # Replace with actual device address
268
269
async with BleakClient(address) as client:
270
print(f"Connected to {client.name}")
271
272
# Read device name characteristic
273
device_name_uuid = "00002a00-0000-1000-8000-00805f9b34fb"
274
name_bytes = await client.read_gatt_char(device_name_uuid)
275
print(f"Device name: {name_bytes.decode()}")
276
277
# Read battery level if available
278
battery_uuid = "00002a19-0000-1000-8000-00805f9b34fb"
279
try:
280
battery_bytes = await client.read_gatt_char(battery_uuid)
281
battery_level = int.from_bytes(battery_bytes, byteorder='little')
282
print(f"Battery level: {battery_level}%")
283
except Exception as e:
284
print(f"Battery level not available: {e}")
285
286
asyncio.run(connect_and_read())
287
```
288
289
### Service and Characteristic Exploration
290
291
```python
292
import asyncio
293
from bleak import BleakClient
294
295
async def explore_services():
296
address = "00:11:22:33:44:55" # Replace with actual device address
297
298
async with BleakClient(address) as client:
299
print(f"Connected to {client.name}")
300
print(f"MTU Size: {client.mtu_size}")
301
302
# Explore all services and characteristics
303
for service in client.services:
304
print(f"\nService: {service.uuid} - {service.description}")
305
306
for char in service.characteristics:
307
print(f" Char: {char.uuid} - {char.description}")
308
print(f" Properties: {char.properties}")
309
print(f" Handle: {char.handle}")
310
311
# List descriptors
312
for desc in char.descriptors:
313
print(f" Desc: {desc.uuid} - {desc.description}")
314
315
asyncio.run(explore_services())
316
```
317
318
### Writing Data
319
320
```python
321
import asyncio
322
from bleak import BleakClient
323
324
async def write_data():
325
address = "00:11:22:33:44:55" # Replace with actual device address
326
char_uuid = "12345678-1234-5678-9012-123456789abc" # Replace with actual UUID
327
328
async with BleakClient(address) as client:
329
print(f"Connected to {client.name}")
330
331
# Write with response (reliable)
332
data = b"Hello, BLE!"
333
await client.write_gatt_char(char_uuid, data, response=True)
334
print("Data written with response")
335
336
# Write without response (faster)
337
await client.write_gatt_char(char_uuid, data, response=False)
338
print("Data written without response")
339
340
asyncio.run(write_data())
341
```
342
343
### Notifications
344
345
```python
346
import asyncio
347
from bleak import BleakClient, BleakGATTCharacteristic
348
349
def notification_handler(characteristic: BleakGATTCharacteristic, data: bytearray):
350
"""Handle incoming notifications."""
351
print(f"Notification from {characteristic.uuid}: {data.hex()}")
352
353
async def handle_notifications():
354
address = "00:11:22:33:44:55" # Replace with actual device address
355
notify_uuid = "12345678-1234-5678-9012-123456789abc" # Replace with actual UUID
356
357
async with BleakClient(address) as client:
358
print(f"Connected to {client.name}")
359
360
# Start notifications
361
await client.start_notify(notify_uuid, notification_handler)
362
print("Notifications started")
363
364
# Keep connection alive to receive notifications
365
await asyncio.sleep(30)
366
367
# Stop notifications (automatic on disconnect)
368
await client.stop_notify(notify_uuid)
369
print("Notifications stopped")
370
371
asyncio.run(handle_notifications())
372
```
373
374
### Async Notification Callbacks
375
376
```python
377
import asyncio
378
from bleak import BleakClient, BleakGATTCharacteristic
379
380
async def async_notification_handler(characteristic: BleakGATTCharacteristic, data: bytearray):
381
"""Async notification handler for complex processing."""
382
print(f"Processing notification from {characteristic.uuid}")
383
384
# Simulate async processing
385
await asyncio.sleep(0.1)
386
387
value = int.from_bytes(data, byteorder='little', signed=False)
388
print(f"Processed value: {value}")
389
390
async def async_notifications():
391
address = "00:11:22:33:44:55" # Replace with actual device address
392
notify_uuid = "12345678-1234-5678-9012-123456789abc" # Replace with actual UUID
393
394
async with BleakClient(address) as client:
395
await client.start_notify(notify_uuid, async_notification_handler)
396
await asyncio.sleep(30) # Receive notifications for 30 seconds
397
398
asyncio.run(async_notifications())
399
```
400
401
### Connection with Pairing
402
403
```python
404
import asyncio
405
from bleak import BleakClient
406
407
async def connect_with_pairing():
408
address = "00:11:22:33:44:55" # Replace with actual device address
409
410
# Enable pairing during connection (not supported on macOS)
411
client = BleakClient(address, pair=True, timeout=30.0)
412
413
async with client:
414
print(f"Connected and paired with {client.name}")
415
416
# Perform operations that might require authentication
417
secured_char_uuid = "12345678-1234-5678-9012-123456789abc"
418
try:
419
data = await client.read_gatt_char(secured_char_uuid)
420
print(f"Secured data: {data.hex()}")
421
except Exception as e:
422
print(f"Could not read secured characteristic: {e}")
423
424
# Only run on platforms that support explicit pairing
425
import platform
426
if platform.system() != "Darwin": # Not macOS
427
asyncio.run(connect_with_pairing())
428
```
429
430
## Types
431
432
```python { .api }
433
# Buffer type for write operations
434
Buffer = Union[bytes, bytearray, memoryview]
435
436
# Platform-specific client arguments
437
class WinRTClientArgs(TypedDict, total=False):
438
address_type: Literal["public", "random"]
439
use_cached_services: bool
440
441
class CBStartNotifyArgs(TypedDict, total=False):
442
notification_discriminator: Optional[NotificationDiscriminator]
443
444
# Notification discriminator for CoreBluetooth
445
NotificationDiscriminator = Callable[[bytes], bool]
446
447
# Callback types
448
DisconnectedCallback = Callable[[BleakClient], None]
449
NotificationCallback = Callable[
450
[BleakGATTCharacteristic, bytearray],
451
Union[None, Awaitable[None]]
452
]
453
```