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.2.2 · 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

Install 1.2.2 or later. 1.0.0 on PyPI was built from an older copy of this code and lacks the fixes listed in the changelog. If 1.2.2 is not on PyPI yet, install from source:

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 and the batch is dropped with reason rejected.

Execution status (executionStatus)

Events can carry an execution status. OUTCOME_BASED rate plans bill each event at the weight set for its status; events without one bill at full price.

The client wrappers don't set one: they emit each event before the broker answers, so the SDK has no outcome to report. Send your own with push():

billing.push(customer_id="cust_acme_001", topic="devices/001/status", qos=1, retained=False,
             event_type="PUBLISH", client_id="device-001", execution_status="FAILED")

The SDK trims and upper-cases the value and leaves it off the event when it's blank. Accepted values: SUCCESS, PARTIAL, TIMEOUT, ERROR, VALIDATION_FAILED, FAILED, FAILURE, CANCELLED, PENDING, BLOCKED, HITL_REQUIRED — any other value is logged as a warning and left off the event, since the ingestor would reject the event.

Dropped events

An event that will never reach Aforo is counted in billing.dropped_count, logged at WARNING, and passed to the opt-in on_drop(events, reason) hook. The events keep their idempotency keys, so sending them again later cannot double-bill.

def on_drop(events, reason):
    dead_letter.write(reason, events)

billing = AforoMqttBilling(..., on_drop=on_drop)

When a 202 response reports refused events without a usable index, they are counted in dropped_count but not passed to the hook, since the SDK cannot tell which events they were. Exceptions raised by the hook are swallowed. An unknown executionStatus is not a drop: the field is left off and the event is still sent.

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.