bench get-app https://github.com/mymi14s/frappe_mqtt

[![Listed on Frappe Gems](https://frappegems.com/api/method/frappe_gems.seo.badge?app=mymi14s%2Ffrappe_mqtt)](https://frappegems.com/gems/apps/mymi14s/frappe_mqtt)

Frappe MQTT

Frappe MQTT - V0.0.3
Easily connect Frappe/ERPNext apps with IoT devices and real-time systems using MQTT messaging. The app supports single or multiple brokers, auto-subscribes from DocTypes, publishes JSON messages, hot-reloads credentials, and operates through simple server APIs. Build event-driven workflows and IoT-ready applications seamlessly within Frappe.

---

Table of Contents

1. Overview & Use Cases
2. Key Features
3. Requirements
4. Installation
5. Configuration
   • Config via site_config.json (single or multiple brokers)
   • MQTT Broker DocType
   • MQTT Topic DocType (Auto‑Subscribe)
   • Broker Topic DocType (Webhook Forwarding)
6. Usage (Python API)
   • Getting a client
   • Publishing (single broker or broadcast)
   • Subscribing
   • Overriding on_message with frappe_db
   • Hot reload behavior
7. Usage (Server API — api.py)
   • frappe_mqtt.api.clients
   • frappe_mqtt.api.publish
   • frappe_mqtt.api.broker_history
   • frappe_mqtt.api.flush_retained
8. Flushing Retained Messages
9. Local Testing with Eclipse Mosquitto
   • Linux / macOS / Windows installs
   • Docker one‑liner
   • Quick CLI test
10. Security & Warnings
11. Troubleshooting
12. FAQ
13. Development & Testing
14. License

---

Overview & Use Cases

Frappe MQTT embeds MQTT client in Frappe site(s). Use-case includes but not limited to:

• IoT ingestion: receive device telemetry (e.g., sensors/#) and trigger workflows.
• ERP events → field devices: publish status updates/orders to devices.
• System telemetry: consume $SYS/# and broker metrics for monitoring.
• Cross‑system glue: bridge legacy systems by publishing/consuming JSON messages.
• Webhook forwarding: push MQTT payloads to external HTTP endpoints automatically.
• Database logging: safely create/update Frappe documents from MQTT callbacks using frappe_db.

It supports multiple brokers, auto‑subscription from DocTypes, JSON‑only payloads, webhooks, thread-safe DB access, and hot reload when credentials change.

---

Key Features (Enhanced)

• Multi‑broker support via site_config.json and/or MQTT Broker DocType.
• Auto‑subscription from MQTT Topic DocType on (re)start/reload.
• Webhook forwarding via Broker Topic DocType — forward messages to HTTP endpoints.
• Publish JSON payloads; payloads are auto‑extended with timestamp.
• Hot reload when site_config.json or Broker DocType changes.
• Flush retained messages (single topic, wildcard branch, or all).
• Role‑based: use Frappe permissions to secure access to server methods if required (MQTT role).
• Scheduler-Based Client Lifecycle — clients auto-initialize on bench migrate or DocType update.
• Thread-Safe Database Access — each client exposes c.client.frappe_db for safe get_doc, new_doc, get_list, get_all in MQTT threads.
• Retained Message History API — fetch latest retained state for monitoring or sync.

---

Requirements

• Frappe >=v15 (Python 3.10+ recommended)
• Paho‑MQTT 2.1.0 (MQTT v5 protocol)
• Optional TLS certificates for secure brokers
• Broker (MQTTV5 protocol) *

---

Installation

# In your bench
bench get-app github.com/mymi14s/frappe_mqtt
bench --site yoursite install-app frappe_mqtt

bench --site yoursite clear-cache
bench restart

Assign MQTT role to users who will call server APIs or configure DocTypes.

---

Configuration

You can configure brokers in two ways — both can coexist.

Config via site_config.json (single or multiple brokers)

Edit your site’s site_config.json and add an mqtt_config section.

Single broker example:

{
  "mqtt_config": [
      {
          "name": "default",
          "host": "localhost",
          "port": 1883,
          "username": "",
          "password": "",
          "ca_certs": "",
          "certfile": "",
          "keyfile": "",
          "tls_insecure": 0,
          "keepalive": 60,
          "error_topic": "errors/frappe"
      }
  ]
}

Multiple brokers + TLS example:

{
  "mqtt_config": [
    {
        "name": "default",
        "host": "localhost",
        "port": 1883,
        "username": "",
        "password": "",
        "ca_certs": "",
        "certfile": "",
        "keyfile": "",
        "tls_insecure": 0,
        "keepalive": 60,
        "error_topic": "errors/frappe"
    },
    {
        "name": "hivemq",
        "host": "localhost",
        "port": 1883,
        "username": "",
        "password": "",
        "ca_certs": "",
        "certfile": "",
        "keyfile": "",
        "tls_insecure": 0,
        "keepalive": 60,
        "error_topic": "errors/frappe"
    }
  ]
}

Notes - host and port are required. - Credentials and TLS files are optional. - The top‑level keys (default, secure, etc.) are broker keys used by the API.

MQTT Broker DocType

ROLE MQTT is required to use the doctypes.

Create MQTT Broker documents in Desk to define brokers at runtime:

• Broker Key (unique): e.g., production, lab, etc.
• Host, Port
• Username, Password (optional; Password field recommended)
• TLS: CA / cert / key (optional)

Brokers defined here are managed the same way as those from site_config.json. The app can use both sources simultaneously.

MQTT Topic DocType (Auto‑Subscribe)

Create MQTT Topic records to declare subscriptions the client should maintain:

• Topic Filter: e.g., sensors/#, devices/+/status
• Broker: which broker key to use (leave blank to subscribe on all active brokers)

On client start or reload, the app auto‑subscribes to all saved topics.

Broker Topic DocType (Webhook Forwarding)

This is an extension of MQTT Topic with webhook capabilities.

• Inherits all fields from MQTT Topic
• Adds webhook_url (string): HTTP endpoint to forward incoming messages to
• When a message arrives on the subscribed topic, it is POSTed as JSON to the webhook_url

Example payload sent to webhook:

{
  "topic": "sensors/temperature",
  "payload": {"value": 25.3, "unit": "C"},
  "qos": 0,
  "retain": false,
  "timestamp": "2025-04-05T10:30:00.123456"
}

---

Usage (Python API)

Import helpers from utility.py.

Getting a client

from frappe_mqtt.mqtt_utility import get_client

# by broker key (from site_config.json or Broker DocType)
client = get_client("default")  # or "secure", "production", etc.

Publishing (single broker or broadcast)

# Single broker
client.publish("devices/device42/state", {"status": "active"}, qos=0, retain=False)

# Broadcast the same message to ALL connected brokers
client.publish("system/restart", {"notice": "rolling"}, broadcast=True)

The client extends your payload automatically with:

{
  "timestamp": "YYYY-MM-DDTHH:MM:SS.mmmmmm"
}

Payload must be a dict (or JSON‑serializable).

Subscribing

# Subscribe to a topic filter (wildcards +/# supported by the broker)
client.subscribe("sensors/+/status", qos=0)

Overriding on_message with frappe_db

You can attach your own on_message handler for advanced processing — including safe database access via client.frappe_db.

# your_app/handler.py

import json

def on_message(self, client, userdata, message_dict, raw):
    # Access raw MQTT message
    print("Topic:", raw.topic)
    print("Payload:", message_dict)

    # Thread-safe Frappe DB access
    db = client.frappe_db  #  This is key!

    # Example: Create a new document
    log = db.new_doc({
        "name": "vjhbgdfckn",
        "doctype": "Temperature Log",
        "timestamp": frappe.utils.now(),
        "value": message_dict.get("value", 0)
    })
    log.insert()

    # Example: Fetch existing doc
    user = db.get_doc("User", "Administrator")
    print("Admin:", user.full_name)

    # Example: Query recent logs
    recent = db.get_list("Temperature Log", fields=["timestamp", "value"], limit=5)
    print("Recent:", recent)

# your_app/__init__.py
from frappe_mqtt.mqtt_utility import MQTTClient
from your_app.handler import on_message

MQTTClient.on_message = on_message

The built‑in logic validates JSON.
Attachments to c.client are standard Paho‑MQTT hooks.
frappe_db supports: get_doc, new_doc, get_list, get_all, set_value, etc.

Hot reload behavior

• Changing MQTT Broker DocType normally triggers a client reload for that broker.
• bench start/serve runs the client
• bench migrate, works same as scheduler.
• Updating site_config.json requires either calling your reload helper (if provided) or restarting the bench.
• Calling get_client(key) returns an existing connection or creates/reloads it if needed.

---

Usage (Server API api.py)

All endpoints are whitelisted and return JSON‑serializable output. Currently it is secured with Role MQTT.

frappe_mqtt.api.clients

Return available broker keys.

# frappe.call (JS) or server call (Python)
# -> {"clients": ["default", "secure", ...]}

frappe_mqtt.api.publish

Publish JSON payload to one broker or broadcast to all.

Args: - topic (str) — required - payload_json (str) — JSON string - qos (int: 0/1/2) default 0 - retain (int: 0/1) default 0 - client_key (str|None) — which broker; ignored if broadcast=1 - broadcast (int: 0/1) — publish to all

Example (JS):

frappe.call({
  method: "frappe_mqtt.api.publish",
  args: {
    topic: "devices/alpha/state",
    payload_json: JSON.stringify({ status: "online" }),
    qos: 0, retain: 0, client_key: "default", broadcast: 0
  }
})

frappe_mqtt.api.broker_history

Fetch a snapshot of retained (or recent) messages matching a topic filter.

Args: - topic (str) — e.g., sensors/# - limit (int) — default 100 - timeout_ms (int) — how long to wait for retained packets (default ~1200ms) - client_key (str|None) — which broker

Returns: list of messages like

[
  {"topic": "sensors/lab/temp", "payload": {"t": 22.1}, "qos": 0, "retain": 1, "timestamp": "..."}
]

frappe_mqtt.api.flush_retained

Clear retained messages for a single topic, a wildcard branch, or all topics.

Args: - client_key (str|None) — broker key (default: "default") - topic (str|None) — exact topic or wildcard (ignored if clear_all=1) - clear_all (int: 0/1) — set to 1 to clear all retained topics - timeout (float) — seconds to discover retained topics for wildcard/all (default ≈ 1.5) - qos (int) — QoS used for the clearing publishes (default 0)

Example: ```js // clear a single retained message frappe.call({ method: "frappemqtt.api.flushretained", args: { clientkey: "default", topic: "devices/x/status", clearall: 0 } })

// clear all retained under a branch frappe.call({ method: "frappemqtt.api.flushretained", args: { clientkey: "default", topic: "devices/#", clearall: 0, time