Controle dispositivos por um bot do Telegram via MQTT

Um bot do Telegram é um dos jeitos mais rápidos de ter um controle remoto e um canal de alertas para dispositivos MQTT: você ganha uma interface de chat gratuita e multiplataforma, notificações push e nenhum app para desenvolver. O bot roda como um pequeno serviço que faz duas coisas — publica comandos como ON em devices/lamp/set quando você digita /on lamp, e assina topics de alerta e de estado para encaminhá-los ao seu chat.

Este tutorial monta essa ponte em Python com python-telegram-bot e aiomqtt, restringe o acesso ao seu próprio chat ID e testa cada caminho com mosquitto_pub e mosquitto_sub antes de envolver qualquer hardware.

Arquitetura

You (Telegram app) ──/on lamp──▶ Telegram API ──▶ bot.py ──PUBLISH testmqtt/tgbot/lamp/set "ON"──▶ broker ──▶ device
device ──PUBLISH testmqtt/tgbot/lamp/state "ON" (retained)──▶ broker ──▶ bot.py (cache for /status)
device ──PUBLISH testmqtt/tgbot/greenhouse/alert "..."──▶ broker ──▶ bot.py ──▶ Telegram chat

O dispositivo nunca fala com o Telegram. Ele só fala MQTT, então um ESP32 ou um Raspberry Pi não precisa de token do bot nem de cliente HTTPS, e depois você pode adicionar um dashboard ou o Home Assistant como segundo controlador nos mesmos topics. A estrutura de topics segue a separação usual entre comando e estado descrita em comando e controle com MQTT:

Crie o bot

1. No Telegram, abra um chat com o @BotFather, envie /newbot e escolha um nome e um username terminado em bot.
2. Copie o token que o BotFather devolve. Ele é secreto: quem tiver o token pode se passar pelo seu bot.
3. Se quiser, envie /setcommands ao BotFather com on - switch a device on, off - switch a device off, status - show device states, para os comandos aparecerem no menu do chat.

Instale as duas bibliotecas (Python 3.10+):

pip install "python-telegram-bot>=21" "aiomqtt>=2"

As duas são bibliotecas asyncio, e é por isso que combinam: o loop de polling do bot e a conexão MQTT rodam no mesmo event loop, e um handler de comando pode fazer await client.publish(...) diretamente. Veja MQTT assíncrono em Python com aiomqtt para mais detalhes sobre o cliente.

O bot

@@BOTCODE@@

Algumas decisões de design que valem destaque:

• Uma única conexão MQTT para o bot inteiro. O worker é dono da conexão e a guarda em bot_data; os handlers a reutilizam em vez de conectar a cada comando. Se o broker cair, o loop reconecta após cinco segundos e, enquanto isso, os handlers respondem "Broker not connected".
• Os estados vêm de mensagens retidas. Como os dispositivos publicam state com a flag retain, o bot conhece o último estado de cada dispositivo logo após assinar, mesmo depois de reiniciar. Veja mensagens retidas.
• Chats desconhecidos recebem silêncio. Só o /whoami responde a todos, para você descobrir seu chat ID. Todo comando que publica confere a allowlist primeiro.
• Os alertas têm debounce. Sem isso, um sensor instável estouraria os limites do Telegram e afogaria os alertas reais.

Rode e teste sem hardware

Inicie o bot só com TELEGRAM_TOKEN definido, envie /whoami no chat e anote o número. Reinicie com a allowlist:

export TELEGRAM_TOKEN="123456:ABC..."        # do BotFather
export ALLOWED_CHAT_IDS="111111111"          # seu chat ID do /whoami
export MQTT_USER="username" MQTT_PASS="password"   # se o seu broker exigir login
python bot.py

Agora faça o papel do dispositivo. Em um terminal, observe o que o bot publica:

mosquitto_sub -h broker.connectmq.com -p 1883 -t 'testmqtt/tgbot/+/set' -v

Envie /on lamp no Telegram. O terminal 1 mostra testmqtt/tgbot/lamp/set ON. Informe o novo estado como um dispositivo faria e depois dispare um alerta:

mosquitto_pub -h broker.connectmq.com -p 1883 -q 1 -r \
  -t testmqtt/tgbot/lamp/state -m ON

mosquitto_pub -h broker.connectmq.com -p 1883 -q 1 \
  -t testmqtt/tgbot/pump/alert -m 'Water level low (12 %)'

Agora /status deve mostrar lamp: ON, e o chat recebe ⚠️ pump: Water level low (12 %). Publique o alerta de novo em menos de 60 segundos e confirme que ele é suprimido. Você também pode fazer tudo isso pelo cliente MQTT online em vez do terminal. Por fim, limpe o estado retido para ele não ficar lá: mosquitto_pub -h broker.connectmq.com -t testmqtt/tgbot/lamp/state -r -n.

Acompanhe os topics do bot no navegador

O lado do dispositivo

Em hardware real, o dispositivo assina o topic set, aciona a saída e publica o estado resultante com retain. Em MicroPython com umqtt.simple, o núcleo cabe em poucas linhas:

@@DEVCODE@@

Publicar o estado depois de agir é o que torna o /status confiável: o bot informa o que o dispositivo fez, não o que você pediu. Adicione um Last Will no estado ou um topic availability para o bot avisar quando um dispositivo ficar offline.

Proteja o bot

Qualquer pessoa no Telegram consegue alcançar um bot, então trate-o como uma API pública:

1. Allowlist de chats, como mostrado. Em grupos, o chat ID é negativo; adicione-o à lista como qualquer outro.
2. Mantenha o token fora do código — variáveis de ambiente ou um gerenciador de segredos, nunca um repositório Git.
3. Use TLS até o broker. O exemplo conecta na porta 8883; a 1883 sem criptografia só serve para experimentos locais.
4. Restrinja o usuário do bot no broker. Ele precisa de escrita em .../set e leitura em .../state e .../alert, nada mais. Verifique as regras com um teste de permissões ACL.

Alternativas sem código

Se preferir uma ferramenta visual, a mesma ponte funciona no n8n (MQTT Trigger → nó Telegram, e Telegram Trigger → nó MQTT — veja automações MQTT com n8n) ou no Node-RED com um nó de Telegram da comunidade. A versão em Python ganha quando você quer um único serviço pequeno, com allowlist explícita, debounce e testes no seu próprio repositório.

Um bot do Telegram transforma o MQTT em algo que você controla pelo bolso: comandos saem como publish, o estado volta como mensagens retidas e os alertas chegam como notificações push. Deixe o dispositivo falando só MQTT, restrinja o bot ao seu chat, faça debounce dos alertas e teste o ciclo inteiro com mosquitto antes de ligar um relé.

Perguntas frequentes