Publicar en MQTT desde AWS Lambda y funciones serverless

Puedes publicar en un broker MQTT desde AWS Lambda (o cualquier función serverless) con una librería cliente MQTT normal; el problema es el ciclo de vida. Una función vive unos cientos de milisegundos, se congela entre invocaciones y puede ejecutarse en decenas de copias en paralelo. Conéctate de forma perezosa, reutiliza el cliente entre invocaciones en caliente, espera el acuse del publish antes de devolver y da a cada entorno de ejecución su propio client ID. Si solo publicas, un puente HTTP puede ser aún más sencillo.

Este tutorial muestra un handler en Python listo para producción (paho-mqtt 2.x), la misma idea en Node.js, la alternativa HTTP y cómo probarlo todo.

El ciclo de vida serverless frente a MQTT

MQTT asume conexiones TCP de larga duración con pings de keep-alive. Lambda asume ejecuciones cortas y sin estado. El desajuste aparece en cuatro sitios:

La congelación es lo más sutil. Si un entorno está inactivo más de aproximadamente 1,5× el intervalo de keep-alive, el broker cierra la conexión. Cuando el entorno se descongela, el cliente sigue creyendo que está conectado hasta que intenta escribir. Por eso un cliente cacheado de forma ingenua fallará o se colgará en la primera invocación tras un periodo tranquilo.

Estrategia 1: conectar en cada invocación

El patrón correcto más simple: conectar, publicar, esperar el PUBACK, desconectar. Nada sobrevive entre llamadas, así que nada puede quedarse obsoleto.

• Ventajas: sin bugs de conexiones obsoletas, sin colisiones de client ID si usas un ID aleatorio, fácil de razonar.
• Inconvenientes: cada invocación paga el handshake de conexión —bastante más con TLS y desde una región lejana— y una ráfaga de invocaciones se convierte en una ráfaga de conexiones nuevas en el broker.

Es una buena opción para disparadores de poco volumen: un job nocturno, una notificación de subida a S3, un webhook ocasional.

Estrategia 2: reutilizar la conexión entre invocaciones en caliente

Para cualquier cosa frecuente, crea el cliente en el ámbito global para que las invocaciones en caliente se salten el handshake, y desconfía de la conexión cacheada tras un periodo de inactividad.

import json, os, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "8883"))
KEEPALIVE = 60
MAX_IDLE = KEEPALIVE            # tras esta pausa, asume que el broker nos desconectó

_client = None
_connected = threading.Event()
_last_used = 0.0

def _on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect refused: {reason_code}")
    else:
        _connected.set()

def _on_disconnect(client, userdata, flags, reason_code, properties):
    _connected.clear()

def _new_client():
    c = mqtt.Client(
        mqtt.CallbackAPIVersion.VERSION2,
        client_id=f"lambda-{uuid.uuid4().hex[:12]}",   # único por entorno
        protocol=mqtt.MQTTv5,
    )
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.tls_set()                                         # bundle de CA del sistema
    c.on_connect, c.on_disconnect = _on_connect, _on_disconnect
    c.connect_async(HOST, PORT, keepalive=KEEPALIVE)
    c.loop_start()
    return c

def _get_client(timeout=5.0):
    global _client
    if _client is not None and time.time() - _last_used > MAX_IDLE:
        _client.disconnect()                            # obsoleta tras una congelación
        _client.loop_stop()
        _connected.clear()
        _client = None
    if _client is None:
        _client = _new_client()
    if not _connected.wait(timeout):
        raise TimeoutError("MQTT connect timed out")
    return _client

def handler(event, context):
    global _last_used
    client = _get_client()
    payload = json.dumps({"source": "lambda", "event": event, "ts": time.time()})
    info = client.publish("testmqtt/lambda/events", payload, qos=1)
    info.wait_for_publish(timeout=5)
    if not info.is_published():
        raise RuntimeError("PUBACK not received")       # deja que Lambda reintente
    _last_used = time.time()
    return {"statusCode": 200, "body": json.dumps({"mid": info.mid})}

Qué hace cada parte:

• connect_async + loop_start: el bucle de red corre en un hilo en segundo plano que gestiona CONNACK, PUBACK y las reconexiones mientras la función se ejecuta.
• MAX_IDLE: si el entorno estuvo inactivo más que el keep-alive, el código descarta el cliente y reconecta en lugar de escribir en un socket muerto.
• wait_for_publish: nunca devuelvas antes de que llegue el acuse de QoS 1. Cuando el handler devuelve, el entorno puede congelarse con el mensaje aún en el buffer del cliente.
• Lanzar una excepción si falla: con disparadores asíncronos, Lambda reintenta las invocaciones fallidas, que es justo lo que quieres ante un publish perdido.
• Client ID aleatorio: dos entornos concurrentes con el mismo ID se desconectarían mutuamente sin parar (client ID duplicado).

Empaquétalo con pip install paho-mqtt -t package/ y comprime la carpeta con el handler, o usa una Lambda layer. Pon el timeout de la función bastante por encima del timeout de conexión (10–15 s es un punto de partida razonable) y guarda las credenciales en variables de entorno respaldadas por Secrets Manager o Parameter Store.

El mismo patrón en Node.js

Con MQTT.js 5 el cliente cacheado es una variable a nivel de módulo. La librería reconecta sola, pero tras una congelación conviene verificar la conexión antes de publicar.

import mqtt from 'mqtt';
import { randomUUID } from 'node:crypto';

let client;
let lastUsed = 0;

async function getClient() {
  if (client && Date.now() - lastUsed > 60_000) { await client.endAsync(true); client = undefined; }
  if (!client) {
    client = await mqtt.connectAsync('mqtts://broker.connectmq.com:8883', {
      clientId: `lambda-${randomUUID().slice(0, 12)}`,
      username: process.env.MQTT_USER,
      password: process.env.MQTT_PASS,
      keepalive: 60,
      connectTimeout: 5000,
    });
  }
  return client;
}

export const handler = async (event) => {
  const c = await getClient();
  await c.publishAsync('testmqtt/lambda/events', JSON.stringify({ source: 'lambda', event }), { qos: 1 });
  lastUsed = Date.now();
  return { statusCode: 200 };
};

publishAsync con QoS 1 solo se resuelve tras el PUBACK, así que devolver después del await es seguro.

Estrategia 3: publicar por HTTP

Si tus funciones solo publican y no necesitan semántica MQTT en el cliente, una petición HTTP encaja de forma natural con serverless: sin estado de conexión, sin client IDs, sin problemas de congelación.

• APIs HTTP del broker. Algunos brokers exponen un endpoint REST que publica un mensaje por ti, por ejemplo la API HTTP de EMQX o el endpoint HTTPS del data plane de AWS IoT Core. Consulta la documentación de tu broker para la autenticación y el formato del payload.
• Tu propio puente. Un servicio pequeño siempre activo que mantiene una conexión MQTT y acepta peticiones POST /publish autenticadas. Centraliza las credenciales y convierte mil conexiones efímeras de Lambda en una sola persistente.

El precio: pierdes la QoS de extremo a extremo —el HTTP 200 te dice que el puente aceptó el mensaje, no que un subscriber lo recibiera— y añades un componente que operar. Para una comparación más amplia de ambos protocolos consulta MQTT vs HTTP.

¿Y suscribirse?

Una función Lambda no puede ser un subscriber MQTT de larga duración: solo se ejecuta mientras atiende una invocación. Invierte la dirección: deja que algo siempre activo se suscriba e invoque la función.

• Una regla o integración del broker que reenvía los mensajes coincidentes a un endpoint HTTP (API Gateway o una function URL de Lambda). Consulta MQTT a webhook para el patrón general.
• Un pequeño servicio suscriptor que empuja los mensajes a SQS, que a su vez dispara Lambda por lotes.

Errores habituales

• Task timed out after 3.00 seconds: el timeout por defecto de Lambda es más corto que una conexión TLS en frío desde otra región. Súbelo y comprueba que el puerto del broker es alcanzable.
• El connect se cuelga en una VPC: las funciones en una VPC necesitan un NAT gateway para salir a internet. Consulta timeout de conexión MQTT.
• Not authorized / reason code 0x87: credenciales incorrectas o una ACL que no permite el topic; la referencia de reason codes da el significado exacto.
• Faltan mensajes después de que la función devolviera: devolviste antes del PUBACK. Espera siempre a que el publish termine.

Probarlo

1. Abre el cliente MQTT online, conéctate a broker.connectmq.com y suscríbete a testmqtt/lambda/#.
2. Invoca la función desde la consola o con aws lambda invoke --function-name mqtt-publisher --payload '{"hello":"world"}' --cli-binary-format raw-in-base64-out out.json.
3. El mensaje debería aparecer en el navegador en menos de un segundo. Invoca de nuevo enseguida: la invocación en caliente debería ser claramente más rápida en la métrica de duración.
4. Espera unos minutos, invoca otra vez y confirma que la ruta de conexión obsoleta reconecta limpiamente en lugar de hacer timeout.

Mira llegar los mensajes de tu Lambda

El endpoint de pruebas compartido tiene límite de tasa y no guarda sesiones, lo cual basta para esta comprobación; para pruebas de carga con muchas invocaciones en paralelo, apunta la función a tu propia instancia privada de broker.

Serverless y MQTT funcionan bien juntos cuando respetas el ciclo de vida: conecta de forma perezosa, desconfía de las conexiones tras una congelación, espera el acuse y da a cada entorno su propia identidad. Cuando solo necesitas publicar sin más a gran escala, un puente HTTP elimina por completo la gestión de conexiones.

Preguntas frecuentes