Connecting to ChirpStack

The ChirpStack backend supplies device inventory, live events, and downlinks over gRPC. Provision devices and gateways in ChirpStack, then connect with an API key.

Connect your collection

Install the backend with pip install "lorawan-connection[chirpstack]".

Save the SenseCAP example library as sensecap_lorawan.py. Save the script below as connect_sensors.py in the same directory. Set CHIRPSTACK_SERVER to your server URL and CHIRPSTACK_API_KEY to your API key. Set CHIRPSTACK_TENANT_ID if your key cannot list tenants.

python connect_sensors.py

The script prints supported devices and readings from all applications accessible to the key. A disconnect or Ctrl+C closes the collection and connection.

"""Print live SenseCAP readings until disconnected or interrupted."""

import asyncio
import os

from lorawan_connection import ConnectionUnavailable, DeviceCollection
from lorawan_connection.backend.chirpstack import ChirpStackConnection
from sensecap_lorawan import S2101


async def main() -> None:
    connection = ChirpStackConnection(
        endpoint=os.environ["CHIRPSTACK_SERVER"],
        api_key=os.environ["CHIRPSTACK_API_KEY"],
        tenant_id=os.environ.get("CHIRPSTACK_TENANT_ID"),
        application_ids=[],
        network_id="my-network",
    )
    devices = DeviceCollection(connection, [S2101])
    disconnected: asyncio.Future[Exception] = asyncio.get_running_loop().create_future()

    def on_disconnect() -> None:
        if not disconnected.done():
            disconnected.set_result(
                connection.error or ConnectionUnavailable("Disconnected")
            )

    def device_added(device: S2101) -> None:
        print(f"Device: {device.descriptor.name}")
        device.add_update_listener(
            lambda: print(
                f"{device.descriptor.name}: "
                f"temperature={device.temperature}, humidity={device.humidity}"
            )
        )

    devices.subscribe_device_added(device_added)
    try:
        connection.application_ids = list(await connection.applications())
        connection.on_disconnect(on_disconnect)
        await connection.async_connect()
        await devices.async_setup()
        raise await disconnected
    finally:
        devices.close()
        await connection.close()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

network_id is the caller’s stable identifier for this connection’s device namespace. Use https for TLS with system trust roots. http explicitly selects an unencrypted connection. TLS errors never cause a plaintext fallback.

Omit tenant_id or pass None to discover applications across all accessible tenants. Supply a tenant UUID to restrict discovery to that tenant, or when the API key cannot list tenants. application_ids selects which applications supply devices and events.

Existing devices and events

The backend polls the device list every 30 seconds and opens an internal ChirpStack event stream for each device. Catalog profiles supply model and vendor identities. An incomplete read preserves the previous device list. Custom profiles without catalog identity remain visible but do not select a device model.

The internal stream returns JSON. The adapter parses each message and constructs a typed event with the relevant fields. SDK messages stay inside the adapter. Newly observed devices receive an ADDED event before their activity is delivered.

The internal API is a compatibility dependency, tested against ChirpStack 4.19.2 and Python bindings 4.19.0. It requires device event retention to be enabled. ChirpStack can send retained events when a stream opens, followed by live events. The backend records the local time before opening each device stream. It accepts server receipt timestamps at or after that time minus five seconds and drops older events. This also applies to devices discovered later.

Keep the client and ChirpStack clocks synchronized. The filter uses the Redis event ID’s timestamp. Clock differences can drop new events or admit retained ones. Missed telemetry is not recovered.

API reference

ChirpStackConnection(endpoint, api_key, *, tenant_id=None, application_ids, network_id, poll_interval=30, channel=None) owns its channel, including an injected test channel.

Method or attribute	Behavior
`await tenants()`	Return `{tenant_uuid: name}`. A scoped key may be denied.
`await applications()`	Return `{application_uuid: name}` for the selected tenant, or all accessible tenants when `tenant_id` is `None` (the default).
`await inventory()`	Read descriptors without starting streams or updating the current device list.
`await async_connect()`	Read existing devices and start polling and live events.
`await async_subscribe(*, brands, listener)`	Report matching existing devices, then live events. Return an unsubscribe function. `brands=None` includes all vendors.
`on_disconnect(listener)`	Register a callback with no arguments. Return an unsubscribe function.
`await async_send_downlink(downlink)`	Validate device scope and enqueue application bytes. Return the queue-item UUID.
`await refresh()`	Refresh the connected device list.
`await close()`	Stop tasks, await cleanup, and close the channel.
`devices`	Current devices, keyed by DevEUI.
`available`	Whether the connection can deliver events.
`error`	The last connection failure, or `None`.

The backend retries a failed device stream independently. Three consecutive stream failures or device-list polling failures mark the connection unavailable. A rejected key or access scope fails immediately. Live events continue while a device-list refresh is in progress.

Events delivered before a collection subscribes are not buffered. The five-second leeway does not guarantee initial readings.

brands=None is a backend extension for discovery tools. Device libraries and MockConnection use the Connection protocol with an explicit set of stack and brand pairs.

Callbacks run synchronously. Callbacks registered with on_disconnect() run after available becomes false. Create a new connection to recover. AuthenticationError reports rejected credentials or access scope. ConnectionUnavailable reports other subscription failures. ChirpStack can return the same authentication error for an invalid key and missing permission. Discovery methods retain gRPC errors so callers can offer a tenant-ID fallback.

Write access

Downlinks use the public DeviceService.Enqueue gRPC method. The connection must be connected, and the device must belong to its selected applications. Before enqueueing, the backend reads the device to check its current application. Use a tenant API key with write access for commands. A read-only key still supports device discovery and live events. Attempted writes raise DownlinkError.

Downlink(dev_eui, f_port, data, confirmed=False, expires_at=None) contains plaintext application bytes. FPort must be 1 to 223. An optional expiry is a timezone-aware datetime. ChirpStack owns encryption, frame counters, scheduling, and expiry. The helper never flushes existing commands or retries an ambiguous RPC result.

The backend’s returned queue ID means accepted, a TX_ACK means transmitted, and a positive ACK means the device acknowledged a confirmed downlink. Models interpret device reports to determine whether the requested action took effect.

Device.async_send_downlink() uses this sender and waits for the corresponding ACK by default. The backend method itself returns after enqueueing. See sending commands for caller timeouts.