How to Create Actions

An action is a function your program serves to the Zelos Agent. The Zelos App shows each action as a form. You fill in the form, run the action, and see its result. Zelos AI can run actions too.

Python, Rust and Go can all define and serve actions. Each action declares its inputs as a JSON Schema, and the app builds the form from that schema. Python has the widest API: decorators, widgets, and input checks before your code runs. Rust and Go build the same kind of form with a smaller set of field types.

Feature	Python	Rust	Go
Text, number, integer, boolean and select fields	Yes	Yes	Yes
title, description, default, minimum, maximum	Yes	Yes	Yes
Fields are required by default	Yes	No	No
Email, date, file, array, object and custom fields	Yes	No	No
pattern, min_length, max_length, multiple_of	Yes	No	No
SDK checks inputs before your code runs	Yes	No	No
Widgets and dynamic choices	Yes	Hand-written schema only	Hand-written schema only
Default timeout	30 s, set with timeout=	None, set with default_timeout_ms	None, set with DefaultTimeout()
read_only, submit_text, standalone	Yes	No	No
Entry-point discovery	Yes	No	No

Quick Start

PythonRustGo

from zelos_sdk import init, action

@action("Read Voltage", "Read voltage from a power rail")
@action.select("rail", choices=["3v3", "5v0", "12v0"])
def read_voltage(rail: str):
    values = {"3v3": 3.28, "5v0": 5.02, "12v0": 11.95}
    return values[rail]

# Serve as power/read_voltage (blocks this process)
init("power", actions=True, block=True)

use std::sync::Arc;

use serde_json::{json, Value};
use tokio_util::sync::CancellationToken;
use zelos::{
    ActionExecuteResult, ActionFn, ActionSchema, ActionsClient, ActionsError, ActionsRegistry,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let read_voltage = ActionFn::new(
        ActionSchema::new("Read Voltage", "Read voltage from a power rail").select(
            "rail",
            &["3v3", "5v0", "12v0"],
            |f| f.required(),
        ),
        |params: Value| async move {
            let volts = match params.get("rail").and_then(Value::as_str) {
                Some("3v3") => 3.28,
                Some("5v0") => 5.02,
                Some("12v0") => 11.95,
                _ => return Err(ActionsError::InvalidInput("unknown rail".to_string())),
            };
            Ok(ActionExecuteResult::done(&json!(volts)))
        },
    );

    let registry = Arc::new(ActionsRegistry::new());
    registry.register("read_voltage".to_string(), Arc::new(read_voltage));

    // Serve as power/read_voltage until Ctrl-C
    let client = ActionsClient::new_with_url("grpc://127.0.0.1:2300".to_string())?;
    let token = CancellationToken::new();
    tokio::select! {
        result = client.serve("power".to_string(), registry, token.clone()) => result?,
        _ = tokio::signal::ctrl_c() => token.cancel(),
    }
    Ok(())
}

This program depends on zelos, tokio (with the macros, rt-multi-thread and signal features), tokio-util and serde_json.

package main

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "log"
    "os"
    "os/signal"

    "github.com/zeloscloud/zelos/go"
)

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
    defer stop()

    schema := zelos.NewActionSchema("Read Voltage", "Read voltage from a power rail").
        Select("rail", []string{"3v3", "5v0", "12v0"}, zelos.Required())

    readVoltage := zelos.NewActionFromSchema(schema, func(ctx context.Context, params json.RawMessage) (*zelos.ActionResult, error) {
        var p struct {
            Rail string `json:"rail"`
        }
        if err := json.Unmarshal(params, &p); err != nil {
            return nil, err
        }
        volts, ok := map[string]float64{"3v3": 3.28, "5v0": 5.02, "12v0": 11.95}[p.Rail]
        if !ok {
            return nil, fmt.Errorf("unknown rail %q", p.Rail)
        }
        return zelos.ActionResultDone(volts), nil
    })

    registry := zelos.NewActionsRegistry()
    registry.Register("read_voltage", readVoltage)

    // Serve as power/read_voltage until Ctrl-C
    client := zelos.NewActionsClient(ctx, "power", registry, zelos.DefaultActionsClientConfig())
    if err := client.Run(); err != nil && !errors.Is(err, context.Canceled) {
        log.Fatal(err)
    }
}

Run your program, then open the Explorer view's Actions section in the Zelos App.

Core Concepts

• An action has a title, a description, input fields, and a handler that returns a result.
• You register each action in a registry under a path, such as read_voltage.
• You serve the registry to the agent under a service name. The agent lists the action as <service name>/<path>, such as power/read_voltage.
• The result carries a value and a status: PASS, FAIL, ERROR or DONE.

In Python, a decorator does most of this work:

• Decorate a function with @action(title, description) to expose it.
• Add input decorators (e.g., @action.number, @action.select) to define typed parameters with validation and UI hints.
• Pass submit_text to relabel the app's Execute button, e.g. @action("Flash Firmware", submit_text="Flash").
• Pass read_only=True when the action changes nothing for any input: no device, file, config or trace. Zelos AI runs a read-only action without asking first, and asks before every other action.
• Serve actions via init(actions=True, block=True) or the Actions client.

Rust and Go have no read_only flag, so Zelos AI asks before it runs any Rust or Go action. They also have no submit_text.

Fields and Validation

PythonRustGo

All field decorators support common options: required (default True), default, title, description, widget, placeholder and ui_options.

• @action.text(name, min_length, max_length, pattern)
• @action.email(name)
• @action.number(name, minimum, maximum, multiple_of)
• @action.integer(name, minimum, maximum, multiple_of)
• @action.boolean(name)
• @action.date(name)
• @action.select(name, choices, depends_on)
• @action.file(name, accept)
• @action.files(name, accept)
• @action.array(name, item, min_items, max_items, unique_items)
• @action.object(name, properties)
• Generic: @action.input(name, type="...", ...) for custom/registered types

Example:

from zelos_sdk import action

@action("Configure Supply", "Set output and limits")
@action.number("voltage", minimum=1.0, maximum=30.0)
@action.number("current_limit", minimum=0.1, maximum=5.0, default=1.0, required=False)
@action.boolean("enable_output", default=False, required=False)
def configure_supply(voltage: float, current_limit: float = 1.0, enable_output: bool = False):
    return {"voltage": voltage, "current_limit": current_limit, "enabled": enable_output}

Before your function runs, the SDK checks each input against its field: type, range, length, pattern and choices. It also rejects a missing required field and an unknown field. A failed check ends the run with an ERROR result.

default pre-fills the form. The SDK does not pass it to your function when a caller omits the field. Give the parameter the same Python default.

ActionSchema builds the JSON Schema. Each field method takes a name and a closure that sets the field's options: title, description, required, default_value, minimum and maximum. The field types are text, number, integer, boolean and select.

use serde_json::{json, Value};
use zelos::{ActionExecuteResult, ActionFn, ActionSchema, ActionsError};

let configure_supply = ActionFn::new(
    ActionSchema::new("Configure Supply", "Set output and limits")
        .number("voltage", |f| {
            f.title("Voltage (V)").minimum(1.0).maximum(30.0).required()
        })
        .number("current_limit", |f| {
            f.minimum(0.1).maximum(5.0).default_value(json!(1.0))
        })
        .boolean("enable_output", |f| f.default_value(json!(false)))
        .select("mode", &["cc", "cv"], |f| f.description("Regulation mode")),
    |params: Value| async move {
        // The SDK does not check params against the schema
        let voltage = params
            .get("voltage")
            .and_then(Value::as_f64)
            .filter(|v| (1.0..=30.0).contains(v))
            .ok_or_else(|| ActionsError::InvalidInput("voltage must be 1-30 V".to_string()))?;
        let current_limit = params
            .get("current_limit")
            .and_then(Value::as_f64)
            .unwrap_or(1.0);
        let enabled = params
            .get("enable_output")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        Ok(ActionExecuteResult::done(&json!({
            "voltage": voltage,
            "current_limit": current_limit,
            "enabled": enabled,
        })))
    },
);

A field is optional unless you call required(). The handler receives the inputs as one serde_json::Value object.

NewActionSchema builds the JSON Schema. Each field method takes a name and options: zelos.Title, zelos.Description, zelos.Required, zelos.Default, zelos.Minimum and zelos.Maximum. The field types are Text, Number, Integer, Boolean and Select.

import (
    "context"
    "encoding/json"
    "fmt"

    "github.com/zeloscloud/zelos/go"
)

schema := zelos.NewActionSchema("Configure Supply", "Set output and limits").
    Number("voltage", zelos.Title("Voltage (V)"), zelos.Minimum(1), zelos.Maximum(30), zelos.Required()).
    Number("current_limit", zelos.Minimum(0.1), zelos.Maximum(5), zelos.Default(1.0)).
    Boolean("enable_output", zelos.Default(false)).
    Select("mode", []string{"cc", "cv"}, zelos.Description("Regulation mode"))

configureSupply := zelos.NewActionFromSchema(schema, func(ctx context.Context, params json.RawMessage) (*zelos.ActionResult, error) {
    p := struct {
        Voltage      *float64 `json:"voltage"`
        CurrentLimit float64  `json:"current_limit"`
        EnableOutput bool     `json:"enable_output"`
    }{CurrentLimit: 1.0}
    if err := json.Unmarshal(params, &p); err != nil {
        return nil, err
    }
    // The SDK does not check params against the schema
    if p.Voltage == nil || *p.Voltage < 1 || *p.Voltage > 30 {
        return nil, fmt.Errorf("voltage must be 1-30 V")
    }
    return zelos.ActionResultDone(map[string]any{
        "voltage":       *p.Voltage,
        "current_limit": p.CurrentLimit,
        "enabled":       p.EnableOutput,
    }), nil
})

A field is optional unless you pass zelos.Required(). The handler receives the inputs as one JSON object in params.

The Rust and Go SDKs do not check inputs against the schema. The app's form enforces the schema, but other callers such as Zelos AI and scripts send inputs directly. Check every input in your handler. Both builders keep fields in the order you declare them, and the form shows them in that order.

The Rust and Go builders emit an empty UI schema. To set widgets or build choices from the form's current values, implement the Action trait (Rust) or interface (Go) yourself. Its schema method receives the form's current values and returns your own JSON Schema and UI schema strings.

UI Widgets

Widgets are Python only. Select the best widget for the input using widget="...":

Widget	Field Types	Purpose
text	text, email	Single-line input
textarea	text	Multi-line input
password	text	Hidden input
number, updown, range	number/integer	Numeric entry, spinner, slider
select, radio, multi_select	select	Dropdown, radio, multi-choice
checkbox, toggle	boolean	Boolean inputs
date	date	Date picker
file, files	file(s)	File uploads
file-picker	text	File path picker (no upload)
folder-picker	text	Folder path picker (no upload)

Example with widgets:

from zelos_sdk import action

@action("Configure Logging")
@action.text("log_prefix", max_length=32)
@action.text("log_filter", widget="textarea")
@action.text("admin_password", widget="password")
@action.boolean("enable_remote", widget="toggle", default=False)
@action.select("log_level", choices=["DEBUG","INFO","WARN","ERROR"], widget="radio")
@action.integer("rotation_size_mb", minimum=1, maximum=1000, widget="updown", default=100)
@action.number("compression_ratio", minimum=0.1, maximum=1.0, widget="range", default=0.8)
def configure_logging(log_prefix: str, log_filter: str, admin_password: str,
                      enable_remote: bool = False, log_level: str = "INFO",
                      rotation_size_mb: int = 100, compression_ratio: float = 0.8):
    return {
        "prefix": log_prefix,
        "level": log_level,
        "rotation_mb": rotation_size_mb,
        "compression": compression_ratio,
        "remote_enabled": enable_remote,
    }

Range sliders with precise increments:

Use multiple_of to control step size for numeric sliders:

@action("Set Interval", "Change sample rate")
@action.number(
    "seconds",
    minimum=0.001,
    maximum=1.0,
    multiple_of=0.001,
    default=0.1,
    title="Interval (seconds)",
    description="Sample interval from 1kHz to 1Hz",
    widget="range",
)
def set_interval(seconds: float):
    """Update the sample interval with 1ms precision."""
    return {"message": f"Interval set to {seconds}s", "interval": seconds}

File Uploads

File fields are Python only. Actions can accept file uploads using @action.file() or @action.files(). Files are received as base64-encoded Data URLs.

from zelos_sdk import action
import base64

@action("Process Configuration File")
@action.file("config_file", accept=".json,.yaml,.toml", description="Upload configuration file")
def process_config(config_file: str):
    """
    Process uploaded configuration file.

    Args:
        config_file: Data URL string like "data:application/json;base64,eyJ..."
    """
    # Parse data URL
    if config_file.startswith("data:"):
        header, encoded = config_file.split(",", 1)
        mime_type = header.split(";")[0].replace("data:", "")

        # Decode from base64
        file_bytes = base64.b64decode(encoded)
        file_text = file_bytes.decode("utf-8")

        return {
            "mime_type": mime_type,
            "size_bytes": len(file_bytes),
            "content_preview": file_text[:100]
        }

@action("Analyze Log Files")
@action.files("log_files", accept=".log,.txt", description="Upload multiple log files")
def analyze_logs(log_files: list[str]):
    """
    Analyze multiple uploaded log files.

    Args:
        log_files: List of data URL strings
    """
    results = []
    for idx, log_url in enumerate(log_files):
        if log_url.startswith("data:"):
            header, encoded = log_url.split(",", 1)
            file_bytes = base64.b64decode(encoded)

            # Count lines
            line_count = file_bytes.decode("utf-8").count("\n")

            results.append({
                "file_index": idx,
                "size_bytes": len(file_bytes),
                "line_count": line_count
            })

    return {"files_processed": len(results), "results": results}

File Size Limits

• Single file (@action.file): 10 MB maximum
• Multiple files (@action.files): 10 files and 50 MB total maximum
• Files are base64-encoded, increasing payload size by ~33%

File Path Picker

For file paths without uploading the file contents, use widget="file-picker":

from zelos_sdk import action

@action("Load Simulation Model")
@action.text("model_path", widget="file-picker", description="Path to model file")
def load_model(model_path: str):
    """
    Load model from a file path.

    Args:
        model_path: File system path as string (e.g., "/path/to/model.mdl")
    """
    return {"loaded_model": model_path}

This widget provides a file browser but returns only the path string, not the file contents.

Folder Path Picker

For selecting output directories or folder paths, use widget="folder-picker":

from zelos_sdk import action

@action("Export Results")
@action.text("input_file", widget="file-picker", description="Input data file")
@action.text("output_folder", widget="folder-picker", description="Output directory")
def export_results(input_file: str, output_folder: str):
    """
    Export processed results to a folder.

    Args:
        input_file: Path to input file
        output_folder: Directory path where results will be saved (e.g., "/path/to/output")
    """
    return {"output_location": output_folder}

This widget provides a folder browser that returns the selected directory path as a string.

To allow users to create new folders during selection, pass the createDirectory option:

@action.text(
    "output_folder",
    widget="folder-picker",
    ui_options={"createDirectory": True},
    description="Select or create output directory"
)

Dynamic Fields

Dynamic choices are Python only. Use functions as choices and declare dependencies with depends_on.

from zelos_sdk import action

def get_channels(sensor_type: str):
    return {
        "temperature": ["ch1_cpu", "ch2_ambient"],
        "pressure": ["ch1_intake", "ch2_exhaust"],
    }.get(sensor_type, [])

@action("Read Sensor Channel")
@action.select("sensor_type", choices=["temperature", "pressure"])
@action.select("channel", choices=get_channels, depends_on="sensor_type")
def read_sensor_channel(sensor_type: str, channel: str):
    return {"sensor_type": sensor_type, "channel": channel}

Choices for a Config Field

Standalone actions are Python only. An extension's config form can offer values an action discovers on the agent host, such as the network interfaces or CAN buses present there. Declare the action standalone=True so the agent can run it out of the installed package before the extension has ever started, and return choices in the order to show:

from zelos_sdk import action

@action("List Interfaces", standalone=True)
def list_interfaces():
    return {
        "status": "success",
        "choices": [
            {"value": "en0", "detail": "up · 10.0.0.5"},
            {"value": "lo0", "detail": "up · loopback"},
        ],
    }

Each choice carries value (what the field takes), an optional label (shown instead of value), and an optional detail (a trailing note). On failure return {"status": "error", "message": "..."}; the form shows the message. Then point the field at the action in config.schema.json:

"interface": {
  "type": "string",
  "ui:widget": "action-choices",
  "ui:options": { "action": "Packet/list_interfaces" }
}

The field stays a free-text input: a user can type a value for a host the agent cannot see. ui:options.params is forwarded to the action as its arguments.

A schema may only name its own extension's actions, addressed as <extension name>/<action>; the form refuses anything else before the agent is asked.

Auto-configure from an Action

A Python standalone action can fill in the whole form: return config, an object whose keys replace the form's, and name it at the schema root. The form shows an Auto-configure button; nothing runs until it is clicked, and the person reviews the result before saving. An optional message shows beside the button, e.g. to say a demo was filled in because nothing was detected. On failure return {"status": "error", "message": "..."} or raise; either message shows there too.

@action("Auto-configure", standalone=True)
def auto_config():
    return {"status": "success", "config": {"interfaces": [{"interface": "en0"}]}}

{
  "type": "object",
  "ui:options": { "autoconfig": "Packet/auto_config" },
  "properties": { "...": "..." }
}

Declare a config parameter to receive the unsaved form, for example to scan the endpoint the person just typed. Actions without it are called with no arguments.

Register and Serve Actions

A registry maps each path to an action. A client connects to the agent and serves one registry under a service name. The client reconnects on its own when the connection drops.

PythonRustGo

@action registers a module-level function in the global registry under its function name. It registers a static method as Class/method. It does not register an instance method; register the instance instead, as shown below.

init(name, actions=True) serves the global registry under name. The name defaults to "python". init connects to ZELOS_AGENT_URL, or to http://localhost:2300 when that variable is unset. Pass url= to choose another agent. Pass block=True to keep the process alive.

Class-based actions keep state and are namespaced when registered.

from zelos_sdk import action, actions_registry, init

class Battery:
    def __init__(self):
        self.status = {"soc_percent": 85.2, "voltage": 398.4}

    @action("Get Status")
    def get_status(self):
        return self.status

    @action("Set Mode")
    @action.select("mode", choices=["normal", "fast", "storage"])
    def set_mode(self, mode: str):
        self.status["mode"] = mode
        return {"mode": mode}

bms = Battery()
actions_registry.register(bms, name="bms")  # bms/get_status, bms/set_mode
init(actions=True, block=True)

Register a function under another path with name. The function keeps its automatic path too, and the registry serves both:

from zelos_sdk import actions_registry

actions_registry.register(read_voltage, name="power/read_voltage")
actions = actions_registry.list()  # e.g. ["power/read_voltage", "read_voltage", ...]

To serve your own registry, create an ActionsClient. It connects to grpc://localhost:2300 unless you pass url=.

from zelos_sdk import init
from zelos_sdk.actions import ActionsRegistry, ActionsClient

custom_registry = ActionsRegistry()
custom_registry.register(read_voltage, name="read_voltage")

client = ActionsClient()
client.serve("power_management", actions_registry=custom_registry)  # power_management/read_voltage
init(trace=False, block=True)  # serve() does not block; keep the process alive

ActionsRegistry::register takes the path and an Arc<dyn Action>. A path can hold namespaces, such as "bms/get_status".

ActionsClient::new_with_url takes the agent URL, such as grpc://127.0.0.1:2300. The Rust SDK reads no environment variable for it. serve takes the service name, an Arc<ActionsRegistry> and a CancellationToken. It runs until you cancel the token. After a connection error, it waits 5 seconds and reconnects. The Quick Start shows the full program.

ActionsRegistry.Register takes the path and an Action. A path can hold namespaces, such as "bms/get_status".

DefaultActionsClientConfig reads the agent URL from ZELOS_URL, or uses grpc://127.0.0.1:2300 when that variable is unset. Set config.URL to choose another agent. NewActionsClient takes a context, the service name, the registry and the config. Run serves until the context is canceled or you call Close, and it reconnects after config.ReconnectDelay (5 seconds by default). WaitUntilConnected and IsConnected report the connection state. The Quick Start shows the full program.

Results and Statuses

Every run ends with a value and one of four statuses:

Status	Meaning
PASS	The action completed and its check passed.
FAIL	The action completed and its check failed.
ERROR	The action could not complete.
DONE	The action completed with no pass/fail classification.

PythonRustGo

A plain return value is a DONE result, and None becomes {}. Return numbers, strings, booleans, or JSON-serializable dicts and lists. Return an ActionExecuteResult to set the status yourself. Its constructors are passed, failed, error and done, and its status is an ActionExecuteStatus.

from zelos_sdk import action
from zelos_sdk.actions import ActionExecuteResult

@action("Check Power Safety")
@action.number("voltage", minimum=300, maximum=500)
@action.number("current", minimum=-200, maximum=200)
def check_power_safety(voltage: float, current: float) -> ActionExecuteResult:
    power_w = voltage * abs(current)
    if power_w > 50_000:
        return ActionExecuteResult.failed(f"Power too high: {power_w:.0f}W (max 50kW)")
    return ActionExecuteResult.passed(f"OK: {power_w:.0f}W")

An exception ends the run with an ERROR result that carries its message. For a cross-field rule, raise a ValidationError with the field name and a message:

from zelos_sdk import action
from zelos_sdk.actions import ValidationError

@action("Configure Converter")
@action.number("vin", minimum=12, maximum=400)
@action.number("vout", minimum=3.3, maximum=48)
@action.select("kind", choices=["boost", "buck"])
def configure_converter(vin: float, vout: float, kind: str):
    if kind == "boost" and vout <= vin:
        raise ValidationError("vout", "Boost requires vout > vin")
    if kind == "buck" and vout >= vin:
        raise ValidationError("vout", "Buck requires vout < vin")
    return {"vin": vin, "vout": vout, "kind": kind}

Return ActionExecuteResult::passed, failed, error or done. Each takes a &serde_json::Value. Return Err(ActionsError) to end the run with an ERROR result whose value is {"error": "<message>"}.

use serde_json::{json, Value};
use zelos::{ActionExecuteResult, ActionFn, ActionSchema, ActionsError};

let check_power = ActionFn::new(
    ActionSchema::new("Check Power Safety", "Compare power against a 50 kW limit")
        .number("voltage", |f| f.minimum(300.0).maximum(500.0).required())
        .number("current", |f| f.minimum(-200.0).maximum(200.0).required()),
    |params: Value| async move {
        let field = |name: &str| {
            params
                .get(name)
                .and_then(Value::as_f64)
                .ok_or_else(|| ActionsError::InvalidInput(format!("missing '{name}'")))
        };
        let power_w = field("voltage")? * field("current")?.abs();
        if power_w > 50_000.0 {
            return Ok(ActionExecuteResult::failed(&json!({ "power_w": power_w })));
        }
        Ok(ActionExecuteResult::passed(&json!({ "power_w": power_w })))
    },
);

Return zelos.ActionResultPass, ActionResultFail, ActionResultError or ActionResultDone. Each takes any value that encoding/json can marshal. Return a non-nil error to end the run with an ERROR result whose value is {"error": "<message>"}.

import (
    "context"
    "encoding/json"
    "fmt"
    "math"

    "github.com/zeloscloud/zelos/go"
)

schema := zelos.NewActionSchema("Check Power Safety", "Compare power against a 50 kW limit").
    Number("voltage", zelos.Minimum(300), zelos.Maximum(500), zelos.Required()).
    Number("current", zelos.Minimum(-200), zelos.Maximum(200), zelos.Required())

checkPower := zelos.NewActionFromSchema(schema, func(ctx context.Context, params json.RawMessage) (*zelos.ActionResult, error) {
    var p struct {
        Voltage *float64 `json:"voltage"`
        Current *float64 `json:"current"`
    }
    if err := json.Unmarshal(params, &p); err != nil {
        return nil, err
    }
    if p.Voltage == nil || p.Current == nil {
        return nil, fmt.Errorf("missing 'voltage' or 'current'")
    }
    powerW := *p.Voltage * math.Abs(*p.Current)
    if powerW > 50_000 {
        return zelos.ActionResultFail(map[string]any{"power_w": powerW}), nil
    }
    return zelos.ActionResultPass(map[string]any{"power_w": powerW}), nil
})

To run an action from your own code and read its result, see Run Actions.

Timeouts

An action can declare a default timeout. A caller can pass its own timeout, which replaces the default for that run. When a run passes its timeout, the caller gets an ERROR result: {"error": "Action timed out after <n>ms"}. When the caller names no timeout, the agent waits up to 15 minutes for the result.

PythonRustGo

Every action has a default timeout of 30 seconds. Set another with timeout, in seconds:

from zelos_sdk import action

@action("Calibrate", "Calibrate one channel", timeout=120)
@action.integer("channel", minimum=0, maximum=7)
def calibrate(channel: int):
    # ... run the calibration ...
    return {"channel": channel}

Python cannot stop a running function. After a timeout, your function keeps running until it returns, and its result is discarded.

An ActionFn has no default timeout. To set one, implement the Action trait and return the timeout from default_timeout_ms. When the timeout passes, the SDK drops the action's future.

use async_trait::async_trait;
use serde_json::{json, Value};
use zelos::{Action, ActionExecuteResult, ActionSchema, ActionsError};

struct Calibrate;

#[async_trait]
impl Action for Calibrate {
    async fn execute(&self, params: Value) -> Result<ActionExecuteResult, ActionsError> {
        let channel = params
            .get("channel")
            .and_then(Value::as_u64)
            .ok_or_else(|| ActionsError::InvalidInput("missing 'channel'".to_string()))?;
        // ... run the calibration ...
        Ok(ActionExecuteResult::passed(&json!({ "channel": channel })))
    }

    fn get_schema_json(
        &self,
        _current_values: Option<Value>,
    ) -> Result<(String, String, String), ActionsError> {
        Ok(ActionSchema::new("Calibrate", "Calibrate one channel")
            .integer("channel", |f| f.minimum(0.0).maximum(7.0).required())
            .to_schema_json())
    }

    fn default_timeout_ms(&self) -> Option<u32> {
        Some(120_000) // 2 minutes
    }
}

This needs the async-trait crate. Register it like any other action: registry.register("calibrate".to_string(), Arc::new(Calibrate)).

An action from NewActionFromSchema or NewAction has no default timeout. To set one, implement the Action interface and return the timeout from DefaultTimeout. When the timeout passes, the SDK cancels the ctx passed to Execute. Return promptly when ctx is done.

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/zeloscloud/zelos/go"
)

type calibrate struct{}

var calibrateSchema = zelos.NewActionSchema("Calibrate", "Calibrate one channel").
    Integer("channel", zelos.Minimum(0), zelos.Maximum(7), zelos.Required())

func (calibrate) Execute(ctx context.Context, params json.RawMessage) (*zelos.ActionResult, error) {
    var p struct {
        Channel *int `json:"channel"`
    }
    if err := json.Unmarshal(params, &p); err != nil {
        return nil, err
    }
    if p.Channel == nil {
        return nil, fmt.Errorf("missing 'channel'")
    }
    // ... run the calibration, returning early when ctx is done ...
    return zelos.ActionResultPass(map[string]any{"channel": *p.Channel}), nil
}

func (calibrate) Schema(currentValues json.RawMessage) (string, string, error) {
    return calibrateSchema.JSONSchema(), calibrateSchema.UISchema(), nil
}

func (calibrate) DefaultTimeout() time.Duration {
    return 2 * time.Minute
}

Register it like any other action: registry.Register("calibrate", calibrate{}).

Custom Field Types

Custom field types are Python only. Register custom types for domain-specific validation.

import re

from zelos_sdk import action
from zelos_sdk.actions import BaseField, FieldType, ValidationError, register_field_type

@register_field_type("can_id")
class CANIdField(BaseField):
    def __init__(self, name: str, extended: bool = False, **kwargs):
        super().__init__(name, **kwargs)
        self.field_type = FieldType.STRING
        self.extended = extended

    def validate(self, value, form_data=None):
        value = super().validate(value, form_data)
        if value is None:
            return value
        clean = value.replace("0x", "").replace("0X", "").upper()
        if not re.match(r"^[0-9A-F]+$", clean):
            raise ValidationError(self.name, "CAN ID must be hex", "format")
        limit = 0x1FFFFFFF if self.extended else 0x7FF
        if int(clean, 16) > limit:
            raise ValidationError(self.name, "CAN ID out of range", "range")
        return value

@action("Configure CAN Message")
@action.input("message_id", type="can_id", extended=False)
@action.text("signal_name", pattern=r"^[a-zA-Z][a-zA-Z0-9_]*$")
@action.integer("data_length", minimum=1, maximum=8)
def configure_can_message(message_id: str, signal_name: str, data_length: int):
    return {"message_id": message_id, "signal_name": signal_name, "dlc": data_length}

Auto-Discovery via Entry Points

Entry-point discovery is Python only. Expose actions automatically when your package is installed.

Add to pyproject.toml:

[project.entry-points."zelos_sdk.actions"]
my_tools = "my_tools.actions"

In my_tools/actions.py:

from zelos_sdk import action

@action("Read VIN")
@action.text("vin", pattern=r"^[A-HJ-NPR-Z0-9]{17}$")
def decode_vin(vin: str) -> dict:
    return {"vin": vin, "manufacturer": vin[:3]}

Discovery imports this module to scan it, so do not call init() here. The host process serves the actions.

Best Practices

• Keep each action within its timeout. In Python, raise the 30-second default with @action(..., timeout=120).
• In Python, write action functions as def, not async def. An async def action raises TypeError.
• In Rust and Go, check every input in the handler. The SDK passes inputs through unchecked.
• In Python, prefer explicit validators (minimum, pattern, choices) for better UX.
• Keep hardware and session state on an object: a Python class, or a Rust or Go type that implements Action.
• Name registrations to keep large action sets organized.