MCP Server MQTT Bridge: Let AI Assistants Read and Control Your Devices

An MCP server for MQTT gives AI assistants a structured, permissioned way to look at live device data and, when you allow it, send commands. In this tutorial you will build a Model Context Protocol server in Python that subscribes to an MQTT broker, caches the latest message per topic and exposes a small set of tools, then connect it to an MCP-capable assistant and ask questions about your devices in plain language.

What MCP Adds to an MQTT System

The Model Context Protocol is an open protocol for connecting AI applications to external tools and data. An MCP server declares tools (functions with typed parameters and descriptions) and the client, typically a desktop assistant or IDE agent, lets the model call them. The server decides what is actually possible.

MQTT and MCP solve different problems and combine well:

The MCP server sits between them: it is an ordinary MQTT client on one side and an MCP tool provider on the other. Because MQTT is push-based and MCP is pull-based, the server keeps a cache of the latest message per topic so tool calls answer instantly instead of waiting for the next publish.

Architecture

AI assistant  <--stdio / MCP-->  mqtt_mcp_server.py  <--MQTT-->  broker  <--MQTT-->  devices
                                  |- background MQTT loop
                                  |- latest-value cache
                                  |- tools: list_topics, get_latest, send_command

The server subscribes to one prefix only, and commands can only be published to .../devices/<id>/cmd with an allowlisted action. The model never gets raw publish access to arbitrary topics.

Step 1: Install Dependencies

Create a virtual environment and install the official MCP Python SDK and paho-mqtt 2.x:

python3 -m venv .venv
source .venv/bin/activate        # on Windows: .venv\Scripts\activate
pip install "mcp[cli]" "paho-mqtt>=2.0"

Step 2: Simulate a Few Devices

To have something to query, run a small simulator that publishes retained state for three devices every five seconds and acknowledges commands. Replace 5d81fa with your own random string everywhere in this tutorial.

# simulator.py
import json
import random
import time
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/5d81fa"
devices = {
    "pump-1": {"running": True, "flow_lpm": 42.0, "pressure_bar": 2.1},
    "tank-1": {"level_pct": 63.0, "temp_c": 21.5},
    "fan-1": {"running": True, "speed_pct": 50},
}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/devices/+/cmd", qos=1)


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    cmd = json.loads(msg.payload)
    state = devices.get(device)
    result = "unknown_device"
    if state is not None:
        if cmd["action"] in ("start", "stop") and "running" in state:
            state["running"] = cmd["action"] == "start"
            result = "applied"
        elif cmd["action"] == "set_speed" and "speed_pct" in state:
            state["speed_pct"] = max(0, min(100, int(cmd["value"])))
            result = "applied"
        else:
            result = "unsupported"
    ack = {"id": cmd.get("id"), "result": result, "ts": int(time.time() * 1000)}
    client.publish(f"{PREFIX}/devices/{device}/ack", json.dumps(ack), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-5d81fa")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    devices["pump-1"]["flow_lpm"] = round(40 + random.uniform(-3, 3), 1) if devices["pump-1"]["running"] else 0.0
    devices["tank-1"]["level_pct"] = round(min(100, max(0, devices["tank-1"]["level_pct"] + random.uniform(-1, 1))), 1)
    for name, state in devices.items():
        payload = {**state, "ts": int(time.time() * 1000)}
        client.publish(f"{PREFIX}/devices/{name}/state", json.dumps(payload), qos=1, retain=True)
    time.sleep(5)

Step 3: Write the MCP Server for MQTT

This is the core of the tutorial. FastMCP from the official SDK turns typed Python functions into MCP tools, using the docstrings as descriptions the model reads.

# mqtt_mcp_server.py
import json
import logging
import sys
import threading
import time
import uuid

import paho.mqtt.client as mqtt
from mcp.server.fastmcp import FastMCP

# stdio transport uses stdout for protocol messages: log to stderr only
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
log = logging.getLogger("mqtt-mcp")

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/5d81fa"
ALLOWED_ACTIONS = {"start", "stop", "set_speed"}

cache: dict[str, dict] = {}
acks: dict[str, dict] = {}
lock = threading.Lock()


def on_connect(client, userdata, flags, reason_code, properties):
    log.info("MQTT connected: %s", reason_code)
    client.subscribe(f"{PREFIX}/#", qos=1)


def on_message(client, userdata, msg):
    text = msg.payload.decode("utf-8", errors="replace")
    with lock:
        cache[msg.topic] = {"payload": text, "retained": msg.retain, "receivedAt": time.time()}
        if msg.topic.endswith("/ack"):
            try:
                ack = json.loads(text)
                if ack.get("id"):
                    acks[ack["id"]] = ack
            except json.JSONDecodeError:
                pass


mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="mcp-bridge-5d81fa")
mqttc.on_connect = on_connect
mqttc.on_message = on_message
mqttc.connect(BROKER, PORT, keepalive=60)
mqttc.loop_start()

mcp = FastMCP("mqtt-devices")


@mcp.tool()
def list_topics(contains: str = "") -> list[str]:
    """List MQTT topics that have received data, optionally filtered by a substring."""
    with lock:
        return sorted(t for t in cache if contains in t)


@mcp.tool()
def get_latest(device_id: str) -> dict:
    """Return the latest state message of a device (for example pump-1) and its age in seconds."""
    topic = f"{PREFIX}/devices/{device_id}/state"
    with lock:
        entry = cache.get(topic)
    if entry is None:
        return {"error": f"no data for {device_id}"}
    try:
        data = json.loads(entry["payload"])
    except json.JSONDecodeError:
        data = entry["payload"]
    return {"device": device_id, "state": data,
            "ageSeconds": round(time.time() - entry["receivedAt"], 1)}


@mcp.tool()
def send_command(device_id: str, action: str, value: float | None = None) -> dict:
    """Send a command to a device. action must be start, stop or set_speed (value 0-100).
    Waits up to 5 seconds for the device acknowledgement."""
    if action not in ALLOWED_ACTIONS:
        return {"error": f"action must be one of {sorted(ALLOWED_ACTIONS)}"}
    if action == "set_speed" and (value is None or not 0 <= value <= 100):
        return {"error": "set_speed requires a value between 0 and 100"}
    with lock:
        known = f"{PREFIX}/devices/{device_id}/state" in cache
    if not known:
        return {"error": f"unknown device {device_id}"}
    cmd_id = str(uuid.uuid4())
    cmd = {"id": cmd_id, "action": action, "value": value, "source": "mcp"}
    mqttc.publish(f"{PREFIX}/devices/{device_id}/cmd", json.dumps(cmd), qos=1)
    deadline = time.time() + 5
    while time.time() < deadline:
        with lock:
            if cmd_id in acks:
                return {"sent": cmd, "ack": acks.pop(cmd_id)}
        time.sleep(0.1)
    return {"sent": cmd, "ack": None, "note": "no acknowledgement within 5 seconds"}


if __name__ == "__main__":
    mcp.run()  # stdio transport by default

Three details matter here:

1. Never print to stdout. With the stdio transport, stdout carries the JSON-RPC protocol. A stray print() corrupts the stream and the client disconnects with a parse error. Log to stderr.
2. The MQTT loop runs in its own thread. loop_start() keeps the cache fresh while the MCP server handles tool calls.
3. Validation lives in the server. The docstring tells the model what is allowed, but the code enforces it, including for values the model invents.

Step 4: Test With the MCP Inspector

Start the simulator in one terminal, then run the server under the inspector, which opens a web UI where you can call tools by hand (it requires Node.js):

python simulator.py
mcp dev mqtt_mcp_server.py

Call list_topics, then get_latest with pump-1, then send_command with fan-1, set_speed, 70. You should see an acknowledgement with "result": "applied". To watch the raw traffic at the same time:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-5d81fa -t "connectmq/demo/5d81fa/#" -v

Step 5: Connect an AI Assistant

MCP clients launch stdio servers from a configuration file. For desktop assistants that use the common mcpServers format, add an entry with absolute paths to your virtual environment's Python and the script:

{
  "mcpServers": {
    "mqtt-devices": {
      "command": "/home/you/mqtt-mcp/.venv/bin/python",
      "args": ["/home/you/mqtt-mcp/mqtt_mcp_server.py"]
    }
  }
}

Restart the assistant and ask: "What is the tank level right now, and is the pump running?" The model calls get_latest twice and answers with live values. Then try "Set fan-1 to 30 percent": you will see the tool call, the acknowledgement and the answer. Most clients ask you to confirm tool calls; keep that confirmation on for any tool that changes device state.

Writing Tool Descriptions the Model Understands

With FastMCP, the function name, parameter names, type hints and docstring are everything the model knows about a tool. Vague descriptions produce vague tool use, so treat them as part of the interface:

1. Say what the tool returns, not only what it does. "Return the latest state message of a device and its age in seconds" tells the model it can reason about freshness.
2. Name the valid values. Listing the allowed actions and ranges in the docstring reduces failed calls, even though the code still enforces them.
3. Use domain names, not topic paths. The model should ask for pump-1, not construct a full topic string. Hiding the topic layout behind device IDs keeps the model from guessing paths and lets you reorganize topics later.
4. Return errors as data. Returning {"error": "unknown device fan-9"} lets the model correct itself or explain the problem to the user, while raising an exception often ends the exchange with a generic failure.
5. Keep the tool list short. Three to six focused tools work better than a dozen overlapping ones. If you need more, split them across servers by purpose.

It also helps to include the age of every value in read results. A model that sees "ageSeconds": 900 will usually mention that the reading is stale, which is exactly the caution you want when the device behind it might be offline.

Hardening the MCP Server for Production

The tutorial runs on the free public MQTT broker, which is ideal for experiments but readable by anyone. Before pointing an assistant at real equipment:

A practical pattern is two MCP servers: a read-only one you can enable freely and a command server you enable only when needed. In a managed MQTT broker such as ConnectMQ, each server gets its own generated username, password and client ID from the portal, and you can watch both clients in the live topic explorer while you test. Remember that the broker does not provide per-topic permissions, so the split between reading and commanding has to be enforced in your servers.

For the design principles behind safe tool surfaces, read our guide to building an LLM agent for IoT devices, and see LLM function calling for IoT for schema validation patterns. The official Model Context Protocol documentation describes other transports such as streamable HTTP if you need to host the server remotely. For the bigger picture, our AI and IoT pillar guide connects these pieces.

Get Started with ConnectMQ

Point your MCP server at a broker you control. ConnectMQ provides a cloud MQTT broker with MQTT 5, TLS on 8883 and an in-browser client for testing. Start on the free Serverless plan, no credit card required, and give your assistant live device context in minutes.

Frequently Asked Questions

Why does the MCP server need a cache?

MQTT pushes messages when devices publish, while MCP tool calls ask for data on demand. The cache stores the latest message per topic so a tool call can answer immediately instead of waiting for the next publish.

Can the AI assistant subscribe to topics directly?

Not through MCP tools in a streaming sense. The server subscribes on the assistant's behalf and exposes snapshots, which also keeps control over which topics are visible.

Is it safe to let an assistant send MQTT commands?

Only through narrow tools with validation, device-side limits, acknowledgements and user confirmation. Never expose a generic publish tool to a model.

Why does my client show a JSON parse error when starting the server?

Something is writing to stdout, which the stdio transport reserves for protocol messages. Send all logging and debug output to stderr.