Skip to content

Build Your First ToloMEO Microservice

This tutorial walks through creating a minimal sensor microservice using Py ToloMEO. You will build a plugin that generates data, a service that connects it to NATS, and run the whole thing locally.

What you will do

Build a service called dummy that publishes a random metric value to events.data every second over NATS.

A fuller version of this service lives in examples/temperature/.

Prerequisites

  • Python ≥ 3.11
  • uv installed
  • A working instance of the ToloMEO Edge agent
  • Py ToloMEO installed: uv add tolomeo

1. Create the Plugin

A plugin wraps your hardware or data source. Subclass SensorPlugin to get built-in state management and metric registration; it is push-only — there is no internal queue or polling loop, so a plugin pushes each reading directly via await self.push_reading({...}).

Create my_service/plugin.py:

import asyncio
import time
from random import random
from typing import Dict

from tolomeo.commands.plugin import PluginCmd, PluginCmdContext
from tolomeo.metrics import Metric
from tolomeo.plugins import SensorPlugin


class PingCmd(PluginCmd):
    """A simple command that returns the plugin ID."""

    @classmethod
    async def execute(cls, context: PluginCmdContext) -> Dict:
        return {"message": f"pong from {context.plugin.id}"}


class DummyPlugin(SensorPlugin):
    class Meta:
        commands = [PingCmd]
        out_metrics = [Metric("dummy_metric", "")]

    async def connect(self) -> bool:
        self.id = "DummyPlugin"         # must be set inside connect()
        self._logger.info("Connected")
        return True

    async def disconnect(self) -> bool:
        self._logger.info("Disconnected")
        return True

    async def after_setup(self) -> None:
        # Start a background task that generates data
        await self.task_manager.add_task("simulate", self._simulate)

    async def _simulate(self) -> None:
        while True:
            await asyncio.sleep(1)
            await self.push_reading({
                "timestamp": round(time.time()),
                f"{self.id}:dummy_metric": random(),
            })

2. Create the Service

A service manages one or more plugins and handles NATS subscriptions. Subclass SingleSensorService for sensor-style services that publish metrics as plugins push them.

Create my_service/service.py:

from tolomeo.services import SingleSensorService
from .plugin import DummyPlugin


class DummyService(SingleSensorService):
    class Meta:
        plugin_class = DummyPlugin

That is the whole service. SingleSensorService acquires the plugin, wires its data and info callbacks, and wakes the params loop once it is registered.

Why there is no setup override

ServiceBase.setup registers the service commands and then calls acquire_plugins, which is the hook to override when a service needs something other than one fixed plugin — several plugins, or discovery:

from tolomeo.services import SensorService

class DummyService(SensorService):
    class Meta:
        plugin_class = DummyPlugin

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

attach_plugin registers the plugin only if attach() succeeded, so one that failed to connect never ends up in the registry being served commands.

3. Wire the Entry Point

Create my_service/main.py:

import asyncio
import logging
from .service import DummyService

logging.basicConfig(level=logging.INFO)


async def main() -> None:
    service = DummyService("dummy")
    try:
        await service.run()
    except asyncio.CancelledError:
        pass


if __name__ == "__main__":
    asyncio.run(main())

4. Run the Service

Start the service:

DEVICE_SERIAL_NUMBER=test001 python -m my_service.main

In a separate terminal, subscribe to NATS to observe the output:

nats sub "events.data"

You should see SenML records arriving every second:

[{"bn": "urn:cpt:device:sn:test001:", "n": "DummyPlugin:dummy_metric", "v": 0.723, "t": 1700000010.0}]

5. Send a Command

nats pub commands.dummy.req \
  '[{"n":"PingCmd","vs":"{\"id\":\"DummyPlugin\"}"}]'

The response appears on events.params:

[{"bn": "urn:cpt:device:sn:test001:", "n": "PingCmd", "vs": "{\"message\": \"pong from DummyPlugin\"}"}]

Next Steps