Skip to content

Implement a Custom Plugin

This guide shows how to implement a plugin that handles custom commands but does not produce continuous sensor data.

When to Use PluginBase

Use PluginBase directly when your plugin responds to commands but does not continuously push metric data. For sensor data, use SensorPlugin instead — see Implement a Sensor Plugin.

1. Define Commands

Each command is a class with a single async execute class method:

from typing import Dict
from tolomeo.commands.plugin import PluginCmd, PluginCmdContext


class RebootCmd(PluginCmd):
    @classmethod
    async def execute(cls, context: PluginCmdContext) -> Dict:
        # context.plugin is the plugin instance
        # context.payload is the decoded command dict
        plugin_id = context.plugin.id
        return {"status": "rebooting", "plugin": plugin_id}


class GetVersionCmd(PluginCmd):
    @classmethod
    async def execute(cls, context: PluginCmdContext) -> Dict:
        return {"version": "1.0.0"}

2. Define the Plugin

Declare your commands in Meta.commands, set self.id inside connect, and release resources in disconnect:

from tolomeo.plugins import PluginBase


class SystemPlugin(PluginBase):
    class Meta:
        commands = [RebootCmd, GetVersionCmd]

    async def connect(self) -> bool:
        self.id = "system"
        return True

    async def disconnect(self) -> bool:
        return True

3. Use Lifecycle Hooks

Override hook methods to run logic at specific lifecycle points without changing the lifecycle contract:

import asyncio


class SystemPlugin(PluginBase):
    # ... (Meta, connect, disconnect unchanged from section 2)

    async def before_connect(self) -> None:
        self._logger.info("Preparing system plugin resources")

    async def after_setup(self) -> None:
        # Start a background monitoring task after all commands are registered
        await self.task_manager.add_task("monitor", self._monitor)

    async def _monitor(self) -> None:
        while True:
            # periodic monitoring
            await asyncio.sleep(30)

Available hooks (called in order during attach):

  • before_connect → connect → after_connect
  • before_setup → setup → after_setup

And during detach:

  • before_disconnect → disconnect → after_disconnect

4. Wire into a Service

from tolomeo.services import NATSService
from .plugin import SystemPlugin


class SystemService(NATSService):
    class Meta:
        plugin_class = SystemPlugin

    async def setup(self) -> None:
        await super().setup()
        plugin = self._plugin_class(logger=self._logger)
        await plugin.attach()
        self.plugins[plugin.id] = plugin

Sending Commands

Commands are dispatched via the NATS subject commands.{service_name}.req using a SenML envelope. The n field is the command class name and the vs field is a JSON-encoded string containing the payload:

nats pub commands.system.req \
  '[{"n":"GetVersionCmd","vs":"{\"id\":\"system\"}"}]'

See also