Crea un bot de Telegram con MQTT para alertas y comandos de dispositivos

Un bot de Telegram con MQTT convierte tu teléfono en una consola de bolsillo para tus dispositivos: las alarmas llegan como mensajes de chat, y un rápido /off pump1 publica un comando de vuelta al dispositivo. En este tutorial construirás ese bot en Python con paho-mqtt y la API HTTP de bots de Telegram, lo protegerás con una lista de chats permitidos y lo ejecutarás contra un broker real. Todo cabe en un solo script con dos dependencias.

Cómo encaja el bot en tu sistema MQTT

El bot es simplemente otro cliente MQTT. Los dispositivos siguen publicando y suscribiéndose exactamente igual que antes, y el bot se sitúa entre el broker y Telegram:

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

Nada en el lado del dispositivo sabe que Telegram existe. Esa separación es importante: más adelante puedes cambiar Telegram por correo o Slack sin tocar el firmware, y nunca pones un token de bot en un microcontrolador. Si quieres repasar qué hace un broker MQTT en este esquema, simplemente enruta mensajes entre topics y clientes.

Diseño de topics para alertas y comandos

Usa una estructura predecible para que el bot pueda extraer los nombres de los dispositivos de los topics:

El prefijo connectmq/demo/tgbot-9214 es un espacio de nombres aleatorio para hacer pruebas en el servidor MQTT público gratuito broker.connectmq.com, pensado para pruebas y prototipos, nunca para datos de producción. Cualquiera suscrito al mismo topic puede leer el área pública, así que elige tu propio sufijo aleatorio y usa el bot solo con dispositivos de prueba hasta que pases a un broker privado.

Paso 1: crear el bot y encontrar tu chat ID

1. En Telegram, abre un chat con @BotFather, envía /newbot y sigue las indicaciones. Recibirás un token como 123456789:AA.... Trátalo como una contraseña.
2. Abre un chat con tu nuevo bot y envíale cualquier mensaje, por ejemplo hola.
3. En un navegador o una terminal, llama a getUpdates para encontrar tu chat ID:

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

Busca "chat":{"id":123456789,...} en la respuesta. Ese número va a la lista de permitidos. Para un grupo, añade el bot al grupo y usa el chat ID negativo del grupo. La documentación de la Telegram Bot API describe todos los métodos que se usan a continuación.

Paso 2: instalar las dependencias

pip install "paho-mqtt>=2.0" requests

El script usa llamadas HTTPS simples en lugar de un framework de bots. Así el manejo de hilos es sencillo: paho ejecuta su bucle de red en un hilo en segundo plano, y el hilo principal hace long polling a Telegram en busca de comandos.

Paso 3: el bot de Telegram con MQTT

Guárdalo como 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"))  # segundos por dispositivo

latest = {}            # dispositivo -> (marca de tiempo, dict del payload)
last_alert = {}        # dispositivo -> marca de tiempo de la última alerta reenviada
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)


# ---------- Lado 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))
        # enviar desde otro hilo para que una llamada HTTP lenta nunca bloquee el bucle MQTT
        threading.Thread(target=broadcast, args=(f"[{level}] {device}: {text}",), daemon=True).start()


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

Paso 4: ejecútalo

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

Simula un dispositivo en otra terminal. Primero algo de telemetría, después una alerta, y luego escucha los comandos:

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"

Tu teléfono muestra [HIGH] pump1: Temperature above 80 C. Envía /status para ver la telemetría en caché y /off pump1 para ver aparecer el comando en la ventana de mosquitto_sub. En hardware real, el dispositivo se suscribe a su propio topic cmd; el tutorial de ESP32 con MQTT muestra el lado del dispositivo.

Reglas de seguridad que no debes saltarte

Un bot de chat capaz de accionar equipos es un control remoto. Trátalo como tal:

1. Lista de chat IDs permitidos. El script ignora cualquier chat que no esté en la lista. Cualquiera puede encontrar tu bot por su nombre, así que esta comprobación es obligatoria.
2. Valida el argumento del dispositivo. El código rechaza /, + y #, de modo que nadie puede publicar fuera del topic previsto ni inyectar comodines.
3. Mantén el token fuera del control de versiones. Usa variables de entorno o un gestor de secretos, y revoca el token con BotFather si se filtra.
4. Nunca controles equipos reales a través del broker público. Cualquiera puede publicar en los topics públicos. Usa un broker privado con credenciales y TLS.
5. Añade una confirmación para las acciones peligrosas. Para cualquier cosa que mueva maquinaria, exige un segundo mensaje como /confirm antes de publicar.

La lista de comprobación más amplia de buenas prácticas de seguridad en MQTT se aplica al bot igual que a los dispositivos.

Gestionar tormentas de alertas

Un sensor que oscila puede enviar cientos de alertas por minuto, y Telegram limita la tasa de los bots que envían mensajes demasiado rápido. El script aplica un tiempo de espera por dispositivo (cinco minutos por defecto) y respeta el valor retry_after que Telegram devuelve con HTTP 429. También omite las alertas retenidas, de modo que una reconexión no vuelve a enviar la alarma de ayer. Para condiciones persistentes, haz que el dispositivo publique una alerta "cleared" aparte para que el historial del chat cuente la historia completa.

Solución de problemas

• No llega ningún mensaje a Telegram. Comprueba que TELEGRAM_CHAT_IDS contiene tu chat ID numérico y que has enviado antes al menos un mensaje al bot; los bots no pueden iniciar conversaciones.
• 409 Conflict en getUpdates. El bot tiene un webhook configurado, o hay una segunda copia del script en ejecución. Elimina el webhook con el método deleteWebhook y ejecuta una sola instancia.
• Los comandos funcionan, las alertas no. El topic de alerta no coincide con <prefix>/<device>/alert, o el tiempo de espera sigue activo para ese dispositivo.
• El bot se reconecta a MQTT una y otra vez. Dos instancias comparten un client ID. Mantén IDs únicos por proceso.

Ejecutar el bot como servicio

Un bot que solo funciona mientras tu portátil está abierto no es gran cosa como sistema de alertas. En una Raspberry Pi o una pequeña VM Linux, ejecútalo bajo systemd para que arranque con el sistema y se reinicie tras un fallo. Pon los secretos en un archivo de entorno legible solo por el usuario del servicio:

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

El mensaje "Bot started" que el script difunde al arrancar sirve también como aviso de reinicio: si lo ves de forma inesperada, el servicio se bloqueó o la máquina se reinició, y journalctl te dice por qué. Ten en cuenta que el bot es un punto único de fallo para las alertas por chat. Si se cae, nada te avisa, así que combínalo con un simple heartbeat: publica desde el bot un estado retenido con un Last Will de offline y deja que otro sistema vigile ese topic.

Pasar a un broker privado de ConnectMQ

Cuando el bot controle dispositivos reales, pásalo a un broker privado. Crea uno en ConnectMQ, copia del portal el usuario, la contraseña y el client ID generados, y define MQTT_HOST, MQTT_PORT=8883, MQTT_USERNAME, MQTT_PASSWORD y MQTT_CLIENT_ID. Con un broker MQTT gestionado, el único proceso que operas es el propio bot.

ConnectMQ también ofrece en el portal reglas de alerta por umbral y un registro de alertas, útiles para ver de un vistazo las alarmas de todo tu broker. El bot lo complementa llevando alertas seleccionadas a las personas allí donde ya están. Si tu equipo prefiere el correo al chat, la misma estructura funciona con SMTP en alertas MQTT por correo con Python. Para otros lenguajes y frameworks, la guía de bibliotecas cliente MQTT compara las opciones.

Empieza con ConnectMQ

Ejecuta hoy mismo tu bot de Telegram contra un broker privado protegido con TLS. Regístrate en app.connectmq.com, elige el plan gratuito Serverless (sin tarjeta de crédito), crea un broker y pega las credenciales en el entorno del bot. El cliente MQTT en el navegador del portal es una forma práctica de lanzar alertas de prueba mientras ajustas el tiempo de espera.

Preguntas frecuentes

¿Necesito un servidor para ejecutar un bot de Telegram con MQTT?

Necesitas algún sitio donde el script se ejecute de forma continua, como una Raspberry Pi, una pequeña VM o un contenedor. Solo establece conexiones salientes, así que no hacen falta puertos abiertos ni IP pública.

¿Pueden varias personas controlar dispositivos con el mismo bot?

Sí. Añade el chat ID de cada persona a TELEGRAM_CHAT_IDS, o añade el bot a un grupo y permite el ID del grupo. Incluye al remitente en el payload del comando, como hace el script, para que los dispositivos puedan registrar quién lo envió.

¿Por qué usar long polling en lugar de un webhook de Telegram?

Los webhooks requieren un endpoint HTTPS público. El long polling funciona detrás de NAT y firewalls, lo que encaja con un bot que se ejecuta en una pasarela o una Raspberry Pi junto a tus dispositivos.

¿Se entregarán las alertas si el bot está desconectado?

Las alertas publicadas mientras el bot está desconectado se pierden, salvo que uses una sesión persistente con QoS 1. Para alarmas críticas, mantén una segunda vía de notificación además del chat.