Passerelle serveur MCP MQTT : laissez les assistants IA lire et piloter vos appareils

Un serveur MCP pour MQTT offre aux assistants IA un moyen structuré et encadré par des permissions de consulter les données des appareils en direct et, si vous l'autorisez, d'envoyer des commandes. Dans ce tutoriel, vous allez créer en Python un serveur Model Context Protocol qui s'abonne à un broker MQTT, met en cache le dernier message de chaque topic et expose un petit ensemble d'outils, puis le connecter à un assistant compatible MCP et interroger vos appareils en langage courant.

Ce que MCP apporte à un système MQTT

Le Model Context Protocol est un protocole ouvert qui connecte les applications d'IA à des outils et données externes. Un serveur MCP déclare des outils (des fonctions avec des paramètres typés et des descriptions) et le client, généralement un assistant de bureau ou un agent intégré à un IDE, permet au modèle de les appeler. C'est le serveur qui décide de ce qui est réellement possible.

MQTT et MCP résolvent des problèmes différents et se combinent très bien :

Le serveur MCP se place entre les deux : d'un côté c'est un client MQTT ordinaire, de l'autre un fournisseur d'outils MCP. Comme MQTT fonctionne en push et MCP en pull, le serveur conserve un cache du dernier message par topic, si bien que les appels d'outils répondent instantanément au lieu d'attendre la prochaine publication.

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

Le serveur ne s'abonne qu'à un seul préfixe, et les commandes ne peuvent être publiées que sur .../devices/<id>/cmd avec une action figurant sur une liste autorisée. Le modèle n'obtient jamais d'accès brut en publication sur des topics arbitraires.

Étape 1 : installer les dépendances

Créez un environnement virtuel et installez le SDK Python officiel de MCP ainsi que 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"

Étape 2 : simuler quelques appareils

Pour avoir quelque chose à interroger, lancez un petit simulateur qui publie toutes les cinq secondes l'état de trois appareils en retained message et accuse réception des commandes. Remplacez 5d81fa par votre propre chaîne aléatoire partout dans ce tutoriel.

# 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)

Étape 3 : écrire le serveur MCP pour MQTT

C'est le cœur du tutoriel. FastMCP, issu du SDK officiel, transforme des fonctions Python typées en outils MCP, en utilisant les docstrings comme descriptions lues par le modèle.

# 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

Trois détails sont importants :

1. N'écrivez jamais sur stdout. Avec le transport stdio, stdout transporte le protocole JSON-RPC. Un print() égaré corrompt le flux et le client se déconnecte avec une erreur d'analyse. Journalisez sur stderr.
2. La boucle MQTT tourne dans son propre thread. loop_start() maintient le cache à jour pendant que le serveur MCP traite les appels d'outils.
3. La validation se fait dans le serveur. La docstring indique au modèle ce qui est autorisé, mais c'est le code qui l'impose, y compris pour les valeurs que le modèle invente.

Étape 4 : tester avec MCP Inspector

Lancez le simulateur dans un terminal, puis exécutez le serveur sous l'inspecteur, qui ouvre une interface web où vous pouvez appeler les outils à la main (Node.js est requis) :

python simulator.py
mcp dev mqtt_mcp_server.py

Appelez list_topics, puis get_latest avec pump-1, puis send_command avec fan-1, set_speed, 70. Vous devriez voir un accusé de réception avec "result": "applied". Pour observer le trafic brut en parallèle :

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

Étape 5 : connecter un assistant IA

Les clients MCP lancent les serveurs stdio à partir d'un fichier de configuration. Pour les assistants de bureau qui utilisent le format courant mcpServers, ajoutez une entrée avec les chemins absolus vers le Python de votre environnement virtuel et vers le script :

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

Redémarrez l'assistant et demandez : « Quel est le niveau du réservoir en ce moment, et la pompe tourne-t-elle ? » Le modèle appelle deux fois get_latest et répond avec les valeurs en direct. Essayez ensuite « Règle fan-1 à 30 pour cent » : vous verrez l'appel d'outil, l'accusé de réception et la réponse. La plupart des clients demandent de confirmer les appels d'outils ; laissez cette confirmation activée pour tout outil qui modifie l'état d'un appareil.

Rédiger des descriptions d'outils que le modèle comprend

Avec FastMCP, le nom de la fonction, les noms des paramètres, les annotations de type et la docstring constituent tout ce que le modèle sait d'un outil. Des descriptions floues produisent un usage flou des outils : traitez-les comme une partie de l'interface.

1. Dites ce que l'outil renvoie, pas seulement ce qu'il fait. « Renvoie le dernier message d'état d'un appareil et son âge en secondes » indique au modèle qu'il peut raisonner sur la fraîcheur des données.
2. Nommez les valeurs valides. Lister les actions et plages autorisées dans la docstring réduit les appels en échec, même si le code continue de les imposer.
3. Utilisez des noms métier, pas des chemins de topic. Le modèle doit demander pump-1, pas construire une chaîne de topic complète. Masquer l'organisation des topics derrière des identifiants d'appareils évite que le modèle devine des chemins et vous permet de réorganiser les topics plus tard.
4. Renvoyez les erreurs sous forme de données. Renvoyer {"error": "unknown device fan-9"} permet au modèle de se corriger ou d'expliquer le problème à l'utilisateur, alors que lever une exception met souvent fin à l'échange sur un échec générique.
5. Gardez une liste d'outils courte. Trois à six outils ciblés fonctionnent mieux qu'une douzaine d'outils qui se chevauchent. S'il vous en faut davantage, répartissez-les entre plusieurs serveurs selon leur finalité.

Il est aussi utile d'inclure l'âge de chaque valeur dans les résultats de lecture. Un modèle qui voit "ageSeconds": 900 signalera généralement que la mesure est périmée, exactement la prudence attendue lorsque l'appareil concerné est peut-être hors ligne.

Durcir le serveur MCP pour la production

Ce tutoriel s'exécute sur le serveur MQTT public gratuit, idéal pour expérimenter mais lisible par tout le monde. Avant de connecter un assistant à de vrais équipements :

Un modèle pratique consiste à utiliser deux serveurs MCP : un serveur en lecture seule que vous pouvez activer librement et un serveur de commande que vous n'activez qu'en cas de besoin. Avec un broker MQTT managé comme ConnectMQ, chaque serveur reçoit depuis le portail ses propres nom d'utilisateur, mot de passe et client ID générés, et vous pouvez observer les deux clients dans l'explorateur de topics en direct pendant vos tests. N'oubliez pas que le broker ne fournit pas de permissions par topic : la séparation entre lecture et commande doit donc être appliquée dans vos serveurs.

Pour les principes de conception de surfaces d'outils sûres, lisez notre guide sur la création d'un agent LLM pour appareils IoT, et consultez le function calling LLM pour l'IoT pour les modèles de validation par schéma. La documentation officielle du Model Context Protocol décrit d'autres transports, comme le streamable HTTP, si vous devez héberger le serveur à distance. Pour une vue d'ensemble, notre guide de référence IA et IoT relie toutes ces briques.

Démarrer avec ConnectMQ

Connectez votre serveur MCP à un broker que vous maîtrisez. ConnectMQ fournit un broker MQTT cloud avec MQTT 5, TLS sur le port 8883 et un client intégré au navigateur pour vos tests. Commencez avec l'offre Serverless gratuite, sans carte bancaire, et donnez en quelques minutes à votre assistant le contexte de vos appareils en direct.

Questions fréquentes

Pourquoi le serveur MCP a-t-il besoin d'un cache ?

MQTT pousse les messages lorsque les appareils publient, tandis que les appels d'outils MCP demandent des données à la demande. Le cache stocke le dernier message de chaque topic pour qu'un appel d'outil puisse répondre immédiatement au lieu d'attendre la prochaine publication.

L'assistant IA peut-il s'abonner directement aux topics ?

Pas au sens d'un flux continu via les outils MCP. Le serveur s'abonne pour le compte de l'assistant et expose des instantanés, ce qui permet aussi de contrôler quels topics sont visibles.

Est-il sûr de laisser un assistant envoyer des commandes MQTT ?

Uniquement au travers d'outils étroits avec validation, limites côté appareil, accusés de réception et confirmation de l'utilisateur. N'exposez jamais d'outil de publication générique à un modèle.

Pourquoi mon client affiche-t-il une erreur d'analyse JSON au démarrage du serveur ?

Quelque chose écrit sur stdout, que le transport stdio réserve aux messages du protocole. Envoyez toute la journalisation et les sorties de débogage vers stderr.