Control Devices from a Telegram Bot via MQTT

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

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

import asyncio
import os
import ssl
import time

import aiomqtt
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

TOKEN = os.environ["TELEGRAM_TOKEN"]
ALLOWED = {int(x) for x in os.environ.get("ALLOWED_CHAT_IDS", "").split(",") if x}
BROKER, PORT = "broker.connectmq.com", 8883
BASE = "testmqtt/tgbot"
DEVICES = {"lamp", "pump", "fan"}
COOLDOWN = 60  # seconds between alerts of the same device

states: dict[str, str] = {}
last_alert: dict[str, float] = {}


async def mqtt_worker(app: Application) -> None:
    """Keep one MQTT connection open, cache states, forward alerts."""
    while True:
        try:
            async with aiomqtt.Client(
                BROKER,
                PORT,
                username=os.environ.get("MQTT_USER"),
                password=os.environ.get("MQTT_PASS"),
                tls_context=ssl.create_default_context(),
            ) as client:
                app.bot_data["mqtt"] = client
                await client.subscribe(f"{BASE}/+/state", qos=1)
                await client.subscribe(f"{BASE}/+/alert", qos=1)
                async for msg in client.messages:
                    _, _, device, kind = msg.topic.value.split("/")
                    text = msg.payload.decode(errors="replace")
                    if kind == "state":
                        states[device] = text
                        continue
                    now = time.monotonic()
                    if now - last_alert.get(device, float("-inf")) < COOLDOWN:
                        continue  # debounce noisy sensors
                    last_alert[device] = now
                    for chat_id in ALLOWED:
                        await app.bot.send_message(chat_id, f"⚠️ {device}: {text}")
        except aiomqtt.MqttError as err:
            app.bot_data.pop("mqtt", None)
            print(f"MQTT connection lost ({err}), retrying in 5 s")
            await asyncio.sleep(5)


def allowed(update: Update) -> bool:
    return update.effective_chat is not None and update.effective_chat.id in ALLOWED


def make_switch(payload: str):
    async def handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not allowed(update):
            return
        if not context.args or context.args[0] not in DEVICES:
            await update.message.reply_text(f"Usage: /{payload.lower()} {'|'.join(sorted(DEVICES))}")
            return
        client = context.bot_data.get("mqtt")
        if client is None:
            await update.message.reply_text("Broker not connected, try again in a few seconds.")
            return
        device = context.args[0]
        await client.publish(f"{BASE}/{device}/set", payload, qos=1)
        await update.message.reply_text(f"Sent {payload} to {device}")

    return handler


async def status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not allowed(update):
        return
    lines = [f"{d}: {states.get(d, 'unknown')}" for d in sorted(DEVICES)]
    await update.message.reply_text("\n".join(lines))


async def whoami(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(f"Your chat ID is {update.effective_chat.id}")


async def post_init(app: Application) -> None:
    app.bot_data["worker"] = asyncio.create_task(mqtt_worker(app))


async def post_shutdown(app: Application) -> None:
    app.bot_data["worker"].cancel()


def main() -> None:
    app = (
        Application.builder()
        .token(TOKEN)
        .post_init(post_init)
        .post_shutdown(post_shutdown)
        .build()
    )
    app.add_handler(CommandHandler("on", make_switch("ON")))
    app.add_handler(CommandHandler("off", make_switch("OFF")))
    app.add_handler(CommandHandler("status", status))
    app.add_handler(CommandHandler("whoami", whoami))
    app.run_polling()


if __name__ == "__main__":
    main()

export TELEGRAM_TOKEN="123456:ABC..."        # from BotFather
export ALLOWED_CHAT_IDS="111111111"          # your chat ID from /whoami
export MQTT_USER="username" MQTT_PASS="password"   # if your broker requires a login
python bot.py

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

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

from machine import Pin
from umqtt.simple import MQTTClient

lamp = Pin(2, Pin.OUT)
BASE = b"testmqtt/tgbot/lamp"

def on_msg(topic, msg):
    lamp.value(1 if msg == b"ON" else 0)
    client.publish(BASE + b"/state", msg, retain=True, qos=1)

client = MQTTClient("lamp-7f3a", "broker.connectmq.com", port=1883)
client.set_callback(on_msg)
client.connect()
client.subscribe(BASE + b"/set", qos=1)
while True:
    client.wait_msg()