MQTT Request Response Pattern with MQTT 5

The MQTT request response pattern lets one client ask a question and get a specific answer back, even though MQTT itself is a publish/subscribe protocol. MQTT 5 added two properties, response topic and correlation data, that make this clean and interoperable. In this guide you will build a working RPC-style call in Python and Node.js, test it on a public broker, and learn how to handle timeouts and failures.

Why Request/Response Over MQTT

Most MQTT traffic is one-way: sensors publish, dashboards subscribe. But real systems also need questions with answers:

"What firmware version are you running?"
"Read register 40001 and tell me the value."
"Did the valve actually move?"
"Run a self-test and report the result."

You could open an HTTP endpoint on every device, but devices often sit behind NAT or cellular networks where inbound connections are impossible. Since the device already holds an outbound MQTT connection, routing the request and the reply through the broker is simpler and keeps one security model. The comparison of MQTT and HTTP goes deeper on why that outbound-only model matters.

How the MQTT Request Response Pattern Works in MQTT 5

The pattern uses two PUBLISH properties defined in MQTT 5:

Property	Set by	Purpose
Response Topic	Requester	Topic the responder should publish the answer to
Correlation Data	Requester	Opaque bytes the responder copies back unchanged

The flow is:

The requester subscribes to its own reply topic, for example connectmq/demo/k3p8/rpc/replies/<requester-id>.
It publishes the request to the responder's request topic with Response Topic set to that reply topic and Correlation Data set to a unique ID.
The responder receives the request, does the work, and publishes the result to the Response Topic, copying the Correlation Data.
The requester matches the incoming Correlation Data to the pending request and completes it.

The broker does not track requests at all. It just forwards two ordinary messages. Everything that makes it "RPC" lives in the clients, which is why the pattern scales well and works on any MQTT 5 broker.

Topic Design for Request/Response

Good topics make the pattern easy to secure and debug:

Requests: <prefix>/rpc/<device-id>/req so each device subscribes only to its own requests.
Replies: <prefix>/rpc/replies/<requester-id> so each caller gets only its own answers.
Never use a shared reply topic for many callers. Every caller would receive every reply and have to discard most of them.

If you want a broader view of how to structure names, see MQTT topic naming conventions.

Python Responder

This responder plays the role of a device. It listens for requests and echoes back a result using the response topic and correlation data from the request. It uses paho-mqtt 2.x with MQTT 5 on the public MQTT test broker:

import json
import uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

BROKER = "broker.connectmq.com"
REQ_TOPIC = "connectmq/demo/k3p8/rpc/pump-07/req"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REQ_TOPIC, qos=1)
    print("responder ready")

def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    correlation = getattr(props, "CorrelationData", None)
    if not response_topic:
        print("request without response topic, ignoring")
        return

    request = json.loads(msg.payload)
    if request.get("method") == "getStatus":
        result = {"ok": True, "rpm": 1480, "running": True}
    else:
        result = {"ok": False, "error": "unknown method"}

    reply_props = Properties(PacketTypes.PUBLISH)
    if correlation is not None:
        reply_props.CorrelationData = correlation
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"responder-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()

Node.js Requester with Timeouts

The requester keeps a map of pending calls keyed by correlation ID and rejects any call that does not get an answer within the timeout. It uses MQTT.js v5 over TLS:

const mqtt = require("mqtt");
const { randomUUID } = require("crypto");

const clientId = "requester-" + randomUUID().slice(0, 8);
const replyTopic = `connectmq/demo/k3p8/rpc/replies/${clientId}`;
const pending = new Map();

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
});

function call(deviceId, method, params = {}, timeoutMs = 5000) {
  const id = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(id);
      reject(new Error(`timeout calling ${method} on ${deviceId}`));
    }, timeoutMs);
    pending.set(id, { resolve, timer });
    client.publish(
      `connectmq/demo/k3p8/rpc/${deviceId}/req`,
      JSON.stringify({ method, params }),
      {
        qos: 1,
        properties: {
          responseTopic: replyTopic,
          correlationData: Buffer.from(id),
          messageExpiryInterval: Math.ceil(timeoutMs / 1000),
        },
      }
    );
  });
}

client.on("message", (topic, payload, packet) => {
  const corr = packet.properties && packet.properties.correlationData;
  if (!corr) return;
  const id = corr.toString();
  const entry = pending.get(id);
  if (!entry) return; // late or unknown reply
  clearTimeout(entry.timer);
  pending.delete(id);
  entry.resolve(JSON.parse(payload.toString()));
});

client.on("connect", () => {
  client.subscribe(replyTopic, { qos: 1 }, async (err) => {
    if (err) return console.error(err);
    try {
      const status = await call("pump-07", "getStatus");
      console.log("reply:", status);
    } catch (e) {
      console.error(e.message);
    } finally {
      client.end();
    }
  });
});

Start the Python responder, then run the Node.js requester. You should see reply: { ok: true, rpm: 1480, running: true }. Stop the responder and run the requester again to see the timeout path.

Notice the request also carries a message expiry equal to the timeout. If the device is offline with a persistent session, the request is discarded once the caller has given up, instead of executing minutes later. The details are in our article on the MQTT message expiry interval.

Request/Response Design Checklist

Subscribe to the reply topic before sending the first request, or you can miss a fast reply.
Use a unique correlation ID per request, such as a UUID.
Always implement a timeout and clean up pending entries.
Set message expiry on requests so stale calls never execute.
Return structured errors (ok: false, error code) rather than staying silent.
Make actions idempotent where possible, because QoS 1 can deliver a request twice. See MQTT duplicate messages.
Do not retain requests or replies.

Designing the Reply Payload

The transport is only half of the contract; the reply format matters just as much once several teams call the same devices. A small, consistent envelope keeps callers simple:

{"ok": false, "code": "BUSY", "message": "self-test already running", "retryAfter": 10}

Some conventions that hold up well in practice:

Always reply, even on failure. A silent responder is indistinguishable from an offline one, and the caller can only time out.
Use machine-readable codes such as BUSY, INVALID_PARAMS or UNKNOWN_METHOD, plus a human-readable message for logs.
Version your methods. getStatus.v2 or a version field lets you change the reply shape without breaking older callers.
Separate acceptance from completion. For anything that takes more than a second or two, reply immediately with {"ok": true, "accepted": true, "jobId": "..."} and publish progress and the final result to a job topic. The caller's timeout then only covers acceptance.
Mark the content type. Setting the MQTT 5 Content Type property to application/json on requests and replies helps generic tools display them correctly.

Working with MQTT 3.1.1 Clients

If some devices only speak MQTT 3.1.1, the properties are unavailable. The common workaround is to put the same information into the payload, for example {"replyTo": "...", "corrId": "...", "method": "getStatus"}. It works, but every party must agree on the field names, whereas MQTT 5 properties are standard and keep the payload clean. Mixed fleets often run both until older firmware is retired.

Troubleshooting

No reply arrives. Check that the responder actually sees the Response Topic. If either side connects with MQTT 3.1.1, properties are dropped. Also confirm the requester is subscribed to the exact reply topic.

Replies go to the wrong caller. Your reply topic is probably shared. Include the requester's client ID in it.

Correlation data does not match. It is binary. Compare bytes, or decode both sides consistently, as the examples do with UTF-8 strings.

Calls hang under load. A single-threaded responder handles one request at a time. Long operations should reply with an "accepted" status and publish the final result later.

Running RPC on ConnectMQ

ConnectMQ supports MQTT 5.0 response topic, correlation data and message expiry, so the code above runs unchanged against your own broker. Swap the host for your broker and add username: "YOUR_USERNAME", password: "YOUR_PASSWORD" and clientId: "YOUR_CLIENT_ID" from the portal. A hosted MQTT broker removes the operational side so you can focus on the request contract. For the bigger picture, start with the MQTT tutorial.

Get Started with ConnectMQ

Create a free broker at app.connectmq.com on the Serverless plan, no credit card required. Run the responder on a device or a laptop, call it from a script, and watch requests and replies flow in the ConnectMQ portal's live topic explorer.

Frequently Asked Questions

Does the broker match requests to replies?

No. The broker forwards two independent messages. The requester matches them using the correlation data it chose.

Is correlation data required?

Not strictly, but without it a client with several outstanding requests cannot tell which reply belongs to which call. Always set it.

Can I use QoS 0 for request/response?

You can, but lost requests then only show up as timeouts. QoS 1 with idempotent handlers is the usual choice.

Is MQTT request/response a replacement for HTTP APIs?

For device commands and queries behind NAT it is often a better fit. For large file transfers or public APIs, HTTP is usually simpler.

MQTT Request/Response Pattern with Response Topic and Correlation Data