Puente servidor MCP MQTT: deja que los asistentes de IA lean y controlen tus dispositivos

Un servidor MCP para MQTT da a los asistentes de IA una forma estructurada y con permisos de consultar datos de dispositivos en vivo y, cuando tú lo permites, enviar comandos. En este tutorial construirás en Python un servidor Model Context Protocol que se suscribe a un broker MQTT, guarda en caché el último mensaje de cada topic y expone un pequeño conjunto de herramientas; después lo conectarás a un asistente compatible con MCP y le harás preguntas sobre tus dispositivos en lenguaje natural.

Qué aporta MCP a un sistema MQTT

Model Context Protocol es un protocolo abierto para conectar aplicaciones de IA con herramientas y datos externos. Un servidor MCP declara herramientas (funciones con parámetros tipados y descripciones) y el cliente, normalmente un asistente de escritorio o un agente de IDE, permite que el modelo las llame. Es el servidor quien decide qué es realmente posible.

MQTT y MCP resuelven problemas distintos y se combinan bien:

El servidor MCP se sitúa entre ambos: por un lado es un cliente MQTT corriente y por el otro un proveedor de herramientas MCP. Como MQTT se basa en push y MCP en pull, el servidor mantiene una caché con el último mensaje de cada topic para que las llamadas a herramientas respondan al instante en lugar de esperar a la siguiente publicación.

Arquitectura

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

El servidor se suscribe a un único prefijo, y los comandos solo pueden publicarse en .../devices/<id>/cmd con una acción incluida en la lista permitida. El modelo nunca obtiene acceso de publicación directo a topics arbitrarios.

Paso 1: instalar las dependencias

Crea un entorno virtual e instala el SDK oficial de MCP para Python y paho-mqtt 2.x:

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

Paso 2: simular algunos dispositivos

Para tener algo que consultar, ejecuta un pequeño simulador que publica cada cinco segundos el estado retenido de tres dispositivos y confirma los comandos. Sustituye 5d81fa por tu propia cadena aleatoria en todo el 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)

Paso 3: escribir el servidor MCP para MQTT

Este es el núcleo del tutorial. FastMCP, del SDK oficial, convierte funciones Python tipadas en herramientas MCP y usa los docstrings como las descripciones que lee el modelo.

# 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

# el transporte stdio usa stdout para los mensajes del protocolo: registra solo en stderr
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()  # transporte stdio por defecto

Aquí importan tres detalles:

1. Nunca imprimas en stdout. Con el transporte stdio, stdout transporta el protocolo JSON-RPC. Un print() perdido corrompe el flujo y el cliente se desconecta con un error de análisis. Registra en stderr.
2. El bucle MQTT se ejecuta en su propio hilo. loop_start() mantiene la caché actualizada mientras el servidor MCP atiende las llamadas a herramientas.
3. La validación vive en el servidor. El docstring le dice al modelo qué está permitido, pero es el código el que lo impone, también para los valores que el modelo se invente.

Paso 4: probar con MCP Inspector

Arranca el simulador en una terminal y luego ejecuta el servidor bajo el inspector, que abre una interfaz web donde puedes llamar a las herramientas a mano (requiere Node.js):

python simulator.py
mcp dev mqtt_mcp_server.py

Llama a list_topics, después a get_latest con pump-1 y luego a send_command con fan-1, set_speed, 70. Deberías ver una confirmación con "result": "applied". Para observar el tráfico en bruto al mismo tiempo:

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

Paso 5: conectar un asistente de IA

Los clientes MCP lanzan los servidores stdio a partir de un archivo de configuración. En los asistentes de escritorio que usan el formato habitual mcpServers, añade una entrada con las rutas absolutas al Python de tu entorno virtual y al script:

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

Reinicia el asistente y pregunta: "¿Cuál es ahora mismo el nivel del depósito y está funcionando la bomba?". El modelo llama dos veces a get_latest y responde con valores en vivo. Después prueba "Pon fan-1 al 30 por ciento": verás la llamada a la herramienta, la confirmación y la respuesta. La mayoría de los clientes te piden confirmar las llamadas a herramientas; mantén esa confirmación activada en cualquier herramienta que cambie el estado de un dispositivo.

Escribir descripciones de herramientas que el modelo entienda

Con FastMCP, el nombre de la función, los nombres de los parámetros, las anotaciones de tipo y el docstring son todo lo que el modelo sabe de una herramienta. Las descripciones vagas producen un uso vago de las herramientas, así que trátalas como parte de la interfaz:

1. Di qué devuelve la herramienta, no solo qué hace. "Devuelve el último mensaje de estado de un dispositivo y su antigüedad en segundos" le indica al modelo que puede razonar sobre la frescura de los datos.
2. Nombra los valores válidos. Enumerar en el docstring las acciones y rangos permitidos reduce las llamadas fallidas, aunque el código siga imponiéndolos.
3. Usa nombres del dominio, no rutas de topic. El modelo debe pedir pump-1, no construir una cadena de topic completa. Ocultar la estructura de topics tras IDs de dispositivo evita que el modelo adivine rutas y te permite reorganizar los topics más adelante.
4. Devuelve los errores como datos. Devolver {"error": "unknown device fan-9"} permite que el modelo se corrija o explique el problema al usuario, mientras que lanzar una excepción suele terminar el intercambio con un fallo genérico.
5. Mantén corta la lista de herramientas. Entre tres y seis herramientas bien enfocadas funcionan mejor que una docena que se solapan. Si necesitas más, repártelas entre varios servidores según su propósito.

También ayuda incluir la antigüedad de cada valor en los resultados de lectura. Un modelo que ve "ageSeconds": 900 normalmente mencionará que la lectura está desactualizada, que es justo la cautela que quieres cuando el dispositivo que hay detrás podría estar desconectado.

Endurecer el servidor MCP para producción

El tutorial se ejecuta sobre el servidor MQTT público gratuito, ideal para experimentar pero legible por cualquiera. Antes de apuntar un asistente a equipos reales:

Un patrón práctico es usar dos servidores MCP: uno de solo lectura que puedes activar libremente y uno de comandos que activas solo cuando hace falta. En un broker MQTT gestionado como ConnectMQ, cada servidor obtiene en el portal su propio usuario, contraseña y client ID generados, y puedes observar ambos clientes en el explorador de topics en vivo mientras pruebas. Recuerda que el broker no ofrece permisos por topic, así que la separación entre leer y enviar comandos debe aplicarse en tus servidores.

Para conocer los principios de diseño de superficies de herramientas seguras, lee nuestra guía para construir un agente LLM para dispositivos IoT, y consulta function calling de LLM para IoT para ver patrones de validación de esquemas. La documentación oficial de Model Context Protocol describe otros transportes, como streamable HTTP, si necesitas alojar el servidor en remoto. Para una visión de conjunto, nuestra guía principal de IA e IoT conecta todas estas piezas.

Empieza con ConnectMQ

Apunta tu servidor MCP a un broker que controles. ConnectMQ ofrece un broker MQTT en la nube con MQTT 5, TLS en el puerto 8883 y un cliente en el navegador para hacer pruebas. Empieza con el plan Serverless gratuito, sin tarjeta de crédito, y da a tu asistente contexto en vivo de tus dispositivos en minutos.

Preguntas frecuentes

¿Por qué necesita una caché el servidor MCP?

MQTT envía los mensajes cuando los dispositivos publican, mientras que las llamadas a herramientas MCP piden datos bajo demanda. La caché guarda el último mensaje de cada topic para que una llamada a herramienta pueda responder de inmediato en lugar de esperar a la siguiente publicación.

¿Puede el asistente de IA suscribirse directamente a los topics?

No mediante herramientas MCP en el sentido de un flujo continuo. El servidor se suscribe en nombre del asistente y expone instantáneas, lo que además permite controlar qué topics son visibles.

¿Es seguro dejar que un asistente envíe comandos MQTT?

Solo mediante herramientas acotadas con validación, límites en el dispositivo, confirmaciones y confirmación del usuario. Nunca expongas a un modelo una herramienta de publicación genérica.

¿Por qué mi cliente muestra un error de análisis JSON al iniciar el servidor?

Algo está escribiendo en stdout, que el transporte stdio reserva para los mensajes del protocolo. Envía todo el registro y la salida de depuración a stderr.