MQTT Push API

The MQTT Push API is a subscribe-only MQTT 5 service. Your client connects with your v5 API token, subscribes to an organization, a project, or one device, and receives one message each time a device uploads. Each message holds the same rows as GET Device Data. Each reading has a reading_id that counts up by one for each device, so your client can tell when it missed readings and fetch them from GET Device Data with since_reading_id. GET Device Data stays the complete record: the push service does not replay readings your client missed.

1. Before you start

You need:

• Your API token. In ZENTRA Cloud, select your name at the top right, then Integrations, and select Show Token.
• The User, Editor, or Administrator role in the organization, or in the project the devices are shared into. A Viewer cannot receive readings.
• An MQTT 5 client library. The examples below use Python 3.10 or later and paho-mqtt 2.1.

The push service does not send hourly and daily rows (such as Hourly Reference ET). Read them from GET Device Data with aggregate=hourly or aggregate=daily.

2. Endpoints and topics

Subscribe to one of these topics:

• zc/v1/orgs/{organization_id}/devices/+/readings gives every device of an organization, including devices added later.
• zc/v1/orgs/{organization_id}/projects/{project_id}/devices/+/readings gives the devices of a project, if you are a member of the project but not of its organization.
• Replace + with a device_id in either topic to get one device.

One topic for each organization or project is enough, however many devices it has. The push service refuses #, a topic that covers every organization, and any topic outside your access. Your client cannot publish.

3. Connection settings

4. Message format

Each message is one device upload, as JSON:

{
            "device_id": "z6-01455",
            "organization_id": "{organization_id}",
            "values": [
              {
                "port_num": 6,
                "measurement": "Water Content",
                "unit": "m³/m³",
                "sensor_name": "TEROS 54",
                "position": -15.0,
                "value": 0.338,
                "timestamp": 1788394500,
                "datetime": "2026-09-02T17:15:00-07:00",
                "error_code": 0,
                "reading_id": 89025
              }
              // ... one row for each other measurement of the reading, and of each other reading in the upload
            ]
          }
          

• Values are always in metric units. There is no units option.
• A large catch-up upload (for example, after a logger was offline for days) arrives as several messages. Each holds whole readings and is at most 60,000 bytes.
• A reading can arrive more than once, and readings can arrive out of order. Store each reading once, keyed on device_id and reading_id, and order readings on timestamp.
• A row from a very old reading can have reading_id set to null. GET Device Data does not return these rows when you page by since_reading_id. The example below skips them.
• Corrected or reprocessed history is not pushed. Read it from GET Device Data.

5. Check your setup

1. Save your API token in a file that only you can read (Linux and macOS shown; on Windows, save it as .zentra\token in your user folder):

   mkdir -p ~/.zentra
             nano ~/.zentra/token
             chmod 600 ~/.zentra/token
             

2. Install the MQTT library: pip install "paho-mqtt>=2.1,<3".
3. Save this script as check_push.py and run it. It gets your connection settings from GET /v5/mqtt/connection, checks that the push endpoint is reachable, and lists the topics you can subscribe to.

import json, os, socket, ssl, urllib.request
          
          token = open(os.path.expanduser("~/.zentra/token")).read().strip()
          request = urllib.request.Request("https://api.zentracloud.io/v5/mqtt/connection", headers={"X-API-Key": token})
          connection = json.load(urllib.request.urlopen(request, timeout=30))
          print("MQTT username:", connection["username"])
          print("Client id:", connection["client_id_prefix"] + "{a name you choose}")
          
          for endpoint in connection["endpoints"]:
              with socket.create_connection((endpoint["host"], endpoint["port"]), timeout=10) as raw:
                  with ssl.create_default_context().wrap_socket(raw, server_hostname=endpoint["host"]) as tls:
                      print(f"Push endpoint: mqtts://{endpoint['host']}:{endpoint['port']} is reachable,", tls.version())
          
          print("Topics you can subscribe to:")
          for topic in connection["topic_filters"]:
              print(" ", topic)
          

If the script stops with HTTP Error 401, the token in your file is wrong or was replaced. If no topics are listed, you do not have the User, Editor, or Administrator role in any organization or project.

6. Example: receive readings in Python

This example prints each reading as it arrives. When a device’s reading_id skips, it fetches the missing readings from GET Device Data. It keeps its state in memory, so use it to test your setup, not as your production client.

Save it as zentra_push.py and run it:

python zentra_push.py                                                         # every topic you can subscribe to
          python zentra_push.py zc/v1/orgs/ORGANIZATION_ID/devices/DEVICE_ID/readings   # only the topics you name, here one device

import json, os, ssl, sys, urllib.request
          
          import paho.mqtt.client as mqtt
          from paho.mqtt.packettypes import PacketTypes
          from paho.mqtt.properties import Properties
          
          API_URL = "https://api.zentracloud.io/v5"
          TOKEN = open(os.path.expanduser("~/.zentra/token")).read().strip()
          last_reading = {}
          
          def api_get(path):
              request = urllib.request.Request(API_URL + path, headers={"X-API-Key": TOKEN})
              with urllib.request.urlopen(request, timeout=60) as response:
                  return json.load(response)
          
          def show(device_id, rows):
              readings = {}
              for row in rows:
                  if row["reading_id"] is not None and row["reading_id"] > last_reading.get(device_id, -1):
                      readings.setdefault(row["reading_id"], []).append(row)
              for reading_id, values in sorted(readings.items()):
                  print(device_id, reading_id, values[0]["datetime"],
                        ", ".join(f'{v["measurement"]} {v["value"]} {v["unit"]}' for v in values))
                  last_reading[device_id] = reading_id
          
          def fill(device_id):
              cursor = last_reading[device_id]
              while cursor is not None:
                  page = api_get(f"/devices/{device_id}/data?since_reading_id={cursor}")
                  show(device_id, page["values"])
                  cursor = page["pagination"]["next_reading_id"]
          
          def on_connect(client, userdata, flags, reason_code, properties):
              print("connect:", reason_code, "(session kept)" if flags.session_present else "(new session)")
              if not reason_code.is_failure:
                  client.subscribe([(topic, mqtt.SubscribeOptions(qos=1, retainHandling=1)) for topic in topics])
          
          def on_subscribe(client, userdata, mid, reason_codes, properties):
              for topic, code in zip(topics, reason_codes):
                  print("subscribe:", topic, code)
          
          def on_message(client, userdata, message):
              payload = json.loads(message.payload)
              device_id = payload["device_id"]
              reading_ids = [row["reading_id"] for row in payload["values"] if row["reading_id"] is not None]
              if device_id in last_reading and reading_ids and min(reading_ids) > last_reading[device_id] + 1:
                  try:
                      fill(device_id)
                  except Exception as error:
                      print(f"could not fetch the readings {device_id} skipped: {error}")
                      return
              show(device_id, payload["values"])
          
          connection = api_get("/mqtt/connection")
          topics = sys.argv[1:] or connection["topic_filters"]
          if not topics:
              sys.exit("no topics: you need the User role or higher in an organization or project")
          
          client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=connection["client_id_prefix"] + "test-1",
                               protocol=mqtt.MQTTv5)
          client.username_pw_set(connection["username"], TOKEN)
          client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
          client.tls_set_context(ssl.create_default_context())
          client.reconnect_delay_set(min_delay=1, max_delay=60)
          properties = Properties(PacketTypes.CONNECT)
          properties.SessionExpiryInterval = 3600
          endpoint = connection["endpoints"][0]
          client.connect_async(endpoint["host"], endpoint["port"], keepalive=30, properties=properties)
          client.loop_forever(retry_first_connection=True)

On the first run, the client starts with the latest reading of each device that uploaded in the last day. To load older readings, call GET Device Data with since_reading_id=0.

7. Connection behavior

Your client is disconnected from time to time. Each case below is expected. A client with the settings above reconnects by itself and subscribes again in its connect handler.

• Push service updates. During the beta, we update the push service while you are connected. An update can disconnect your client. A client that reconnects at once is usually back in 1 to 2 seconds and keeps its session, so the readings sent meanwhile still arrive.
• Membership changes. When you are added to or removed from an organization or project, or your role changes, all your clients are disconnected and their sessions end. Each client reconnects at once with a new session.
• New API token. When you regenerate your API token, your clients are disconnected and the old token is refused. Put the new token in your token file and restart the example.
• Scheduled reconnects. From time to time, the push service closes a connection to check your access again. Your client reconnects at once and keeps its session.
• Unexpected outages. If part of the push service fails, your client can lose its connection without notice. It reconnects when its keep alive detects the lost connection, and it may get a new session.
• Your client is offline. The push service can keep your session for the session expiry you set (one hour in the example). After that, or if the session is lost, your client gets a new session.

After a reconnect, check the Session Present flag, not the disconnect reason code. If the session is new, the readings sent while your client was disconnected are not delivered: fill the gap from GET Device Data, as the example does.

Limits:

• Each user can have only a limited number of connections open at once. A connection over the limit is refused with 0x87 Not authorized.
• A client id or an IP address that connects too often is refused for a short time. Use a growing delay between attempts, as the example does.
• Two running clients with the same client id disconnect each other again and again. Give each copy its own name.

8. Troubleshooting

The push service reports a problem with an MQTT 5 reason code.

Some client libraries, including paho-mqtt 2.1, report every disconnect from the push service as reason code 0 (Normal disconnection). Do not depend on the disconnect reason code: reconnect and subscribe again in every case.