Build an MQTT Telegram Bot for Device Alerts and Commands

An MQTT Telegram bot turns your phone into a pocket console for your devices: alarms arrive as chat messages, and a quick /off pump1 publishes a command back to the device. In this tutorial you will build that bot in Python with paho-mqtt and the Telegram Bot HTTP API, protect it with a chat allowlist, and run it against a real broker. The whole thing is one script with two dependencies.

How the Bot Fits Into Your MQTT System

The bot is just another MQTT client. Devices keep publishing and subscribing exactly as before, and the bot sits between the broker and Telegram:

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

Nothing on the device side knows Telegram exists. That separation matters: you can swap Telegram for email or Slack later without touching firmware, and you never put a bot token on a microcontroller. If you want a refresher on what an MQTT broker does in this picture, it simply routes messages between topics and clients.

Topic Design for Alerts and Commands

Use a predictable structure so the bot can parse device names from topics:

The prefix connectmq/demo/tgbot-9214 is a random namespace for testing on the free public MQTT broker broker.connectmq.com. The public area can be read by anyone subscribed to the same topic, so pick your own random suffix and use the bot only with test devices until you move to a private broker.

Step 1: Create the Bot and Find Your Chat ID

1. In Telegram, open a chat with @BotFather, send /newbot and follow the prompts. You receive a token like 123456789:AA.... Treat it like a password.
2. Open a chat with your new bot and send it any message, for example hi.
3. In a browser or terminal, call getUpdates to find your chat ID:

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

Look for "chat":{"id":123456789,...} in the response. That number goes into the allowlist. For a group, add the bot to the group and use the group's negative chat ID. The Telegram Bot API documentation describes every method used below.

Step 2: Install Dependencies

pip install "paho-mqtt>=2.0" requests

The script uses plain HTTPS calls instead of a bot framework. That keeps threading simple: paho runs its network loop in a background thread, and the main thread long-polls Telegram for commands.

Step 3: The MQTT Telegram Bot

Save as 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"))  # seconds per device

latest = {}            # device -> (timestamp, payload dict)
last_alert = {}        # device -> timestamp of last forwarded alert
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 side ----------

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))
        # send from a separate thread so a slow HTTP call never blocks the MQTT loop
        threading.Thread(target=broadcast, args=(f"[{level}] {device}: {text}",), daemon=True).start()


# ---------- Telegram side ----------

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

Step 4: Run It

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

Simulate a device in another terminal. First some telemetry, then an alert, then listen for commands:

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"

Your phone shows [HIGH] pump1: Temperature above 80 C. Send /status to see the cached telemetry and /off pump1 to watch the command appear in the mosquitto_sub window. On real hardware, the device subscribes to its own cmd topic; the ESP32 MQTT tutorial shows the device side.

Security Rules You Should Not Skip

A chat bot that can switch equipment is a remote control. Treat it accordingly:

1. Allowlist chat IDs. The script ignores every chat that is not listed. Anyone can find your bot by name, so this check is mandatory.
2. Validate the device argument. The code rejects /, + and #, so nobody can publish outside the intended topic or inject wildcards.
3. Keep the token out of source control. Use environment variables or a secrets manager, and revoke the token with BotFather if it leaks.
4. Never control real equipment through the public broker. Anyone can publish to public topics. Use a private broker with credentials and TLS.
5. Add confirmation for dangerous actions. For anything that moves machinery, require a second message such as /confirm before publishing.

The broader checklist in MQTT security best practices applies to the bot just as it does to devices.

Handling Alert Storms

A flapping sensor can send hundreds of alerts per minute, and Telegram rate-limits bots that message too fast. The script applies a per-device cooldown (five minutes by default) and honours the retry_after value Telegram returns with HTTP 429. It also skips retained alerts, so a reconnect does not re-send yesterday's alarm. For persistent conditions, have the device publish a separate "cleared" alert so the chat history tells the full story.

Troubleshooting

• No messages arrive in Telegram. Check that TELEGRAM_CHAT_IDS contains your numeric chat ID and that you sent the bot at least one message first; bots cannot start conversations.
• 409 Conflict from getUpdates. A webhook is set for the bot, or a second copy of the script is running. Delete the webhook with the deleteWebhook method and run only one instance.
• Commands work, alerts do not. The alert topic does not match <prefix>/<device>/alert, or the cooldown is still active for that device.
• Bot reconnects to MQTT repeatedly. Two instances share a client ID. Keep IDs unique per process.

Running the Bot as a Service

A bot that only runs while your laptop is open is not much of an alerting system. On a Raspberry Pi or a small Linux VM, run it under systemd so it starts at boot and restarts after crashes. Put the secrets in an environment file readable only by the service user:

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

The "Bot started" message the script broadcasts on startup doubles as a restart notification: if you see it unexpectedly, the service crashed or the machine rebooted, and journalctl tells you why. Keep in mind that the bot is a single point of failure for chat alerts. If it goes down, nothing tells you, so pair it with a simple heartbeat: publish a retained status from the bot with a Last Will of offline, and let another system watch that topic.

Moving to a Private ConnectMQ Broker

When the bot controls real devices, move it to a private broker. Create one on ConnectMQ, copy the generated username, password and client ID from the portal, and set MQTT_HOST, MQTT_PORT=8883, MQTT_USERNAME, MQTT_PASSWORD and MQTT_CLIENT_ID. With a hosted MQTT broker, the only process you operate is the bot itself.

ConnectMQ also offers threshold alert rules and an alerts log in the portal, which are useful for seeing alarms across your broker at a glance. The bot complements that by pushing selected alerts to people where they already are. If your team prefers inboxes over chat, the same structure works with SMTP in MQTT email alerts with Python. For other languages and frameworks, the MQTT client libraries guide compares the options.

Get Started with ConnectMQ

Run your Telegram bot against a private, TLS-protected broker today. Sign up at app.connectmq.com, choose the free Serverless plan (no credit card), create a broker and paste the credentials into the bot's environment. The in-browser MQTT client in the portal is a handy way to fire test alerts while you tune the cooldown.

Frequently Asked Questions

Do I need a server to run an MQTT Telegram bot?

You need somewhere for the script to run continuously, such as a Raspberry Pi, a small VM or a container. It only makes outbound connections, so no open ports or public IP are required.

Can several people control devices through the same bot?

Yes. Add each person's chat ID to TELEGRAM_CHAT_IDS, or add the bot to a group and allowlist the group ID. Include the sender in the command payload, as the script does, so devices can log who issued it.

Why use long polling instead of a Telegram webhook?

Webhooks require a public HTTPS endpoint. Long polling works from behind NAT and firewalls, which suits a bot running on a gateway or Raspberry Pi next to your devices.

Will alerts be delivered if the bot is offline?

Alerts published while the bot is disconnected are missed unless you use a persistent session with QoS 1. For critical alarms, keep a second notification path in addition to chat.