To connect your device to Enapter Cloud, you first need to generate a communication config.

Generate a communication config using Enapter CLI:

enapter3 device communication-config generate --device-id "$YOUR_DEVICE_ID" --protocol MQTTS | jq .config | base64 --wrap=0

Here's a basic example:

# my_device.py

import asyncio
import enapter

async def main():
    await enapter.standalone.run(MyDevice())

class MyDevice(enapter.standalone.Device):
    async def run(self):
        while True:
            await self.send_telemetry({})
            await self.send_properties({"model": "0.0.1"})
            await asyncio.sleep(10)

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

The most straightforward way to implement your own standalone device is:

1. Subclass enapter.standalone.Device.
2. Override async def run(self) -> None method to send telemetry and properties.
3. Pass an instance of your device to enapter.standalone.run.

enapter.standalone.run will take care of loading the communication config and connecting your device to an upstream.

Use the generated communication config to run your device:

export ENAPTER_STANDALONE_COMMUNICATION_CONFIG="$YOUR_COMMUNICATION_CONFIG"
python my_device.py

enapter.standalone.Device dispatches incoming command execution requests to the corresponding methods on your subclass.

If this command is defined in the manifest:

configure_connection:
  display_name: Configure Connection
  group: connection
  arguments:
    ip_address:
      display_name: IP Address
      type: string
      required: true
    token:
      display_name: Bearer Authentication Token
      type: string
      required: false

Then you would implement the following command handler in your device class:

async def cmd_configure_connection(ip_address: str, token: str | None = None):
    ...

By default, the cmd_ prefix is used to locate command handlers.

Blocking (CPU-bound) code should not be called directly. For example, if a function performs a CPU-intensive calculation for 1 second, all concurrent asyncio Tasks and IO operations would be delayed by 1 second.

Instead, use asyncio.to_thread to offload such work to a thread pool:

await asyncio.to_thread(blocking_call())

Here's an example of a simple Dockerfile:

FROM python:3.13-alpine

WORKDIR /app

RUN python -m venv .venv
COPY requirements.txt requirements.txt
RUN .venv/bin/pip install -r requirements.txt

COPY script.py script.py

CMD [".venv/bin/python", "script.py"]

docker run --network host ...