Building a Real-Time MQTT Pub/Sub Pipeline with Python and CloudAMQP

Every IoT project eventually needs the same thing: a lightweight, reliable way to move messages between devices and code. MQTT has become the de facto standard for this — it’s small, fast, and works equally well for a single sensor publishing readings as it does for a fleet of devices.

In this walkthrough, we’ll build a minimal but production-shaped MQTT publisher and consumer in Python, using paho-mqtt and a free-tier CloudAMQP broker (their managed service exposes both AMQP and MQTT endpoints). Along the way, we'll hit — and fix — the handful of gotchas that trip up almost everyone doing this for the first time on macOS.

Why MQTT

MQTT is a publish/subscribe protocol built for constrained networks and devices. Instead of services calling each other directly, publishers send messages to a topic, and any number of subscribers listening on that topic receive them. Neither side needs to know the other exists — the broker handles routing.

This makes it a natural fit for sensor data, telemetry, and lightweight device-to-cloud messaging, where you want producers and consumers to stay decoupled.

Tools used

• Python 3.10
• paho-mqtt (2.x) — the standard Python MQTT client library
• CloudAMQP — managed broker offering both AMQP and MQTT endpoints, TLS included
• certifi — trusted CA bundle, needed to fix a common macOS TLS issue

pip install paho-mqtt certifi

python3 -m pip show paho-mqtt

Step 1: Get broker credentials on CloudAMQP

Spin up a free CloudAMQP instance and open its dashboard. Under MQTT details you’ll find everything you need:

Note: First create team :for ex:IOT_temperature — then click on the create new instance

1.click on the create instance here it is the topic name

Name as ex:temperature_server as instance

name ==Topic name

2. choose plan

3. Check the configure

4. finally we created the instance , now click on the instance we created here for ex: temperature-server

5. We can see the deatils of the created for the instance

6. Coding part

Get Ramanamuttana’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

producer.py

from paho.mqtt import client as mqtt
import ssl, certifi
import time
import json
import random

clientId = "Youtube-Producer"
hostname = "swallow.lmq.cloudamqp.com"
port = 8883
username = "bunjvgtn:bunjvgtn"
password = "JPQTw0oweROKmMdRrpP18GUXoqHyXD0I"
topic = "sensors/temperature-server"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, clientId)
client.username_pw_set(username, password)
client.tls_set(ca_certs=certifi.where())
client.connect(hostname, port)

client.loop_start()
time.sleep(1)  # let the connection fully establish

for i in range(1, 1001):
    temperature = round(random.uniform(15.0, 35.0), 2)
    payload = json.dumps({
        "id": i,
        "value": temperature,
        "unit": "celsius"
    })

    client.publish(topic, payload)
    print(f"[{i}/1000] Published: {payload}")

    time.sleep(1)

print("Done sending 1000 messages")

time.sleep(1)
client.loop_stop()
client.disconnect()

below code is Consumer.py

from paho.mqtt import client as mqtt
import certifi
import json
from datetime import datetime

clientId = "Youtube-Consumer"
hostname = "swallow.lmq.cloudamqp.com"
port = 8883
username = "bunjvgtn:bunjvgtn"
password = "JPQTw0oweROKmMdRrpP18GUXoqHyXD0I"
topic = "sensors/temperature-server"

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        print("Connected successfully")
        client.subscribe(topic)
        print(f"Subscribed to: {topic}")
        print("Waiting for messages...\n")
    else:
        print(f"Failed to connect, reason code: {reason_code}")

def on_message(client, userdata, msg):
    timestamp = datetime.now().strftime("%H:%M:%S")
    payload = msg.payload.decode()

    print(f"[{timestamp}] Topic: {msg.topic}")
    try:
        data = json.loads(payload)
        print(f"           Data: {data}")
    except json.JSONDecodeError:
        print(f"           Raw: {payload}")
    print("-" * 40)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, clientId)
client.username_pw_set(username, password)
client.tls_set(ca_certs=certifi.where())

client.on_connect = on_connect
client.on_message = on_message

client.connect(hostname, port)
print("Connecting to broker.....")
client.loop_forever()

first run the consumer ready to accept the messages then producer

7. check on the Lavinmq browser

we can see the results on consumer console

sample data