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. Read Configuration (Optional)

To take options from the deployment's configuration file, declare a Pydantic model in Meta.config_model and read self.options:

from pydantic import BaseModel, ConfigDict, PositiveInt


class SystemOptions(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True)

    monitor_interval_s: PositiveInt = 30


class SystemPlugin(PluginBase):
    # ... (connect, disconnect, before_connect, after_setup unchanged from sections 2-3)

    class Meta:
        commands = [RebootCmd, GetVersionCmd]  # unchanged from section 2
        config_model = SystemOptions           # carried over from section 2, plus config_model

    async def _monitor(self) -> None:
        while True:
            await asyncio.sleep(self.options.monitor_interval_s)

_monitor here replaces section 3's version; after_setup still schedules it, so the task keeps running, now sleeping for the configured monitor_interval_s instead of a hardcoded 30 seconds.

[plugins.system]
monitor_interval_s = 60

Options are keyed by the plugin id, so they are readable only once connect has assigned it. extra="forbid" means a mistyped key stops the plugin at attach time and names it. See the Configuration Reference.

5. Wire into a Service

Declare the plugin class and let ServiceBase acquire it:

from tolomeo.services import NATSService
from .plugin import SystemPlugin


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

    async def acquire_plugins(self) -> None:
        await self.attach_plugin(self.new_plugin())

attach_plugin instantiates nothing itself — it wires the plugin's callbacks (via wire_plugin), attaches it, and registers it only if the attach succeeded, so a plugin that failed to connect never enters the registry. Override wire_plugin to bind your own handlers.

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