Crie um Bot do Telegram com MQTT para Alertas e Comandos de Dispositivos

Um bot do Telegram com MQTT transforma seu celular em um console de bolso para os seus dispositivos: os alarmes chegam como mensagens no chat, e um rápido /off pump1 publica um comando de volta para o dispositivo, por exemplo a bomba de uma estação de saneamento ou de um pivô de irrigação numa fazenda. Neste tutorial você vai construir esse bot em Python com paho-mqtt e a API HTTP de bots do Telegram, protegê-lo com uma lista de chats permitidos e executá-lo em um broker real. Tudo cabe em um único script com duas dependências.

Como o Bot se Encaixa no Seu Sistema MQTT

O bot é apenas mais um cliente MQTT. Os dispositivos continuam publicando e assinando exatamente como antes, e o bot fica entre o broker e o Telegram:

device --publish alert--> MQTT broker --> telegram_bot.py --sendMessage--> your chat
your chat --/off pump1--> telegram_bot.py --publish cmd--> MQTT broker --> device

Nada do lado do dispositivo sabe que o Telegram existe. Essa separação é importante: você pode trocar o Telegram por e-mail ou Slack depois sem mexer no firmware, e nunca coloca um token de bot em um microcontrolador. Se quiser relembrar o que um broker MQTT faz nesse cenário: ele simplesmente roteia mensagens entre topics e clientes.

Design de Topics para Alertas e Comandos

Use uma estrutura previsível para que o bot consiga extrair os nomes dos dispositivos dos topics:

O prefixo connectmq/demo/tgbot-9214 é um namespace aleatório para testes no servidor MQTT público (gratuito) broker.connectmq.com, feito para testes e protótipos. O servidor MQTT público pode ser lido por qualquer pessoa que assine o mesmo topic, então escolha seu próprio sufixo aleatório e use o bot apenas com dispositivos de teste até migrar para um broker privado.

Passo 1: Crie o Bot e Descubra o ID do Seu Chat

1. No Telegram, abra uma conversa com o @BotFather, envie /newbot e siga as instruções. Você recebe um token como 123456789:AA.... Trate-o como uma senha.
2. Abra uma conversa com o seu novo bot e envie qualquer mensagem, por exemplo oi.
3. Em um navegador ou terminal, chame getUpdates para descobrir o ID do seu chat:

curl -s "https://api.telegram.org/botYOUR_BOT_TOKEN/getUpdates"

Procure "chat":{"id":123456789,...} na resposta. Esse número vai para a lista de permitidos. Para um grupo, adicione o bot ao grupo e use o ID negativo do chat do grupo. A documentação da Telegram Bot API descreve todos os métodos usados abaixo.

Passo 2: Instale as Dependências

pip install "paho-mqtt>=2.0" requests

O script usa chamadas HTTPS simples em vez de um framework de bots. Isso mantém o uso de threads simples: o paho executa seu loop de rede em uma thread em segundo plano, e a thread principal faz long polling no Telegram em busca de comandos.

Passo 3: O Bot do Telegram com MQTT

Salve como telegram_bot.py:

import json
import os
import threading
import time
import uuid

import paho.mqtt.client as mqtt
import requests

BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
ALLOWED_CHATS = {int(c) for c in os.environ["TELEGRAM_CHAT_IDS"].split(",")}
API = f"https://api.telegram.org/bot{BOT_TOKEN}"

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"tgbot-{uuid.uuid4().hex[:8]}")
PREFIX = os.environ.get("TOPIC_PREFIX", "connectmq/demo/tgbot-9214")
ALERT_COOLDOWN = int(os.environ.get("ALERT_COOLDOWN", "300"))  # segundos por dispositivo

latest = {}            # dispositivo -> (timestamp, dict do payload)
last_alert = {}        # dispositivo -> timestamp do último alerta encaminhado
state_lock = threading.Lock()


def send(chat_id: int, text: str):
    try:
        r = requests.post(f"{API}/sendMessage", json={"chat_id": chat_id, "text": text}, timeout=10)
        if r.status_code == 429:
            retry = r.json().get("parameters", {}).get("retry_after", 5)
            time.sleep(retry)
            requests.post(f"{API}/sendMessage", json={"chat_id": chat_id, "text": text}, timeout=10)
    except requests.RequestException as exc:
        print(f"Telegram send failed: {exc}")


def broadcast(text: str):
    for chat_id in ALLOWED_CHATS:
        send(chat_id, text)


# ---------- Lado MQTT ----------

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"MQTT connect failed: {reason_code}")
        return
    client.subscribe(f"{PREFIX}/+/telemetry", qos=0)
    client.subscribe(f"{PREFIX}/+/alert", qos=1)
    print("MQTT connected and subscribed")


def on_message(client, userdata, msg):
    parts = msg.topic.split("/")
    device, kind = parts[-2], parts[-1]
    try:
        payload = json.loads(msg.payload)
    except (json.JSONDecodeError, UnicodeDecodeError):
        payload = {"raw": msg.payload.decode("utf-8", errors="replace")}

    now = time.time()
    if kind == "telemetry":
        with state_lock:
            latest[device] = (now, payload)
    elif kind == "alert" and not msg.retain:
        with state_lock:
            if now - last_alert.get(device, 0) < ALERT_COOLDOWN:
                return
            last_alert[device] = now
        level = str(payload.get("level", "alert")).upper()
        text = payload.get("text", json.dumps(payload))
        # envia de uma thread separada para que uma chamada HTTP lenta nunca bloqueie o loop MQTT
        threading.Thread(target=broadcast, args=(f"[{level}] {device}: {text}",), daemon=True).start()


# ---------- Lado Telegram ----------

def handle_command(client: mqtt.Client, chat_id: int, text: str):
    parts = text.strip().split()
    cmd = parts[0].split("@")[0].lower()
    arg = parts[1] if len(parts) > 1 else None

    if cmd in ("/start", "/help"):
        send(chat_id, "Commands:\n/status - latest values\n/on <device>\n/off <device>")
    elif cmd == "/status":
        with state_lock:
            snapshot = dict(latest)
        if not snapshot:
            send(chat_id, "No telemetry received yet.")
            return
        lines = []
        for device, (ts, data) in sorted(snapshot.items()):
            age = int(time.time() - ts)
            lines.append(f"{device}: {json.dumps(data)} ({age}s ago)")
        send(chat_id, "\n".join(lines))
    elif cmd in ("/on", "/off"):
        if not arg or "/" in arg or "+" in arg or "#" in arg:
            send(chat_id, f"Usage: {cmd} <device>")
            return
        body = json.dumps({"action": cmd[1:], "by": f"telegram:{chat_id}"})
        info = client.publish(f"{PREFIX}/{arg}/cmd", body, qos=1)
        info.wait_for_publish(timeout=5)
        send(chat_id, f"Sent '{cmd[1:]}' to {arg}" if info.is_published() else "Publish failed, broker not reachable")
    else:
        send(chat_id, "Unknown command. Try /help")


def poll_telegram(client: mqtt.Client):
    offset = None
    while True:
        try:
            r = requests.get(f"{API}/getUpdates", params={"timeout": 30, "offset": offset}, timeout=40)
            for update in r.json().get("result", []):
                offset = update["update_id"] + 1
                message = update.get("message") or {}
                chat_id = message.get("chat", {}).get("id")
                text = message.get("text", "")
                if chat_id not in ALLOWED_CHATS:
                    print(f"Ignoring chat {chat_id}")
                    continue
                if text.startswith("/"):
                    handle_command(client, chat_id, text)
        except requests.RequestException as exc:
            print(f"Telegram polling error: {exc}")
            time.sleep(5)


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=60)
    client.loop_start()
    broadcast("Bot started")
    poll_telegram(client)


if __name__ == "__main__":
    main()

Passo 4: Execute

export TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN
export TELEGRAM_CHAT_IDS=123456789
python telegram_bot.py

Simule um dispositivo em outro terminal. Primeiro um pouco de telemetria, depois um alerta e, por fim, escute os comandos:

mosquitto_pub -h broker.connectmq.com -p 1883 -i tg-dev-9214 \
  -t connectmq/demo/tgbot-9214/pump1/telemetry -m '{"temp": 41.2, "running": true}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i tg-dev-9214 \
  -t connectmq/demo/tgbot-9214/pump1/alert -m '{"level": "high", "text": "Temperature above 80 C"}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i tg-listen-9214 -v \
  -t "connectmq/demo/tgbot-9214/+/cmd"

Seu celular mostra [HIGH] pump1: Temperature above 80 C. Envie /status para ver a telemetria em cache e /off pump1 para ver o comando aparecer na janela do mosquitto_sub. Em hardware real, o dispositivo assina seu próprio topic cmd; o tutorial de MQTT com ESP32 mostra o lado do dispositivo.

Regras de Segurança que Você Não Deve Pular

Um bot de chat capaz de ligar e desligar equipamentos é um controle remoto. Trate-o como tal:

1. Use uma lista de IDs de chat permitidos. O script ignora todo chat que não está na lista. Qualquer pessoa pode encontrar seu bot pelo nome, então essa verificação é obrigatória.
2. Valide o argumento do dispositivo. O código rejeita /, + e #, para que ninguém consiga publicar fora do topic pretendido nem injetar curingas.
3. Mantenha o token fora do controle de versão. Use variáveis de ambiente ou um gerenciador de segredos e revogue o token no BotFather se ele vazar.
4. Nunca controle equipamentos reais pelo servidor MQTT público. Qualquer pessoa pode publicar em topics públicos. Use um broker privado com credenciais e TLS.
5. Adicione confirmação para ações perigosas. Para qualquer coisa que movimente máquinas, exija uma segunda mensagem, como /confirm, antes de publicar.

O checklist mais amplo de boas práticas de segurança MQTT se aplica ao bot tanto quanto aos dispositivos.

Lidando com Tempestades de Alertas

Um sensor instável pode enviar centenas de alertas por minuto, e o Telegram limita a taxa de bots que enviam mensagens rápido demais. O script aplica um tempo de espera por dispositivo (cinco minutos por padrão) e respeita o valor retry_after que o Telegram retorna com HTTP 429. Ele também ignora alertas retidos, para que uma reconexão não reenvie o alarme de ontem. Para condições persistentes, faça o dispositivo publicar um alerta separado de "normalizado", para que o histórico do chat conte a história completa.

Solução de Problemas

• Nenhuma mensagem chega ao Telegram. Verifique se TELEGRAM_CHAT_IDS contém o ID numérico do seu chat e se você enviou pelo menos uma mensagem ao bot antes; bots não podem iniciar conversas.
• 409 Conflict no getUpdates. Há um webhook configurado para o bot, ou uma segunda cópia do script está em execução. Exclua o webhook com o método deleteWebhook e execute apenas uma instância.
• Os comandos funcionam, os alertas não. O topic de alerta não corresponde a <prefix>/<device>/alert, ou o tempo de espera ainda está ativo para aquele dispositivo.
• O bot reconecta ao MQTT repetidamente. Duas instâncias compartilham um client ID. Mantenha IDs únicos por processo.

Executando o Bot como Serviço

Um bot que só roda enquanto seu notebook está aberto não é exatamente um sistema de alertas. Em um Raspberry Pi ou em uma pequena VM Linux, execute-o sob o systemd para que ele inicie no boot e reinicie após falhas. Coloque os segredos em um arquivo de ambiente legível apenas pelo usuário do serviço:

sudo tee /etc/telegram-mqtt-bot.env >/dev/null <<'EOF'
TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN
TELEGRAM_CHAT_IDS=123456789
MQTT_HOST=broker.connectmq.com
MQTT_PORT=1883
EOF
sudo chmod 600 /etc/telegram-mqtt-bot.env

sudo tee /etc/systemd/system/telegram-mqtt-bot.service >/dev/null <<'EOF'
[Unit]
Description=MQTT Telegram bot
After=network-online.target
Wants=network-online.target

[Service]
EnvironmentFile=/etc/telegram-mqtt-bot.env
ExecStart=/usr/bin/python3 /opt/telegram-bot/telegram_bot.py
Restart=always
RestartSec=5
User=pi

[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload
sudo systemctl enable --now telegram-mqtt-bot
journalctl -u telegram-mqtt-bot -f

A mensagem "Bot started" que o script envia ao iniciar também funciona como notificação de reinício: se você a vir inesperadamente, o serviço caiu ou a máquina reiniciou, e o journalctl diz o motivo. Lembre-se de que o bot é um ponto único de falha para os alertas no chat. Se ele cair, nada avisa você, então combine-o com um heartbeat simples: publique a partir do bot um status retido com um Last Will offline e deixe outro sistema monitorar esse topic.

Migrando para um Broker Privado do ConnectMQ

Quando o bot passar a controlar dispositivos reais, migre-o para um broker privado. Crie um no ConnectMQ, copie do portal o nome de usuário, a senha e o client ID gerados e defina MQTT_HOST, MQTT_PORT=8883, MQTT_USERNAME, MQTT_PASSWORD e MQTT_CLIENT_ID. Com um broker MQTT gerenciado, o único processo que você opera é o próprio bot.

O ConnectMQ também oferece regras de alerta por limite e um log de alertas no portal, úteis para ver os alarmes de todo o broker de relance. O bot complementa isso ao enviar alertas selecionados às pessoas onde elas já estão. Se sua equipe prefere a caixa de entrada ao chat, a mesma estrutura funciona com SMTP em alertas por e-mail MQTT com Python. Para outras linguagens e frameworks, o guia de bibliotecas cliente MQTT compara as opções.

Comece a Usar o ConnectMQ

Execute hoje mesmo seu bot do Telegram em um broker MQTT gerenciado, privado e protegido por TLS. Cadastre-se em app.connectmq.com, escolha o plano Serverless gratuito (sem cartão de crédito), crie um broker e cole as credenciais no ambiente do bot. O cliente MQTT no navegador do portal é uma forma prática de disparar alertas de teste enquanto você ajusta o tempo de espera.

Perguntas frequentes

Preciso de um servidor para executar um bot do Telegram com MQTT?

Você precisa de um lugar onde o script rode continuamente, como um Raspberry Pi, uma pequena VM ou um contêiner. Ele só faz conexões de saída, então não são necessárias portas abertas nem IP público.

Várias pessoas podem controlar dispositivos pelo mesmo bot?

Sim. Adicione o ID do chat de cada pessoa a TELEGRAM_CHAT_IDS, ou adicione o bot a um grupo e coloque o ID do grupo na lista de permitidos. Inclua o remetente no payload do comando, como o script faz, para que os dispositivos possam registrar quem o enviou.

Por que usar long polling em vez de um webhook do Telegram?

Webhooks exigem um endpoint HTTPS público. O long polling funciona atrás de NAT e firewalls, o que combina com um bot rodando em um gateway ou Raspberry Pi ao lado dos seus dispositivos.

Os alertas serão entregues se o bot estiver offline?

Alertas publicados enquanto o bot está desconectado são perdidos, a menos que você use uma sessão persistente com QoS 1. Para alarmes críticos, mantenha um segundo canal de notificação além do chat.