Creating a Plugin

This guide walks you through building a real, working MeshDash plugin from scratch. By the end, you'll have a plugin with: a Python API endpoint, a UI page, a background task with watchdog heartbeat, and optional one-time dependency setup. Every code example is copy-paste ready and derived from the real hello_mesh reference plugin shipped with MeshDash.

Step 1 — Create the Directory & Manifest

All plugins live in the plugins/ directory inside your MeshDash installation. Create your plugin directory first, then the manifest file.

mkdir -p plugins/my_plugin/static

Create plugins/my_plugin/manifest.json:

{
  "id": "my_plugin",
  "name": "My Plugin",
  "version": "1.0.0",
  "author": "Your Name",
  "description": "Does something useful.",
  "watchdog": true,
  "entry_point": "main.py",
  "router_prefix": "/api/plugins/my_plugin",
  "static_prefix": "/static/plugins/my_plugin",
  "bridge": null,
  "permissions": [],
  "nav_menu": [
    {
      "label": "My Plugin",
      "href": "/plugin/my_plugin/index.html",
      "icon": "fa-star"
    }
  ]
}

Step 2 — Create the Python Entry Point

Create plugins/my_plugin/main.py. This file registers a route and a background task with watchdog heartbeat. No imports at module scope except declarations and stdlib — all heavy lifting goes in init_plugin.

# plugins/my_plugin/main.py

import os, sys, time, asyncio, threading, json
from fastapi import APIRouter, HTTPException, Query

# ── Module scope: declarations only ──────────────────────────────────────────
PLUGIN_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, PLUGIN_DIR)   # allow importing helper modules from plugin dir

plugin_router = APIRouter()

# Module-level globals — populated by init_plugin()
_context = None
_md = None
_db = None
_cm = None
_loop = None
_log = None
_pid = None
_wd = None


@plugin_router.get("/nodes")
async def list_nodes(limit: int = Query(50, ge=1, le=500)):
    """Return all currently visible nodes."""
    db = _context and _context.get("db_manager")
    if not db:
        raise HTTPException(503, "Plugin not initialized")
    rows = await asyncio.to_thread(db.get_all_nodes)
    return {"plugin": _pid, "nodes": list(rows.values())[:limit]}


@plugin_router.get("/status")
async def status():
    """Return plugin & mesh status."""
    md = _context and _context.get("meshtastic_data")
    return {
        "plugin_id": _pid,
        "nodes_visible": len(md.nodes) if md else 0,
        "radio_status": md.connection_status if md else "unknown",
        "uptime_readable": _uptime(),
    }


# ── Background worker ────────────────────────────────────────────────────────
_start_time = time.time()

def _uptime():
    return time.strftime("%H:%M:%S", time.gmtime(time.time() - _start_time))


def _worker():
    """Background task — logs node count every 60 seconds."""
    while True:
        try:
            md = _context and _context.get("meshtastic_data")
            count = len(md.nodes) if md else 0
            _log.info("Background check: %d nodes visible", count)
        except Exception as e:
            _log.error("Worker error: %s", e)
        time.sleep(60)


async def _watchdog_heartbeat():
    """Ping the core watchdog every 30s — required when watchdog:true."""
    while True:
        try:
            await asyncio.sleep(30)
            if _wd is not None and _pid:
                _wd[_pid] = time.time()
                _log.debug("Watchdog heartbeat sent")
        except asyncio.CancelledError:
            _log.info("Watchdog heartbeat stopped")
            return
        except Exception as e:
            _log.warning("Watchdog heartbeat error: %s", e)


# ── Lifecycle: called once after import ──────────────────────────────────────
def init_plugin(context: dict):
    global _context, _md, _db, _cm, _loop, _log, _pid, _wd
    _context = context
    _md      = context["meshtastic_data"]
    _db      = context["db_manager"]
    _cm      = context["connection_manager"]
    _loop    = context["event_loop"]
    _log     = context["logger"]
    _pid     = context["plugin_id"]
    _wd      = context.get("plugin_watchdog")  # only set for watchdog:true

    # Start background thread
    threading.Thread(target=_worker, daemon=True).start()
    _log.info("Background worker started")

    # Start watchdog heartbeat (watchdog:true → required)
    if _loop:
        asyncio.run_coroutine_threadsafe(_watchdog_heartbeat(), _loop)
        _log.info("Watchdog heartbeat started")
    else:
        _log.warning("event_loop not in context — watchdog heartbeat skipped")

Step 3 — Create the UI Page

Create plugins/my_plugin/static/index.html. This page loads inside the MeshDash chrome when the user clicks "My Plugin" in the sidebar. CSS variables from the active theme are available — no need to hardcode colours.

<!-- plugins/my_plugin/static/index.html -->
<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>My Plugin</title>
    <link rel="stylesheet" href="/static/plugins/my_plugin/style.css">
</head>
<body>
    <div class="plugin-page">
        <div class="toolbar">
            <h2><i class="fas fa-star"></i> My Plugin</h2>
            <button onclick="loadData()"><i class="fas fa-sync"></i> Refresh</button>
        </div>
        <div id="status" class="status-bar">Loading...</div>
        <div id="content"></div>
    </div>
    <script src="/static/plugins/my_plugin/app.js"></script>
</body>
</html>

Create plugins/my_plugin/static/style.css:

/* plugins/my_plugin/static/style.css */

/* All theme variables are available when framed via /plugin/ */
/* See ?page=plugin-development#static-files-heading for full reference */

body {
    font-family: var(--sans, system-ui, sans-serif);
    background: var(--bg1, #0d1117);
    color: var(--txt, #e0e0e0);
    margin: 0;
    padding: 0;
}
.plugin-page  { padding: 20px; }
.toolbar      { display: flex; align-items: center; gap: 12px;
                border-bottom: 1px solid var(--bd2, #30363d);
                padding-bottom: 10px; margin-bottom: 16px; }
.toolbar h2   { margin: 0; font-family: var(--sans); }
.status-bar   { font-size: 0.8rem; color: var(--txt2, #888); margin-bottom: 12px; }
.card         { background: var(--bg2, #161b22); border: 1px solid var(--bd2, #30363d);
                border-radius: var(--r, 6px); padding: 12px 16px; margin-bottom: 8px; }
.card .label  { font-size: 0.75rem; color: var(--txt2); text-transform: uppercase; letter-spacing: 0.05em; }
.card .value  { font-family: var(--mono, monospace); font-size: 0.9rem; color: var(--txt); margin-top: 4px; }
button         { background: var(--acc, #4a9eff); color: #fff; border: none;
                border-radius: 4px; padding: 6px 14px; cursor: pointer; font-size: 0.82rem; }
button:hover   { opacity: 0.85; }
table          { width: 100%; border-collapse: collapse; font-size: 0.85rem; }
th             { background: var(--bg2); color: var(--txt2); font-weight: 600;
                padding: 8px 10px; text-align: left; border-bottom: 1px solid var(--bd2); }
td             { padding: 7px 10px; border-bottom: 1px solid var(--bd2); color: var(--txt); }
tr:hover td    { background: var(--bg3, #2a2a2a); }

Create plugins/my_plugin/static/app.js:

// plugins/my_plugin/static/app.js

const API = "/api/plugins/my_plugin";

// ── Fetch helper ────────────────────────────────────────────────────────────
async function apiFetch(path, options = {}) {
    const resp = await fetch(`${API}${path}`, {
        headers: { "Content-Type": "application/json" },
        ...options,
    });
    if (!resp.ok) {
        const err = await resp.json().catch(() => ({}));
        throw new Error(err.detail || `HTTP ${resp.status}`);
    }
    return resp.json();
}

// ── Load node data ──────────────────────────────────────────────────────────
async function loadData() {
    document.getElementById("status").textContent = "Loading...";
    try {
        const data = await apiFetch("/nodes");
        renderNodes(data.nodes);
        document.getElementById("status").textContent =
            `${data.nodes.length} nodes visible`;
    } catch (err) {
        document.getElementById("status").textContent = `Error: ${err.message}`;
        console.error(err);
    }
}

function renderNodes(nodes) {
    if (!nodes || !nodes.length) {
        document.getElementById("content").innerHTML = "<p>No nodes in range yet.</p>";
        return;
    }
    const html = nodes.map(n => `
        <div class="card">
            <div class="label">${n.node_id || '—'}</div>
            <div class="value">${n.long_name || n.short_name || 'Unknown'}
                ${n.battery_level != null ? ` &bull; 🔋 ${n.battery_level}%` : ''}
                ${n.snr != null ? ` &bull; 📡 ${n.snr}dB` : ''}
            </div>
        </div>`).join("");
    document.getElementById("content").innerHTML = html;
}

// ── Polling ─────────────────────────────────────────────────────────────────
let _pollTimer = null;
function startPolling(ms = 15000) { stopPolling(); _pollTimer = setInterval(loadData, ms); }
function stopPolling()  { if (_pollTimer) { clearInterval(_pollTimer); _pollTimer = null; } }

// ── Auto-start ──────────────────────────────────────────────────────────────
document.addEventListener("DOMContentLoaded", () => {
    loadData();
    startPolling(15000);
});

At this point you have a working plugin — restart MeshDash and your plugin appears in the sidebar with a live node-count page.

Step 4 — Optional: Add a Background Task with Send

Let's make the background worker do something useful — check every 5 minutes and send a message if any node's battery drops below 15%.

Replace the _worker function in main.py:

def _worker():
    """Check nodes every 5 minutes — send alert if battery < 15%."""
    while True:
        try:
            md = _context.get("meshtastic_data")
            if not md:
                time.sleep(300)
                continue

            # Check all nodes for low battery
            low = []
            for nid, n in md.nodes.items():
                level = n.get("battery_level")
                if level is not None and level < 15:
                    name = n.get("long_name") or n.get("short_name") or nid
                    low.append(f"{name} ({level}%)")

            if low:
                msg = "⚠ LOW BATTERY: " + ", ".join(low)
                _log.warning(msg)

                # Send via the radio — from a thread, use run_coroutine_threadsafe
                if _cm and _cm.is_ready.is_set():
                    asyncio.run_coroutine_threadsafe(
                        _cm.sendText(msg, destinationId="^all", channelIndex=0),
                        _loop
                    )
        except Exception as e:
            _log.error("Worker error: %s", e)
        time.sleep(300)

Key patterns in this code:

• Read _md.nodes synchronously — direct dict access, no I/O, safe in a thread.
• Check _cm.is_ready.is_set() before sending — prevents crash when radio is disconnected.
• Use asyncio.run_coroutine_threadsafe — the only correct way to call async sendText from a thread.

Step 5 — Optional: Add One-Time Setup for Dependencies

If your plugin needs Python packages not in MeshDash's environment, add a setup.py guarded by a sentinel file.

Create plugins/my_plugin/setup.py:

"""One-time setup — run as subprocess by main.py on first startup only."""
import subprocess, sys, os

PLUGIN_DIR = os.path.dirname(os.path.abspath(__file__))

def run():
    packages = ["Pillow>=10.0.0", "numpy>=1.24.0"]
    print(f"[my_plugin setup] Installing {len(packages)} package(s)...")
    for pkg in packages:
        print(f"[my_plugin setup] Installing: {pkg}")
        result = subprocess.run(
            [sys.executable, "-m", "pip", "install", pkg, "--quiet"],
            capture_output=True, text=True
        )
        if result.returncode != 0:
            print(f"[my_plugin setup] FAILED: {pkg}")
            sys.exit(1)
        print(f"[my_plugin setup] OK: {pkg}")

    # Write sentinel LAST — only if everything succeeded
    sentinel = os.path.join(PLUGIN_DIR, ".setup_complete")
    with open(sentinel, "w") as f: f.write("1")
    print("[my_plugin setup] Setup complete.")

if __name__ == "__main__":
    run()

Add the sentinel check at the top of init_plugin:

# plugins/my_plugin/main.py — add to init_plugin

import subprocess

PLUGIN_DIR   = os.path.dirname(os.path.abspath(__file__))
SENTINEL     = os.path.join(PLUGIN_DIR, ".setup_complete")
SETUP_SCRIPT = os.path.join(PLUGIN_DIR, "setup.py")

def _run_setup(logger):
    if not os.path.exists(SETUP_SCRIPT):
        return True
    logger.info("First run detected — executing setup.py ...")
    try:
        result = subprocess.run(
            [sys.executable, SETUP_SCRIPT],
            capture_output=True, text=True, timeout=300
        )
        for line in result.stdout.splitlines():
            logger.info(line)
        if result.returncode != 0:
            for line in result.stderr.splitlines():
                logger.error(line)
            logger.error("setup.py exited with code %d", result.returncode)
            return False
        return True
    except subprocess.TimeoutExpired:
        logger.error("setup.py timed out after 300 seconds")
        return False
    except Exception as e:
        logger.error("setup.py failed: %s", e)
        return False

def init_plugin(context: dict):
    global _context, _md, _db, _cm, _loop, _log, _pid, _wd
    _context = context
    _md      = context["meshtastic_data"]
    _db      = context["db_manager"]
    _cm      = context["connection_manager"]
    _loop    = context["event_loop"]
    _log     = context["logger"]
    _pid     = context["plugin_id"]
    _wd      = context.get("plugin_watchdog")

    # ── Step 1: Run one-time setup if sentinel is absent ─────────────────────
    if not os.path.exists(SENTINEL):
        ok = _run_setup(_log)
        if not ok:
            raise RuntimeError("Plugin setup failed — check plugin logs")
    else:
        _log.info("Setup sentinel found — skipping setup")

    # ── Step 2: Import packages that setup.py installed ──────────────────────
    # IMPORTANT: import inside init_plugin, never at module scope
    try:
        from PIL import Image
        import numpy as np
        _log.info("my_plugin ready — PIL and numpy available")
    except ImportError as e:
        raise RuntimeError(f"Required package missing after setup: {e}")

    # ... rest of init_plugin continues as before ...

To reset setup (e.g. after adding a new dependency):

rm plugins/my_plugin/.setup_complete
# Restart MeshDash

Step 6 — Decision Tree: Do You Need the Watchdog?

Does your plugin have a background thread or long-running async task?
  │
  ├─ YES → Set "watchdog": true
  │         Add _watchdog_heartbeat() coroutine
  │         Heartbeat within 120s or plugin is marked "hung"
  │         Core auto-recovers: 3x attempts, 5s apart
  │
  │         Use case: workers, listeners, periodic tasks
  │         Don't use: pure API routes, static-only plugins
  │
  └─ NO → Set "watchdog": false
           Omit _watchdog_heartbeat entirely
           Core never monitors your plugin
           Plugin can be idle indefinitely

  Rule of thumb: If you write while True: or threading.Thread —
  set watchdog: true.

Step 7 — Decision Tree: Direct Memory Access vs HTTP Proxy

Where is the code that needs the data?
  │
  ├─ Python route handler (async def) → DIRECT ACCESS
  │   md = context["meshtastic_data"]
  │   nodes = md.nodes    # instant, no network, no auth
  │
  ├─ Background thread → DIRECT ACCESS (with thread-safe pattern)
  │   nodes = list(_md.nodes.values())   # sync read, GIL-safe
  │   asyncio.run_coroutine_threadsafe(sendText(), _loop)  # async calls
  │
  └─ Browser JavaScript (frontend) → HTTP API
      GET /api/plugins/my_plugin/nodes
      Must go through the network — JS can't reach Python objects

Step 8 — Full Checklist Before Deploying

Directory exists: plugins/my_plugin/     ✓
manifest.json                            ✓
  "id" matches directory name            ✓
  "watchdog" explicitly true/false       ✓
main.py                                  ✓
  PLUGIN_DIR from __file__               ✓
  init_plugin(context) function           ✓
  Store context in module globals        ✓
  Watchdog heartbeat (if watchdog:true)  ✓
  cm.is_ready checked before sendText    ✓
  DB calls via asyncio.to_thread         ✓
  async calls via run_coroutine_threadsafe ✓
static/                                  ✓
  index.html                             ✓
  app.js (or separate JS file)           ✓
  style.css                              ✓
  API calls use /api/plugins/{id}/...    ✓
  File paths use absolute URLs           ✓

Your plugin is ready. If you followed every step, restart MeshDash and you'll see "My Plugin" in the sidebar. Click it to see the node list, and the background worker will start monitoring battery levels in the logs.

Next steps: Read the Plugin Development reference for the complete API. When you're ready to share your plugin, ZIP the directory, submit it to the plugin store, and every MeshDash user can install it with one click.