aforo-mqtt-metering

Meter MQTT client traffic — PUBLISH, SUBSCRIBE, UNSUBSCRIBE, CONNECT, DISCONNECT — by wrapping a paho-mqtt (sync) or aiomqtt (async) client. Use it when you connect to a third-party broker (AWS IoT, HiveMQ Cloud, EMQ X Cloud) and want to meter what your client sends and receives.

Version: 1.0.0 · Apache-2.0 · Changelog · User guide

For broker-side (server-level) metering, use the companion EMQ X Erlang plugin in aforo-nextgen-docker/emqx-plugin-aforo-metering/. This Python SDK is client-side.

Install

Intended public install:

pip install aforo-mqtt-metering              # core
pip install "aforo-mqtt-metering[paho]"      # paho-mqtt (sync)
pip install "aforo-mqtt-metering[aiomqtt]"   # aiomqtt (async)
pip install "aforo-mqtt-metering[httpx]"     # faster HTTP flush than stdlib urllib

Not yet on PyPI — install from source for now:

git clone https://github.com/aforoai/SDKs.git
cd SDKs/aforo-metering-sdks/python-mqtt     # folder holding setup.py
pip install -e .
pip install -e ".[paho]"               # or [aiomqtt] / [httpx]

The core package has no required dependencies — the MQTT client libraries and HTTP client are optional extras.

Quickstart — paho-mqtt (sync)

Best when you publish/subscribe from a device or service against a managed broker and want per-event billing without rewriting your MQTT code.

import os
import paho.mqtt.client as mqtt
from aforo_mqtt_metering import AforoMqttBilling, wrap_paho_client

billing = AforoMqttBilling(
    tenant_id="tenant_acme",
    product_id="prod_mqtt_iot_telemetry",
    api_key=os.environ["AFORO_API_KEY"],
    ingestor_url="https://api.aforo.ai",
)

client = mqtt.Client(client_id="device-001")
wrap_paho_client(billing, client, customer_id="cust_acme_001")

client.tls_set()
client.connect("broker.example.com", 8883)
client.subscribe("sensors/+/temperature")
client.publish("devices/001/status", '{"online": true}')
client.loop_forever()

Quickstart — aiomqtt (async)

import aiomqtt
from aforo_mqtt_metering import AforoMqttBilling, wrap_aiomqtt_client

billing = AforoMqttBilling(
    tenant_id="tenant_acme",
    product_id="prod_mqtt_iot_telemetry",
    api_key=os.environ["AFORO_API_KEY"],
    ingestor_url="https://api.aforo.ai",
)

async def main():
    async with aiomqtt.Client("broker.example.com", port=8883, tls_context=ssl_ctx) as c:
        wrap_aiomqtt_client(billing, c, customer_id="cust_acme_001", client_id="device-001")
        await c.subscribe("sensors/+/temperature")
        await c.publish("devices/001/status", '{"online": true}')
        async for msg in c.messages:
            print(msg.topic, msg.payload)

Each metered event POSTs to https://api.aforo.ai/v1/ingest/batch with X-API-Key: <api_key> and X-Tenant-Id: <tenant_id>, carrying mqttEventType, mqttTopic, mqttQos, mqttRetained, mqttClientId, and dataBytes.

⚠ Events are sent to the ingestor's /v1/ingest/batch path as {"events": [...]}, at most 1000 events per request (larger buffers are split). Set ingestor_url to the host only — the SDK appends the path.

customer_id is passed in when you wrap the client — supply it from your trusted device/account mapping, not from anything the broker peer controls. tenant_id is fixed from config and sent as a header.

Configuration

Constructor arguments for AforoMqttBilling(...):

Event metric names follow mqtt_broker.<event_type lowercased> (e.g. mqtt_broker.publish, mqtt_broker.subscribe). Retry is fixed at 3 attempts (1s / 2s backoff between them); 408 and 5xx are retried, 429 waits for Retry-After (capped at 60 s), and any other 4xx is not retried.

Walk me through it

Install → wrap the client → publish/subscribe → confirm the event in Aforo, step by step, is in USER_GUIDE.md.

What this doesn't cover

Inbound message delivery is not billed unless you set emit_deliver_events=True — it's off by default to keep volume down. This is client-side metering only; for broker-wide accounting (every client on the broker) use the EMQ X plugin. It doesn't enforce QoS limits or topic ACLs. Pricing, and QoS/retained tiering via filter conditions, are configured in the Aforo console.