Node-RED HTTP zu MQTT: REST-Endpunkte, die an Ihren Broker publizieren

Mit einer HTTP-zu-MQTT-Brücke in Node-RED kann jedes System, das eine Web-Anfrage senden kann, Befehle auf MQTT-Geräten auslösen, ohne dass dieses System eine MQTT-Bibliothek braucht. In diesem Tutorial stellen Sie in Node-RED einen REST-Endpunkt bereit, der die Anfrage validiert, einen Befehl an einen Broker publiziert, optional auf die Bestätigung des Geräts wartet und einen korrekten HTTP-Statuscode zurückgibt.

Wann eine HTTP-zu-MQTT-Brücke sinnvoll ist

MQTT ist das richtige Protokoll zwischen Geräten und Broker, doch viele Programme rund um ein Werk oder ein Gebäude beherrschen nur HTTP:

• Ein ERP- oder MES-System, das beim Start eines Fertigungsauftrags eine URL aufrufen kann.
• Ein Webhook aus einem SaaS-Produkt, von einem Zahlungsanbieter oder einem Formular-Tool.
• Shell-Skripte und Cronjobs, die bereits curl verwenden.
• Mobile Apps oder Dashboards, die keine langlebigen MQTT-Verbindungen halten sollen.
• Ein externer Integrator, der nur einen REST-Vertrag akzeptiert.

Statt jedem dieser Systeme MQTT beizubringen, geben Sie ihnen eine kleine, klar definierte HTTP-API. Node-RED übersetzt jede Anfrage in ein Publish auf dem richtigen Topic. Die Geräte bleiben reine MQTT-Clients, und der Broker bleibt der einzige Ort, an dem Nachrichten fließen.

Architektur des Flows

HTTP client --POST /api/devices/pump1/command--> Node-RED
Node-RED --publish connectmq/demo/h7r2pq/pump1/cmd--> MQTT broker --> device
device --publish connectmq/demo/h7r2pq/pump1/ack--> MQTT broker --> Node-RED
Node-RED --HTTP 200 or 202 or 504--> HTTP client

Wir bauen drei Endpunkte:

Für die Demo nutzen wir den kostenlosen öffentlichen MQTT-Broker unter broker.connectmq.com mit dem Präfix connectmq/demo/h7r2pq. Verwenden Sie ein eigenes zufälliges Präfix und senden Sie niemals echte Steuerbefehle über den öffentlichen Bereich, denn dort kann jeder mitlesen und publizieren. Der öffentliche MQTT-Broker ist für Tests und Prototypen gedacht, nicht für Produktionsdaten.

Schritt 1: Der HTTP-Endpunkt und die Validierung

Fügen Sie einen http in-Node mit der Methode POST und der URL /api/devices/:id/command hinzu. Node-RED parst JSON-Bodys automatisch, wenn der Client Content-Type: application/json sendet, und legt Pfadparameter in msg.req.params ab.

Verbinden Sie ihn mit einem Function-Node mit zwei Ausgängen: Der erste führt zu MQTT, der zweite gibt Fehler direkt an einen http response-Node zurück.

const PREFIX = "connectmq/demo/h7r2pq";
const ALLOWED_DEVICES = ["pump1", "pump2", "valve3"];
const ALLOWED_ACTIONS = ["start", "stop", "setpoint"];
const API_KEY = env.get("BRIDGE_API_KEY");   // im Flow oder in der Umgebung setzen

function fail(code, message) {
    msg.statusCode = code;
    msg.payload = { error: message };
    return [null, msg];
}

if (!API_KEY || msg.req.headers["x-api-key"] !== API_KEY) {
    return fail(401, "invalid or missing API key");
}

const id = msg.req.params.id;
if (!ALLOWED_DEVICES.includes(id)) return fail(404, "unknown device " + id);

const body = msg.payload;
if (typeof body !== "object" || body === null) return fail(400, "JSON body required");
if (!ALLOWED_ACTIONS.includes(body.action)) return fail(400, "action must be one of " + ALLOWED_ACTIONS.join(", "));
if (body.action === "setpoint" && typeof body.value !== "number") {
    return fail(400, "setpoint requires a numeric value");
}

const correlationId = Date.now().toString(36) + Math.random().toString(36).slice(2, 8);
const command = {
    action: body.action,
    value: body.value,
    correlationId,
    issuedAt: new Date().toISOString(),
    source: "http"
};

msg.deviceId = id;
msg.correlationId = correlationId;
msg.wait = msg.req.query.wait === "1";
msg.topic = `${PREFIX}/${id}/cmd`;
msg.payload = JSON.stringify(command);
msg.qos = 1;
msg.retain = false;
return [msg, null];

Bauen Sie das MQTT-Topic niemals direkt aus Benutzereingaben zusammen, ohne es gegen eine Allow-List zu prüfen. Eine Anfrage für das Gerät # oder ../../other würde sonst auf Topics publizieren, die Sie nie freigeben wollten.

Schritt 2: Publizieren und antworten

Verbinden Sie den ersten Ausgang mit zwei Zielen:

1. Einem mqtt out-Node mit leerem Topic-Feld, damit er msg.topic verwendet, verbunden mit Ihrem Broker-Konfigurations-Node.
2. Einem Function-Node, der entscheidet, wie dem HTTP-Aufrufer geantwortet wird:

if (msg.wait) {
    // Anfrage parken, bis das Gerät bestätigt oder der Timer abläuft
    const pending = flow.get("pending") || {};
    pending[msg.correlationId] = msg;
    flow.set("pending", pending);

    setTimeout(() => {
        const p = flow.get("pending") || {};
        const waiting = p[msg.correlationId];
        if (waiting) {
            delete p[msg.correlationId];
            flow.set("pending", p);
            waiting.statusCode = 504;
            waiting.payload = { error: "device did not acknowledge in time", correlationId: msg.correlationId };
            node.send(waiting);
        }
    }, 5000);
    return null;
}

msg.statusCode = 202;
msg.payload = { status: "accepted", correlationId: msg.correlationId, topic: msg.topic };
return msg;

Verbinden Sie diese Funktion und den Fehlerausgang aus Schritt 1 mit demselben http response-Node. Der Response-Node verwendet msg.statusCode und sendet msg.payload als JSON.

Die asynchrone 202-Antwort ist der richtige Standard. MQTT-Geräte können schlafen, sich gerade neu verbinden oder beschäftigt sein, und ein HTTP-Client sollte deshalb nicht hängen bleiben. Die Variante wait=1 ist nützlich für interaktive Werkzeuge, bei denen jemand einen Button klickt und wissen möchte, ob die Pumpe tatsächlich angelaufen ist.

Schritt 3: Gerätebestätigungen verarbeiten

Das Gerät publiziert eine Bestätigung, die die Correlation-ID zurückgibt, zum Beispiel auf connectmq/demo/h7r2pq/pump1/ack:

{ "correlationId": "lrx2k9ab12cd", "result": "ok", "state": "running" }

Abonnieren Sie mit einem mqtt in-Node connectmq/demo/h7r2pq/+/ack (Ausgabe: geparstes JSON) und ordnen Sie die Nachricht der geparkten Anfrage zu:

const ack = msg.payload;
const pending = flow.get("pending") || {};
const waiting = pending[ack.correlationId];
if (!waiting) return null;   // bereits abgelaufen oder eine 202-Anfrage

delete pending[ack.correlationId];
flow.set("pending", pending);

waiting.statusCode = ack.result === "ok" ? 200 : 409;
waiting.payload = { status: ack.result, state: ack.state, correlationId: ack.correlationId };
return waiting;

Verbinden Sie ihn mit demselben http response-Node. Da die geparkte Nachricht noch das ursprüngliche Objekt msg.res enthält, geht die Antwort an den richtigen HTTP-Aufrufer zurück. Offene Anfragen liegen im Arbeitsspeicher. Das ist unproblematisch, denn die zugehörigen HTTP-Verbindungen verschwinden bei einem Neustart von Node-RED ebenfalls.

Wenn Sie MQTT 5 nutzen, können Sie auf dem ausgehenden Befehl msg.responseTopic und msg.correlationData setzen, statt die ID ins JSON einzubetten. Der mqtt out-Node überträgt sie als MQTT-5-Properties, und Geräte, die diese unterstützen, können auf dem angeforderten Topic antworten. Der Leitfaden zu den MQTT-5-Funktionen erklärt Request/Response ausführlicher.

Schritt 4: Ein Lese-Endpunkt für den Gerätezustand

Geräte sollten ihren Zustand als Retained Messages auf connectmq/demo/h7r2pq/<id>/state publizieren. Abonnieren Sie connectmq/demo/h7r2pq/+/state und speichern Sie jeden Zustand im Flow-Kontext:

const id = msg.topic.split("/")[3];
const states = flow.get("states") || {};
states[id] = { state: msg.payload, receivedAt: new Date().toISOString() };
flow.set("states", states);
return null;

Fügen Sie dann einen http in-Node für GET /api/devices/:id/state hinzu:

const states = flow.get("states") || {};
const s = states[msg.req.params.id];
if (!s) {
    msg.statusCode = 404;
    msg.payload = { error: "no state received for this device" };
} else {
    msg.statusCode = 200;
    msg.payload = s;
}
return msg;

Da die Zustandsnachrichten retained sind, erhält Node-RED den letzten Wert jedes Geräts sofort nach dem Wiederverbinden. Der Lese-Endpunkt ist also direkt nach einem Neustart einsatzbereit.

Test mit curl

Setzen Sie den API-Key als Umgebungsvariable für Node-RED (BRIDGE_API_KEY=change-me), deployen Sie und rufen Sie die Endpunkte auf dem Standardport 1880 auf:

# Fire and forget
curl -i -X POST http://localhost:1880/api/devices/pump1/command \
  -H "Content-Type: application/json" -H "X-API-Key: change-me" \
  -d '{"action":"start"}'

# Auf Bestätigung warten (liefert nach 5 s 504, wenn kein Gerät antwortet)
curl -i -X POST "http://localhost:1880/api/devices/pump1/command?wait=1" \
  -H "Content-Type: application/json" -H "X-API-Key: change-me" \
  -d '{"action":"setpoint","value":42.5}'

# Bestätigung des Geräts simulieren (correlationId aus dem cmd-Topic kopieren)
mosquitto_sub -h broker.connectmq.com -p 1883 -i dev-h7r2pq -v -t "connectmq/demo/h7r2pq/pump1/cmd"
mosquitto_pub -h broker.connectmq.com -p 1883 -i ack-h7r2pq -t "connectmq/demo/h7r2pq/pump1/ack" \
  -m '{"correlationId":"PASTE_ID","result":"ok","state":"running"}'

Eine Node-RED-Brücke von HTTP zu MQTT absichern

Ein HTTP-Endpunkt, der Pumpen bewegt, ist eine Angriffsfläche. Arbeiten Sie diese Liste ab, bevor Sie ihn über localhost hinaus erreichbar machen:

1. Jede Anfrage authentifizieren. Die obige API-Key-Prüfung ist das Minimum. Für Basic Authentication über alle HTTP-in-Endpunkte setzen Sie httpNodeAuth in settings.js mit einem bcrypt-Hash, den Sie mit node-red admin hash-pw erzeugen.
2. TLS terminieren vor Node-RED mit einem Reverse Proxy wie Nginx oder Caddy. Geben Sie Port 1880 nicht direkt ins Internet frei.
3. Den Editor schützen mit adminAuth, oder betreiben Sie den Editor auf einem anderen, nur intern erreichbaren Pfad oder Port.
4. Eingaben validieren gegen Allow-Lists für Geräte und Aktionen, und Werte auf ihren Typ prüfen.
5. Rate-Limiting am Proxy einrichten, und bedenken, dass auch der Broker Limits durchsetzt. Der öffentliche Broker erlaubt 10 Nachrichten pro Sekunde und Client, Produktionspläne haben eigene Kontingente.
6. Private Broker-Zugangsdaten über TLS auf Port 8883 für die mqtt-Nodes verwenden.

Unser Artikel zum Absichern von Node-RED behandelt die Editor- und Settings-Seite im Detail.

Umstieg auf einen Produktions-Broker

Ersetzen Sie die Demo-Broker-Konfiguration durch einen privaten ConnectMQ-Broker: Port 8883, TLS aktiviert, und die im Portal erzeugten Werte YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID. ConnectMQ bietet weder Webhooks noch eine eigene HTTP-Publish-API, und genau deshalb gibt es dieses Muster: Node-RED ist die HTTP-Schicht, der Broker bleibt auf MQTT fokussiert. Betreiben Sie den Broker als verwalteten MQTT-Broker, müssen Sie sich nur noch um die Node-RED-Instanz kümmern.

Ihre Node-RED-Brücke auf ConnectMQ betreiben

Erstellen Sie unter app.connectmq.com kostenlos einen Broker im Serverless-Plan, ohne Kreditkarte, und verbinden Sie Ihre Node-RED-Brücke per TLS damit. Im Live-Topic-Explorer des Portals sehen Sie Befehle und Bestätigungen, während Sie Ihre Endpunkte mit curl testen. Die Grundlagen der MQTT-Nodes finden Sie im Node-RED-MQTT-Tutorial, und für die umgekehrte Richtung, also Alarme an Menschen zu senden, im Leitfaden zum Node-RED-Telegram-Bot.

Häufig gestellte Fragen

Kann Node-RED Webhooks empfangen und an MQTT publizieren?

Ja. Ein http in-Node nimmt den Webhook an, ein Function-Node validiert ihn und baut ein Topic, und ein mqtt out-Node publiziert ihn. Antworten Sie schnell mit einem 2xx-Status, damit der Absender des Webhooks keine Wiederholung startet.

Sollte der HTTP-Endpunkt auf die Antwort des Geräts warten?

Geben Sie standardmäßig 202 Accepted zurück und lassen Sie Clients den Zustand separat abfragen. Auf eine Bestätigung zu warten ist für die interaktive Nutzung sinnvoll, setzen Sie dabei aber immer ein kurzes Timeout.

Welchen QoS sollte ich für Befehle aus HTTP verwenden?

QoS 1 ist ein guter Standard für Befehle, kombiniert mit idempotenten Aktionen wie start und stop statt toggle, damit eine doppelte Zustellung keinen Schaden anrichtet.

Bietet ConnectMQ eine REST-Publish-API?

Nein. ConnectMQ ist ein MQTT-Broker mit MQTT-, TLS- und WebSocket-Endpunkten. Verwenden Sie Node-RED oder einen kleinen eigenen Dienst, um HTTP-Anfragen in MQTT-Publishes zu übersetzen.