Extend MeshDash.
Build the Tools
Your Mesh Needs.

plugins/
└── my_plugin/
    ├── manifest.json    ← required
    ├── main.py          ← entry point
    ├── requirements.txt ← pip deps
    ├── static/          ← served at /static/plugins/my_plugin/
    │   └── index.html
    └── README.md        ← required for store submission

{
  "id":            "my_plugin",
  "name":          "My Plugin",
  "version":       "1.0.0",
  "author":        "Your Name",
  "description":   "What this plugin does.",
  "entry_point":   "main.py",
  "router_prefix": "/api/plugins/my_plugin",
  "static_prefix": "/static/plugins/my_plugin",
  "watchdog":      false,
  "nav_menu": [{
    "label": "My Plugin",
    "icon":  "fa-puzzle-piece",
    "href":  "/plugin/my_plugin/index.html"
  }]
}

from fastapi import APIRouter
plugin_router = APIRouter()
_ctx = {}

def init_plugin(context: dict):
    global _ctx
    _ctx = context
    _ctx['logger'].info("my_plugin loaded")

@plugin_router.get("/hello")
async def hello():
    node = _ctx['meshtastic_data'].local_node_id
    return {"node": node, "status": "ok"}

@plugin_router.post("/announce")
async def announce(text: str):
    conn = _ctx['connection_manager']
    await conn.sendText(text, destinationId="^all",
                        channelIndex=0)
    return {"sent": True}

import time, threading
from fastapi import APIRouter

plugin_router = APIRouter()
_running = False

@plugin_router.get("/status")
async def status():
    return {"running": _running}

def _loop(ctx):
    global _running
    pid      = ctx["plugin_id"]
    logger   = ctx["logger"]
    watchdog = ctx["plugin_watchdog"]
    conn     = ctx["connection_manager"]
    _running = True
    logger.info("Background task started")

    while _running:
        try:
            # ── Do your work here ──────────────
            logger.debug("poll cycle")

            # MUST heartbeat if watchdog=true
            watchdog[pid] = time.time()
        except Exception as e:
            logger.error(f"Poll error: {e}")
        time.sleep(30)

def init_plugin(context: dict):
    t = threading.Thread(
        target=_loop, args=(context,), daemon=True)
    t.start()
    context["logger"].info("Plugin init complete")