Servidor MCP para MQTT: Deixe Assistentes de IA Lerem e Controlarem Seus Dispositivos

Um servidor MCP para MQTT dá aos assistentes de IA uma forma estruturada e controlada por permissões de consultar dados de dispositivos ao vivo e, quando você permitir, enviar comandos. Neste tutorial você vai construir em Python um servidor Model Context Protocol que assina um broker MQTT, guarda em cache a mensagem mais recente de cada topic e expõe um pequeno conjunto de ferramentas; depois vai conectá-lo a um assistente compatível com MCP e fazer perguntas sobre seus dispositivos em linguagem natural.

O que o MCP acrescenta a um sistema MQTT

O Model Context Protocol é um protocolo aberto para conectar aplicações de IA a ferramentas e dados externos. Um servidor MCP declara ferramentas (funções com parâmetros tipados e descrições) e o client, geralmente um assistente de desktop ou um agente de IDE, permite que o modelo as chame. O servidor decide o que de fato é possível.

MQTT e MCP resolvem problemas diferentes e se combinam bem:

O servidor MCP fica entre os dois: de um lado é um client MQTT comum e, do outro, um provedor de ferramentas MCP. Como o MQTT é baseado em push e o MCP em pull, o servidor mantém um cache com a mensagem mais recente de cada topic, para que as chamadas de ferramenta respondam na hora em vez de esperar pelo próximo publish.

Arquitetura

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

O servidor assina apenas um prefixo, e os comandos só podem ser publicados em .../devices/<id>/cmd com uma ação presente na lista de permissões. O modelo nunca recebe acesso bruto de publish a topics arbitrários.

Passo 1: instalar as dependências

Crie um ambiente virtual e instale o SDK oficial do MCP para Python e o 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"

Passo 2: simular alguns dispositivos

Para ter o que consultar, execute um pequeno simulador que publica, a cada cinco segundos, o estado de três dispositivos como retained message e confirma os comandos recebidos. Substitua 5d81fa pela sua própria string aleatória em todo este 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)

Passo 3: escrever o servidor MCP para MQTT

Este é o núcleo do tutorial. O FastMCP, do SDK oficial, transforma funções Python tipadas em ferramentas MCP, usando as docstrings como as descrições que o modelo lê.

# 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

Três detalhes são importantes aqui:

1. Nunca escreva no stdout. Com o transporte stdio, o stdout transporta o protocolo JSON-RPC. Um print() perdido corrompe o fluxo e o client se desconecta com um erro de parse. Registre os logs no stderr.
2. O loop MQTT roda na sua própria thread. loop_start() mantém o cache atualizado enquanto o servidor MCP atende às chamadas de ferramenta.
3. A validação fica no servidor. A docstring diz ao modelo o que é permitido, mas é o código que impõe as regras, inclusive para valores que o modelo inventar.

Passo 4: testar com o MCP Inspector

Inicie o simulador em um terminal e depois execute o servidor sob o inspector, que abre uma interface web onde você pode chamar as ferramentas manualmente (é preciso ter Node.js):

python simulator.py
mcp dev mqtt_mcp_server.py

Chame list_topics, depois get_latest com pump-1 e, em seguida, send_command com fan-1, set_speed, 70. Você deve ver uma confirmação com "result": "applied". Para acompanhar o tráfego bruto ao mesmo tempo:

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

Passo 5: conectar um assistente de IA

Os clients MCP iniciam servidores stdio a partir de um arquivo de configuração. Para assistentes de desktop que usam o formato comum mcpServers, adicione uma entrada com os caminhos absolutos do Python do seu ambiente virtual e do script:

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

Reinicie o assistente e pergunte: "Qual é o nível do tanque agora e a bomba está ligada?" O modelo chama get_latest duas vezes e responde com valores ao vivo. Depois experimente "Coloque o fan-1 em 30 por cento": você verá a chamada de ferramenta, a confirmação e a resposta. A maioria dos clients pede que você confirme as chamadas de ferramenta; mantenha essa confirmação ativada para qualquer ferramenta que altere o estado de um dispositivo.

Escrevendo descrições de ferramentas que o modelo entende

Com o FastMCP, o nome da função, os nomes dos parâmetros, as anotações de tipo e a docstring são tudo o que o modelo sabe sobre uma ferramenta. Descrições vagas produzem uso vago das ferramentas, então trate-as como parte da interface:

1. Diga o que a ferramenta retorna, não só o que ela faz. "Retorna a mensagem de estado mais recente de um dispositivo e sua idade em segundos" informa ao modelo que ele pode raciocinar sobre a atualidade do dado.
2. Indique os valores válidos. Listar as ações e faixas permitidas na docstring reduz chamadas com falha, mesmo que o código continue impondo as regras.
3. Use nomes do domínio, não caminhos de topic. O modelo deve pedir pump-1, e não montar uma string de topic completa. Esconder a organização dos topics atrás de IDs de dispositivo evita que o modelo adivinhe caminhos e permite reorganizar os topics depois.
4. Retorne erros como dados. Retornar {"error": "unknown device fan-9"} permite que o modelo se corrija ou explique o problema ao usuário, enquanto lançar uma exceção muitas vezes encerra a interação com uma falha genérica.
5. Mantenha a lista de ferramentas curta. De três a seis ferramentas focadas funcionam melhor do que uma dúzia com funções sobrepostas. Se precisar de mais, divida-as entre servidores por finalidade.

Também ajuda incluir a idade de cada valor nos resultados de leitura. Um modelo que vê "ageSeconds": 900 geralmente menciona que a leitura está desatualizada, que é exatamente a cautela que você quer quando o dispositivo por trás dela pode estar offline.

Reforçando o servidor MCP para produção

O tutorial roda no servidor MQTT público (gratuito), que é ideal para experimentos, mas pode ser lido por qualquer pessoa. Antes de apontar um assistente para equipamentos reais:

Um padrão prático é usar dois servidores MCP: um somente leitura, que você pode habilitar à vontade, e um servidor de comandos, que você habilita apenas quando necessário. Em um broker MQTT gerenciado como o ConnectMQ, cada servidor recebe pelo portal seu próprio usuário, senha e client ID gerados, e você pode observar os dois clients no explorador de topics ao vivo enquanto testa. Lembre-se de que o broker não oferece permissões por topic, então a separação entre leitura e comando precisa ser imposta nos seus servidores.

Para os princípios de projeto por trás de superfícies de ferramentas seguras, leia nosso guia sobre como construir um agente LLM para dispositivos IoT e veja function calling de LLM para IoT para padrões de validação de schema. A documentação oficial do Model Context Protocol descreve outros transportes, como o streamable HTTP, caso você precise hospedar o servidor remotamente. Para a visão geral, nosso guia completo de IA e IoT conecta todas essas peças.

Comece com o ConnectMQ

Aponte seu servidor MCP para um broker que você controla. O ConnectMQ oferece um broker MQTT na nuvem com MQTT 5, TLS na porta 8883 e um client no navegador para testes. Comece no plano Serverless gratuito, sem cartão de crédito, e dê ao seu assistente contexto ao vivo dos dispositivos em minutos.

Perguntas frequentes

Por que o servidor MCP precisa de um cache?

O MQTT envia mensagens quando os dispositivos publicam, enquanto as chamadas de ferramenta do MCP pedem dados sob demanda. O cache armazena a mensagem mais recente de cada topic para que uma chamada de ferramenta possa responder imediatamente em vez de esperar pelo próximo publish.

O assistente de IA pode assinar topics diretamente?

Não por meio de ferramentas MCP, no sentido de streaming. O servidor assina os topics em nome do assistente e expõe snapshots, o que também mantém o controle sobre quais topics ficam visíveis.

É seguro deixar um assistente enviar comandos MQTT?

Somente por meio de ferramentas estreitas com validação, limites no dispositivo, confirmações e confirmação do usuário. Nunca exponha uma ferramenta genérica de publish a um modelo.

Por que meu client mostra um erro de parse de JSON ao iniciar o servidor?

Algo está escrevendo no stdout, que o transporte stdio reserva para as mensagens do protocolo. Envie todos os logs e saídas de depuração para o stderr.