Geräte per Telegram-Bot über MQTT steuern

Ein Telegram-Bot ist einer der schnellsten Wege zu einer Fernbedienung und einem Alarmkanal für MQTT-Geräte: Du bekommst eine kostenlose, plattformübergreifende Chat-Oberfläche, Push-Benachrichtigungen und musst keine App bauen. Der Bot läuft als kleiner Service, der zwei Dinge tut: Er published Befehle wie ON auf devices/lamp/set, wenn du /on lamp tippst, und er abonniert Alarm- und State-Topics und leitet sie in deinen Chat weiter.

Dieses Tutorial baut diese Brücke in Python mit python-telegram-bot und aiomqtt, beschränkt sie auf deine eigene Chat-ID und testet jeden Pfad mit mosquitto_pub und mosquitto_sub, bevor überhaupt Hardware ins Spiel kommt.

Architektur

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

Das Gerät spricht nie mit Telegram. Es spricht nur MQTT – ein ESP32 oder Raspberry Pi braucht also weder Bot-Token noch HTTPS-Client, und du kannst später ein Dashboard oder Home Assistant als zweiten Controller auf denselben Topics ergänzen. Das Topic-Schema folgt der üblichen Trennung von Befehl und Zustand aus Command and Control mit MQTT:

Bot anlegen

1. Öffne in Telegram einen Chat mit @BotFather, sende /newbot und wähle einen Namen sowie einen Username, der auf bot endet.
2. Kopiere das Token, das BotFather zurückgibt. Es ist geheim: Wer es hat, kann sich als dein Bot ausgeben.
3. Optional schickst du BotFather /setcommands mit on - switch a device on, off - switch a device off, status - show device states, damit die Befehle im Chat-Menü erscheinen.

Installiere die beiden Bibliotheken (Python 3.10+):

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

Beide sind asyncio-Bibliotheken, deshalb passen sie zusammen: Die Polling-Schleife des Bots und die MQTT-Verbindung laufen im selben Event Loop, und ein Command-Handler kann direkt await client.publish(...) aufrufen. Mehr zum Client findest du unter asynchrones MQTT in Python mit aiomqtt.

Der Bot

@@BOTCODE@@

Ein paar Designentscheidungen im Detail:

• Eine MQTT-Verbindung für den ganzen Bot. Der Worker besitzt die Verbindung und legt sie in bot_data ab; Handler verwenden sie wieder, statt pro Befehl neu zu verbinden. Fällt der Broker weg, verbindet sich die Schleife nach fünf Sekunden neu, und bis dahin antworten die Handler mit "Broker not connected".
• Zustände kommen aus Retained Messages. Weil Geräte state mit Retain-Flag publishen, kennt der Bot direkt nach dem Subscribe den letzten Zustand jedes Geräts – auch nach einem Neustart. Siehe Retained Messages.
• Unbekannte Chats bekommen Schweigen. Nur /whoami antwortet allen, damit du deine Chat-ID herausfinden kannst. Jeder Befehl, der published, prüft zuerst die Allowlist.
• Alarme werden entprellt. Sonst würde ein flatternder Sensor Telegrams Rate Limits reißen und echte Alarme untergehen lassen.

Ohne Hardware starten und testen

Starte den Bot nur mit TELEGRAM_TOKEN, sende /whoami im Chat und notier dir die Zahl. Dann neu starten mit Allowlist:

export TELEGRAM_TOKEN="123456:ABC..."        # von BotFather
export ALLOWED_CHAT_IDS="111111111"          # deine Chat-ID aus /whoami
export MQTT_USER="username" MQTT_PASS="password"   # falls dein Broker ein Login verlangt
python bot.py

Jetzt spielst du selbst das Gerät. Beobachte in einem Terminal, was der Bot published:

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

Sende /on lamp in Telegram. Terminal 1 zeigt testmqtt/tgbot/lamp/set ON. Melde den neuen Zustand wie ein Gerät und lös dann einen Alarm aus:

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 %)'

/status sollte jetzt lamp: ON zeigen, und im Chat kommt ⚠️ pump: Water level low (12 %) an. Publish den Alarm innerhalb von 60 Sekunden erneut und prüf, dass er unterdrückt wird. Das alles geht statt im Terminal auch mit dem Online-MQTT-Client. Lösch zum Schluss den Retained State, damit er nicht liegen bleibt: mosquitto_pub -h broker.connectmq.com -t testmqtt/tgbot/lamp/state -r -n.

Bot-Topics im Browser beobachten

Die Geräteseite

Auf echter Hardware abonniert das Gerät sein set-Topic, schaltet den Ausgang und published den resultierenden Zustand mit Retain. In MicroPython mit umqtt.simple sind das nur ein paar Zeilen:

@@DEVCODE@@

Den Zustand nach dem Schalten zu publishen macht /status vertrauenswürdig: Der Bot meldet, was das Gerät getan hat, nicht, was du angefordert hast. Ergänze ein Last Will auf dem State oder ein availability-Topic, damit der Bot dir sagen kann, wenn ein Gerät offline geht.

Absichern

Jeder auf Telegram kann einen Bot erreichen, behandle ihn also wie eine öffentliche API:

1. Chats per Allowlist freigeben, wie gezeigt. Bei Gruppen ist die Chat-ID negativ; trag sie wie jede andere ein.
2. Token raus aus dem Code – Umgebungsvariablen oder ein Secret Manager, niemals ein Git-Repository.
3. TLS zum Broker nutzen. Das Beispiel verbindet sich auf Port 8883; unverschlüsseltes 1883 ist nur für lokale Experimente okay.
4. Den Broker-User des Bots einschränken. Er braucht Schreibrechte auf .../set und Leserechte auf .../state und .../alert, sonst nichts. Prüf die Regeln mit einem ACL-Berechtigungstest.

Alternativen ohne Code

Wenn du ein visuelles Tool bevorzugst, funktioniert dieselbe Brücke in n8n (MQTT Trigger → Telegram-Node und Telegram Trigger → MQTT-Node – siehe MQTT-Automatisierung mit n8n) oder in Node-RED mit einem Community-Telegram-Node. Die Python-Variante gewinnt, wenn du einen einzelnen kleinen Service mit expliziter Allowlist, Entprellung und Tests im eigenen Repository willst.

Ein Telegram-Bot macht MQTT zu etwas, das du aus der Hosentasche steuerst: Befehle gehen als Publish raus, der Zustand kommt als Retained Message zurück und Alarme landen als Push-Benachrichtigung. Lass das Gerät nur MQTT sprechen, beschränke den Bot auf deinen Chat, entprelle Alarme und teste die ganze Schleife mit mosquitto, bevor du ein Relais anschließt.

Häufig gestellte Fragen