Commands

A command is something your device code agrees to do when asked: power off an outlet, restart a process, change a setpoint. There are two ways to use them:

• Declare commands in your code with @px.command and run them from the Commands page or a Command panel on a dashboard. This is the current way.
• Send a command over the API to a handler registered with px.on_command. This is the older way and still works.

Commands travel over the device's live WebSocket, which needs Pro. A device on Free never holds that connection, so nothing can reach it: the API call below returns 200 with { "queued": false }.

Declare commands on the device

Needs plexus-python 0.12.0 or newer, and an API key created with Receive commands ticked on the API Keys page. With any other key, Plexus ignores the declarations and the device keeps sending telemetry as before.

from plexus import Plexus

px = Plexus(source_id="pod-07")

@px.command("power_off", title="Power off", danger="critical", idempotent=True,
            expires_in=30,
            params={"outlet": {"type": "integer", "minimum": 1, "maximum": 8}})
def power_off(run, outlet):
    pdu.outlet(outlet).off()
    return {"outlet": outlet, "state": "off"}

px.serve()   # blocks until Ctrl+C or SIGTERM

Declare commands before the first send(). The handler is called as handler(run, **params), with each parameter already checked against what you declared. What it returns becomes the run's result; an exception marks the run failed. px.serve() is for a program that only answers commands. If your program already loops and calls px.send(), you don't need it.

Then open Commands in Plexus, select the command on the device you want, and click Run command. Each run shows its status and result.

Send a command over the API

This endpoint reaches handlers registered with px.on_command (see below). It does not run commands declared with @px.command. The device must be connected to receive one — always check the queued field in the response to confirm delivery.

The API key needs the write scope. A read-only key gets 403, and so does a key limited to one device when it is used to command a different device.

Request body

Example

curl -X POST https://api.plexus.company/v1/sources/robot-01/commands \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"command": "reboot"}'

curl -X POST https://api.plexus.company/v1/sources/robot-01/commands \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"command": "set_speed", "params": {"rpm": 3000}}'

Response

{ "queued": true }

queued reflects delivery: true means the command reached a connected device. If the device is offline, the request still returns HTTP 200 with { "queued": false } and the command is dropped — so inspect this field rather than relying on the status code. queued: false also comes back when the device is online but its command buffer (64 commands) is full. A 502 is returned only when the gateway itself is unreachable.

The older way: on_command

px.on_command is deprecated: it still works, and raises a DeprecationWarning. Use @px.command for new code. Register handlers before the first send() call:

import time

from plexus import Plexus

px = Plexus(api_key="YOUR_API_KEY", source_id="robot-01")

def handle_set_speed(command, params):
    motor.set_rpm(params["rpm"])
    return {"ok": True}

def handle_reboot(command, params):
    import subprocess
    subprocess.Popen(["reboot"])

px.on_command("set_speed", handle_set_speed)
px.on_command("reboot", handle_reboot)

while True:
    px.send("motor.rpm", motor.current_rpm())
    time.sleep(0.1)

The handler receives (command_name, params_dict) and can return a dict result or raise to signal an error.