Un bot Telegram MQTT transforme votre téléphone en console de poche pour vos appareils : les alarmes arrivent sous forme de messages, et un rapide /off pump1 publie une commande vers l'appareil. Dans ce tutoriel, vous allez construire ce bot en Python avec paho-mqtt et l'API HTTP Telegram Bot, le protéger par une liste blanche de chats et l'exécuter avec un vrai broker. Le tout tient dans un seul script avec deux dépendances.
La place du bot dans votre système MQTT
Le bot n'est qu'un client MQTT de plus. Les appareils continuent de publier et de s'abonner exactement comme avant, et le bot se place entre le broker et Telegram :
device --publish alert--> MQTT broker --> telegram_bot.py --sendMessage--> your chat
your chat --/off pump1--> telegram_bot.py --publish cmd--> MQTT broker --> device
Rien côté appareil ne sait que Telegram existe. Cette séparation compte : vous pourrez remplacer Telegram par l'e-mail ou Slack plus tard sans toucher au firmware, et vous ne placez jamais de token de bot sur un microcontrôleur. Si vous voulez un rappel sur le rôle d'un broker MQTT dans ce schéma, il se contente de router les messages entre topics et clients.
Conception des topics pour alertes et commandes
Utilisez une structure prévisible pour que le bot puisse extraire le nom des appareils à partir des topics :
| Topic | Sens | Exemple de payload |
|---|---|---|
connectmq/demo/tgbot-9214/<device>/telemetry |
de l'appareil vers le bot | {"temp": 41.2} |
connectmq/demo/tgbot-9214/<device>/alert |
de l'appareil vers le bot | {"level": "high", "text": "Temperature above 80 C"} |
connectmq/demo/tgbot-9214/<device>/cmd |
du bot vers l'appareil | {"action": "off", "by": "telegram"} |
Le préfixe connectmq/demo/tgbot-9214 est un espace de noms aléatoire pour les tests sur le serveur MQTT public gratuit broker.connectmq.com. Ce serveur, réservé aux tests, peut être lu par quiconque est abonné au même topic : choisissez votre propre suffixe aléatoire et n'utilisez le bot qu'avec des appareils de test tant que vous n'êtes pas passé à un broker privé.
Étape 1 : créer le bot et trouver votre chat ID
- Dans Telegram, ouvrez une conversation avec @BotFather, envoyez
/newbotet suivez les instructions. Vous recevez un token comme123456789:AA.... Traitez-le comme un mot de passe. - Ouvrez une conversation avec votre nouveau bot et envoyez-lui n'importe quel message, par exemple
hi. - Dans un navigateur ou un terminal, appelez
getUpdatespour trouver votre chat ID :
curl -s "https://api.telegram.org/botYOUR_BOT_TOKEN/getUpdates"
Cherchez "chat":{"id":123456789,...} dans la réponse. Ce nombre va dans la liste blanche. Pour un groupe, ajoutez le bot au groupe et utilisez le chat ID négatif du groupe. La documentation de l'API Telegram Bot décrit chaque méthode utilisée ci-dessous.
Étape 2 : installer les dépendances
pip install "paho-mqtt>=2.0" requests
Le script utilise de simples appels HTTPS au lieu d'un framework de bot. Cela garde une gestion des threads simple : paho exécute sa boucle réseau dans un thread d'arrière-plan, et le thread principal interroge Telegram en long polling pour récupérer les commandes.
Étape 3 : le bot Telegram MQTT
Enregistrez-le sous 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")) # secondes par appareil
latest = {} # appareil -> (horodatage, dict du payload)
last_alert = {} # appareil -> horodatage de la dernière alerte transmise
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)
# ---------- Côté 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))
# envoi depuis un thread séparé pour qu'un appel HTTP lent ne bloque jamais la boucle MQTT
threading.Thread(target=broadcast, args=(f"[{level}] {device}: {text}",), daemon=True).start()
# ---------- Côté 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()
Étape 4 : l'exécuter
export TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN
export TELEGRAM_CHAT_IDS=123456789
python telegram_bot.py
Simulez un appareil dans un autre terminal : d'abord de la télémétrie, puis une alerte, puis l'écoute des commandes :
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"
Votre téléphone affiche [HIGH] pump1: Temperature above 80 C. Envoyez /status pour voir la télémétrie en cache et /off pump1 pour voir la commande apparaître dans la fenêtre mosquitto_sub. Sur du vrai matériel, l'appareil s'abonne à son propre topic cmd ; le tutoriel MQTT ESP32 montre la partie appareil.
Les règles de sécurité à ne pas négliger
Un bot de chat capable de commuter des équipements est une télécommande. Traitez-le comme tel :
- Liste blanche des chat IDs. Le script ignore tout chat qui n'est pas listé. N'importe qui peut trouver votre bot par son nom : cette vérification est donc obligatoire.
- Valider l'argument appareil. Le code rejette
/,+et#, pour que personne ne puisse publier hors du topic prévu ni injecter de wildcards. - Garder le token hors du contrôle de version. Utilisez des variables d'environnement ou un gestionnaire de secrets, et révoquez le token avec BotFather en cas de fuite.
- Ne jamais piloter de vrais équipements via le serveur MQTT public gratuit. N'importe qui peut publier sur les topics publics. Utilisez un broker privé avec identifiants et TLS.
- Ajouter une confirmation pour les actions dangereuses. Pour tout ce qui met en mouvement des machines, exigez un second message comme
/confirmavant de publier.
La liste plus complète des bonnes pratiques de sécurité MQTT s'applique au bot comme aux appareils.
Gérer les tempêtes d'alertes
Un capteur instable peut envoyer des centaines d'alertes par minute, et Telegram limite le débit des bots qui envoient des messages trop vite. Le script applique un délai de carence par appareil (cinq minutes par défaut) et respecte la valeur retry_after que Telegram renvoie avec le HTTP 429. Il ignore aussi les alertes retenues, pour qu'une reconnexion ne renvoie pas l'alarme de la veille. Pour les conditions persistantes, faites publier par l'appareil une alerte distincte de « retour à la normale » afin que l'historique du chat raconte toute l'histoire.
Dépannage
- Aucun message n'arrive dans Telegram. Vérifiez que
TELEGRAM_CHAT_IDScontient votre chat ID numérique et que vous avez d'abord envoyé au moins un message au bot ; les bots ne peuvent pas démarrer de conversation. 409 Conflictsur getUpdates. Un webhook est défini pour le bot, ou une seconde copie du script tourne. Supprimez le webhook avec la méthodedeleteWebhooket n'exécutez qu'une seule instance.- Les commandes fonctionnent, pas les alertes. Le topic d'alerte ne correspond pas à
<prefix>/<device>/alert, ou le délai de carence est encore actif pour cet appareil. - Le bot se reconnecte sans cesse à MQTT. Deux instances partagent un client ID. Gardez des IDs uniques par processus.
Exécuter le bot en tant que service
Un bot qui ne tourne que lorsque votre portable est ouvert ne fait pas un vrai système d'alerte. Sur un Raspberry Pi ou une petite VM Linux, exécutez-le sous systemd pour qu'il démarre au boot et redémarre après un plantage. Placez les secrets dans un fichier d'environnement lisible uniquement par l'utilisateur du service :
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
Le message « Bot started » que le script diffuse au démarrage sert aussi de notification de redémarrage : si vous le voyez de manière inattendue, le service a planté ou la machine a redémarré, et journalctl vous dit pourquoi. Gardez à l'esprit que le bot est un point de défaillance unique pour les alertes par chat. S'il tombe, rien ne vous prévient : associez-le donc à un simple heartbeat en publiant depuis le bot un état retenu avec un Last Will offline, et laissez un autre système surveiller ce topic.
Passer à un broker ConnectMQ privé
Lorsque le bot pilote de vrais appareils, passez-le sur un broker privé. Créez-en un sur ConnectMQ, copiez depuis le portail le username, le mot de passe et le client ID générés, puis définissez MQTT_HOST, MQTT_PORT=8883, MQTT_USERNAME, MQTT_PASSWORD et MQTT_CLIENT_ID. Avec un broker MQTT managé, le seul processus que vous exploitez est le bot lui-même.
ConnectMQ propose aussi des règles d'alerte à seuil et un journal des alertes dans le portail, utiles pour voir d'un coup d'œil les alarmes de tout votre broker. Le bot les complète en poussant des alertes sélectionnées vers les personnes là où elles se trouvent déjà. Si votre équipe préfère la boîte de réception au chat, la même structure fonctionne avec SMTP dans les alertes e-mail MQTT avec Python. Pour d'autres langages et frameworks, le guide des bibliothèques clientes MQTT compare les options.
Démarrer avec ConnectMQ
Faites tourner votre bot Telegram sur un broker privé protégé par TLS dès aujourd'hui. Inscrivez-vous sur app.connectmq.com, choisissez le plan Serverless gratuit (sans carte bancaire), créez un broker et collez les identifiants dans l'environnement du bot. Le client MQTT intégré au navigateur du portail est pratique pour déclencher des alertes de test pendant que vous ajustez le délai de carence.
Questions fréquentes
Ai-je besoin d'un serveur pour exécuter un bot Telegram MQTT ?
Il vous faut un endroit où le script peut tourner en continu, comme un Raspberry Pi, une petite VM ou un conteneur. Il n'établit que des connexions sortantes : aucun port ouvert ni IP publique n'est nécessaire.
Plusieurs personnes peuvent-elles piloter des appareils via le même bot ?
Oui. Ajoutez le chat ID de chaque personne à TELEGRAM_CHAT_IDS, ou ajoutez le bot à un groupe et mettez l'ID du groupe en liste blanche. Incluez l'expéditeur dans le payload de la commande, comme le fait le script, pour que les appareils puissent journaliser qui l'a émise.
Pourquoi utiliser le long polling plutôt qu'un webhook Telegram ?
Les webhooks nécessitent un endpoint HTTPS public. Le long polling fonctionne derrière un NAT et des pare-feu, ce qui convient à un bot tournant sur une passerelle ou un Raspberry Pi à côté de vos appareils.
Les alertes seront-elles livrées si le bot est hors ligne ?
Les alertes publiées pendant que le bot est déconnecté sont perdues, sauf si vous utilisez une session persistante avec le QoS 1. Pour les alarmes critiques, conservez un second canal de notification en plus du chat.
