Einen MQTT Telegram Bot für Gerätealarme und Befehle bauen

Ein MQTT Telegram Bot macht Ihr Telefon zur Hosentaschen-Konsole für Ihre Geräte: Alarme kommen als Chatnachrichten an, und ein kurzes /off pump1 publiziert einen Befehl zurück an das Gerät. In diesem Tutorial bauen Sie diesen Bot in Python mit paho-mqtt und der HTTP-API für Telegram-Bots, schützen ihn mit einer Chat-Allowlist und betreiben ihn gegen einen echten Broker. Das Ganze ist ein einziges Skript mit zwei Abhängigkeiten.

Wie der Bot in Ihr MQTT-System passt

Der Bot ist einfach ein weiterer MQTT-Client. Geräte publizieren und abonnieren genau wie bisher, und der Bot sitzt zwischen Broker und Telegram:

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

Auf der Geräteseite weiß nichts von Telegram. Diese Trennung ist wichtig: Sie können Telegram später gegen E-Mail oder Slack tauschen, ohne die Firmware anzufassen, und Sie legen nie ein Bot-Token auf einem Mikrocontroller ab. Falls Sie auffrischen möchten, was ein MQTT-Broker in diesem Bild tut: Er leitet einfach Nachrichten zwischen Topics und Clients weiter. Im Werk kann ein solcher Bot zum Beispiel ein schneller Kanal für die Rufbereitschaft der Instandhaltung sein, die Störmeldungen einer Linie auch nachts erhält.

Topic-Design für Alarme und Befehle

Verwenden Sie eine vorhersehbare Struktur, damit der Bot Gerätenamen aus den Topics lesen kann:

Das Präfix connectmq/demo/tgbot-9214 ist ein zufälliger Namensraum zum Testen auf dem kostenlosen öffentlichen MQTT-Broker broker.connectmq.com. Der öffentliche MQTT-Broker ist für Tests und Prototypen gedacht und kann von jedem gelesen werden, der dasselbe Topic abonniert. Wählen Sie also ein eigenes zufälliges Suffix und nutzen Sie den Bot nur mit Testgeräten, bis Sie auf einen privaten Broker umsteigen.

Schritt 1: Den Bot anlegen und Ihre Chat-ID finden

1. Öffnen Sie in Telegram einen Chat mit @BotFather, senden Sie /newbot und folgen Sie den Anweisungen. Sie erhalten ein Token wie 123456789:AA.... Behandeln Sie es wie ein Passwort.
2. Öffnen Sie einen Chat mit Ihrem neuen Bot und senden Sie ihm eine beliebige Nachricht, zum Beispiel hi.
3. Rufen Sie im Browser oder Terminal getUpdates auf, um Ihre Chat-ID zu finden:

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

Suchen Sie in der Antwort nach "chat":{"id":123456789,...}. Diese Zahl kommt in die Allowlist. Für eine Gruppe fügen Sie den Bot zur Gruppe hinzu und verwenden die negative Chat-ID der Gruppe. Die Dokumentation der Telegram Bot API beschreibt jede unten verwendete Methode.

Schritt 2: Abhängigkeiten installieren

pip install "paho-mqtt>=2.0" requests

Das Skript nutzt einfache HTTPS-Aufrufe statt eines Bot-Frameworks. Das hält das Threading einfach: paho betreibt seine Netzwerkschleife in einem Hintergrund-Thread, und der Haupt-Thread fragt Telegram per Long Polling nach Befehlen.

Schritt 3: Der MQTT Telegram Bot

Speichern Sie dies als 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"))  # Sekunden pro Gerät

latest = {}            # Gerät -> (Zeitstempel, Payload-Dict)
last_alert = {}        # Gerät -> Zeitstempel des letzten weitergeleiteten Alarms
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)


# ---------- MQTT-Seite ----------

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))
        # aus einem eigenen Thread senden, damit ein langsamer HTTP-Aufruf nie die MQTT-Schleife blockiert
        threading.Thread(target=broadcast, args=(f"[{level}] {device}: {text}",), daemon=True).start()


# ---------- Telegram-Seite ----------

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()

Schritt 4: Starten

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

Simulieren Sie in einem anderen Terminal ein Gerät. Erst etwas Telemetrie, dann einen Alarm, dann lauschen Sie auf Befehle:

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"

Ihr Telefon zeigt [HIGH] pump1: Temperature above 80 C. Senden Sie /status, um die zwischengespeicherte Telemetrie zu sehen, und /off pump1, um den Befehl im Fenster von mosquitto_sub erscheinen zu sehen. Auf echter Hardware abonniert das Gerät sein eigenes cmd-Topic; das ESP32 MQTT Tutorial zeigt die Geräteseite.

Sicherheitsregeln, die Sie nicht überspringen sollten

Ein Chatbot, der Anlagen schalten kann, ist eine Fernbedienung. Behandeln Sie ihn entsprechend:

1. Chat-IDs auf eine Allowlist setzen. Das Skript ignoriert jeden Chat, der nicht aufgeführt ist. Jeder kann Ihren Bot über den Namen finden, daher ist diese Prüfung Pflicht.
2. Das Geräteargument validieren. Der Code lehnt /, + und # ab, sodass niemand außerhalb des vorgesehenen Topics publizieren oder Wildcards einschleusen kann.
3. Das Token aus der Versionsverwaltung heraushalten. Verwenden Sie Umgebungsvariablen oder einen Secrets Manager und widerrufen Sie das Token bei BotFather, falls es durchsickert.
4. Niemals echte Anlagen über den öffentlichen MQTT-Broker steuern. Jeder kann auf öffentliche Topics publizieren. Verwenden Sie einen privaten Broker mit Zugangsdaten und TLS.
5. Bestätigung für gefährliche Aktionen ergänzen. Für alles, was Maschinen bewegt, verlangen Sie vor dem Publizieren eine zweite Nachricht wie /confirm. In industriellen Anlagen bleiben Sicherheitsfunktionen und Freigaben ohnehin in der SPS und ihren Verriegelungen; der Bot darf höchstens Anforderungen senden, die die Steuerung selbst prüft.
6. Datenschutz klären. Nachrichten laufen über die Server von Telegram. Stimmen Sie mit Ihrem Datenschutzbeauftragten ab, welche Daten nach DSGVO und internen Richtlinien in den Chat dürfen.

Die umfassendere Checkliste unter Best Practices für MQTT-Sicherheit gilt für den Bot genauso wie für Geräte.

Umgang mit Alarmstürmen

Ein flatternder Sensor kann Hunderte Alarme pro Minute senden, und Telegram drosselt Bots, die zu schnell Nachrichten senden. Das Skript wendet eine Abklingzeit pro Gerät an (standardmäßig fünf Minuten) und beachtet den Wert retry_after, den Telegram mit HTTP 429 zurückgibt. Außerdem überspringt es retained Alarme, damit ein Reconnect nicht den Alarm von gestern erneut sendet. Für andauernde Zustände lassen Sie das Gerät einen separaten "cleared"-Alarm publizieren, damit der Chatverlauf die ganze Geschichte erzählt.

Fehlerbehebung

• In Telegram kommen keine Nachrichten an. Prüfen Sie, ob TELEGRAM_CHAT_IDS Ihre numerische Chat-ID enthält und ob Sie dem Bot zuerst mindestens eine Nachricht gesendet haben; Bots können keine Unterhaltungen beginnen.
• 409 Conflict von getUpdates. Für den Bot ist ein Webhook gesetzt, oder eine zweite Kopie des Skripts läuft. Löschen Sie den Webhook mit der Methode deleteWebhook und betreiben Sie nur eine Instanz.
• Befehle funktionieren, Alarme nicht. Das Alarm-Topic passt nicht zu <prefix>/<device>/alert, oder die Abklingzeit ist für dieses Gerät noch aktiv.
• Der Bot verbindet sich wiederholt neu mit MQTT. Zwei Instanzen teilen sich eine Client ID. Halten Sie IDs pro Prozess eindeutig.

Den Bot als Dienst betreiben

Ein Bot, der nur läuft, solange Ihr Laptop aufgeklappt ist, ist kaum ein Alarmsystem. Auf einem Raspberry Pi oder einer kleinen Linux-VM betreiben Sie ihn unter systemd, damit er beim Booten startet und nach Abstürzen neu startet. Legen Sie die Secrets in eine Umgebungsdatei, die nur der Dienstbenutzer lesen kann:

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

Die Nachricht "Bot started", die das Skript beim Start verschickt, dient zugleich als Neustartmeldung: Sehen Sie sie unerwartet, ist der Dienst abgestürzt oder die Maschine neu gestartet, und journalctl verrät Ihnen den Grund. Bedenken Sie, dass der Bot ein Single Point of Failure für Chat-Alarme ist. Fällt er aus, meldet Ihnen das niemand. Kombinieren Sie ihn daher mit einem einfachen Heartbeat: Publizieren Sie vom Bot einen retained Status mit einem Last Will offline und lassen Sie ein anderes System dieses Topic überwachen.

Umstieg auf einen privaten ConnectMQ Broker

Sobald der Bot echte Geräte steuert, ziehen Sie ihn auf einen privaten Broker um. Erstellen Sie einen bei ConnectMQ, kopieren Sie den generierten Benutzernamen, das Passwort und die Client ID aus dem Portal und setzen Sie MQTT_HOST, MQTT_PORT=8883, MQTT_USERNAME, MQTT_PASSWORD und MQTT_CLIENT_ID. Mit einem verwalteten MQTT-Broker ist der Bot selbst der einzige Prozess, den Sie betreiben.

ConnectMQ bietet im Portal außerdem Schwellwert-Alarmregeln und ein Alarmprotokoll, mit denen Sie Alarme über Ihren Broker hinweg auf einen Blick sehen. Der Bot ergänzt das, indem er ausgewählte Alarme dorthin schickt, wo die Menschen ohnehin sind. Bevorzugt Ihr Team Postfächer statt Chat, funktioniert dieselbe Struktur mit SMTP unter MQTT-E-Mail-Benachrichtigungen mit Python. Für andere Sprachen und Frameworks vergleicht der Leitfaden zu MQTT-Client-Bibliotheken die Optionen.

Starten Sie mit ConnectMQ

Betreiben Sie Ihren Telegram-Bot noch heute gegen einen privaten, per TLS geschützten Broker. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan (ohne Kreditkarte), erstellen Sie einen Broker und tragen Sie die Zugangsdaten in die Umgebung des Bots ein. Der MQTT-Client im Browser des Portals eignet sich gut, um Testalarme auszulösen, während Sie die Abklingzeit feinjustieren.

Häufig gestellte Fragen

Brauche ich einen Server, um einen MQTT Telegram Bot zu betreiben?

Sie brauchen einen Ort, an dem das Skript dauerhaft läuft, etwa einen Raspberry Pi, eine kleine VM oder einen Container. Es baut nur ausgehende Verbindungen auf, daher sind keine offenen Ports und keine öffentliche IP nötig.

Können mehrere Personen Geräte über denselben Bot steuern?

Ja. Fügen Sie die Chat-ID jeder Person zu TELEGRAM_CHAT_IDS hinzu, oder fügen Sie den Bot einer Gruppe hinzu und setzen Sie die Gruppen-ID auf die Allowlist. Nehmen Sie wie das Skript den Absender in den Befehls-Payload auf, damit Geräte protokollieren können, wer den Befehl erteilt hat.

Warum Long Polling statt eines Telegram-Webhooks?

Webhooks erfordern einen öffentlichen HTTPS-Endpunkt. Long Polling funktioniert hinter NAT und Firewalls, was zu einem Bot passt, der auf einem Gateway oder Raspberry Pi neben Ihren Geräten läuft.

Werden Alarme zugestellt, wenn der Bot offline ist?

Alarme, die publiziert werden, während der Bot getrennt ist, gehen verloren, sofern Sie keine persistente Session mit QoS 1 nutzen. Halten Sie für kritische Alarme zusätzlich zum Chat einen zweiten Benachrichtigungsweg bereit.