Actions

zelos_sdk.actions

Actions system for Zelos SDK

This module provides a comprehensive action definition, registration, and execution system.

Action(callable_obj, title, description=None, timeout=30.0, fields=None, instance=None)

Defines an executable action with input validation and schema generation.

This class wraps a callable (function or method) with metadata, parameter validation, and execution management. It supports both synchronous and asynchronous execution with timeout handling.

Example: >>> @action("Add Numbers", "Add two numbers") ... @action.number("x") ... @action.number("y") ... def add(x: int, y: int) -> int: ... return x + y

Initialize an action with its callable and metadata.

Parameters:

Name	Type	Description	Default
callable_obj	Callable[..., T]	Function or method to execute	required
title	str	Human-readable title for the action	required
description	Optional[str]	Optional description of what the action does	None
timeout	float	Maximum execution time in seconds	30.0
fields	Optional[List[BaseField]]	List of input field definitions	None
instance	Optional[Any]	Class instance if this is a bound method	None

Raises:

Type	Description
ValueError	If timeout is not positive or title is empty
TypeError	If callable_obj is async (not supported)

__repr__()

Returns:

Type	Description
str	A string representation of the Action

execute(**kwargs)

Execute the action with the given parameters.

Validates all inputs, checks required fields, and executes the underlying callable. Only synchronous functions are supported.

The method handles different return types: - Regular values: Returned as-is (success) - None: Returned as empty success result - ActionResult: Returned as-is (explicit status control) - Exceptions: Raised as ActionExecutionError (failure)

Parameters:

Name	Type	Description	Default
kwargs	Any	Keyword arguments to pass to the callable	{}

Returns:

Type	Description
Union[T, ActionResult]	Result of the callable execution or ActionResult

Raises:

Type	Description
ValidationError	If input validation fails or required fields are missing
ActionExecutionError	If execution fails

to_schema(current_values=None)

Build a JSON Schema for this action's inputs.

Parameters:

Name	Type	Description	Default
current_values	Optional[Dict[str, Any]]	Optional dict of current form data, used for dynamic choices	None

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the JSON Schema

to_ui_schema()

UI Schema for this action's inputs and provides hints for rendering form inputs in a user interface.

Returns:

Type	Description
Dict[str, Any]	A dictionary representing the UI Schema

ActionExecutionError

Bases: Exception

Exception raised when executing an action fails.

ActionResult(value, status)

Result wrapper for action execution that includes both the result value and execution status.

This class provides a clean way to return explicit success/failure results from actions while maintaining the simple return patterns for most use cases.

When Rust types are available, this class wraps the native Rust ActionResult.

Initialize an ActionResult.

Args: value: The result value (can be any JSON-serializable type) status: The execution status (PASS or FAIL)

error(value) classmethod

Create an error result.

failure(value) classmethod

Create a failure result.

success(value) classmethod

Create a successful result.

to_dict()

Convert to dictionary representation.

ActionValidationError

Bases: Exception

Exception raised for validation errors in actions.

ActionsClient

Python wrapper for the Actions gRPC client

cancel()

Stop the background serving task.

Returns: None

execute(action_path, params)

Execute an action.

Args: action_path (str): Full action path (e.g., "service/action" or "service/name/action") params (dict): Parameters for the action

Returns: object: Result from the action execution

list()

List available actions.

Returns: dict: Dictionary mapping service names to lists of action names

serve(name, actions_registry=...)

Start serving actions in a background task (non-blocking).

This method spawns a background task that handles the bidirectional stream and automatically reconnects if the stream fails.

Args: name (str): Name for this service actions_registry (ActionsRegistry): The actions registry to serve

ArrayField(name, item, additional_items=None, min_items=None, max_items=None, unique_items=False, **kwargs)

Bases: BaseField

Field for handling arrays of items, supporting both fixed and variable lengths.

to_ui_schema()

Extend the base UI schema with items and additionalItems if necessary.

validate(value, form_data=None)

Validate the array, enforcing constraints like min/max items, uniqueness, and item validations.

BaseField(name, title=None, description=None, required=True, default=None, widget=None, ui_options=None, placeholder=None, requires=None, **kwargs)

Base class for all fields.

to_json_schema(current_values=None)

Return this field's JSON Schema. Subclasses may override to include additional constraints.

to_ui_schema()

Return this field's UI Schema.

validate(value, form_data=None)

Validate the field's value. Subclasses should override to implement specific validations.

BooleanField(name, **kwargs)

Bases: BaseField

Field for boolean input.

validate(value, form_data=None)

Validate that the value is a boolean.

DateField(name, **kwargs)

Bases: TextField

Field for date input, inherits from TextField with a predefined pattern and format.

EmailField(name, **kwargs)

Bases: TextField

Field for email input, inherits from TextField with a predefined pattern.

ExecuteStatus

Bases: Enum

Enumeration of action execution status values.

FieldType

Bases: Enum

Enumeration of supported field types for action inputs.

has_value(value) classmethod

Check if a value is a valid field type.

FileField(name, accept=None, **kwargs)

Bases: BaseField

Field for uploading a single file as a data URL.

validate(value, form_data=None)

Validate that the value is a valid data URL with correct Base64 encoding.

FilesField(name, accept=None, **kwargs)

Bases: BaseField

Field for uploading multiple files as data URLs.

validate(value, form_data=None)

Validate that the value is a list of valid data URLs.

IntegerField(name, minimum=None, maximum=None, **kwargs)

Bases: BaseField

Field for integer input with optional min and max constraints.

NumberField(name, minimum=None, maximum=None, **kwargs)

Bases: BaseField

Field for numerical input (integer or float) with optional min and max constraints.

SelectField(name, choices, depends_on=None, **kwargs)

Bases: BaseField

Field for selecting options from a predefined list. Supports dynamic choices based on dependencies.

validate(value, form_data=None)

Validate that the selected value is among the allowed choices.

TextField(name, min_length=None, max_length=None, pattern=None, **kwargs)

Bases: BaseField

Field for text input with optional length and pattern constraints.

ValidationError(field_name, message, code='invalid')

Bases: Exception

Exception raised for validation errors in fields.

Parameters:

Name	Type	Description	Default
field_name	str	The name of the field where validation failed.	required
message	str	Description of the validation error.	required
code	str	Error code categorizing the type of validation error.	'invalid'

Widget

Bases: Enum

Enumeration of supported UI widgets for action inputs.

Widgets are grouped by their primary type and usage.

has_value(value) classmethod

Check if a value is a valid widget type.

create_field(name, type, **kwargs)

Factory method to create a field based on its type.

Parameters:

Name	Type	Description	Default
name	str	The name of the field.	required
type	str	The type of the field.	required
kwargs		Additional keyword arguments for field initialization.	{}

Returns:

Type	Description
BaseField	An instance of a subclass of BaseField.

Raises:

Type	Description
ValueError	If the field type is unsupported or required arguments are missing.

discover_actions(entry_point_group='zelos_sdk.actions', skip_errors=True, registry=None)

Discover and register all actions from entry points.

Parameters:

Name	Type	Description	Default
entry_point_group	str	Entry point group to scan for actions	'zelos_sdk.actions'
skip_errors	bool	If True, continue after errors; if False, raise exceptions	True

Returns:

Type	Description
Dict[str, List[str]]	Dictionary mapping entry point names to lists of registered actions

Raises:

Type	Description
ActionDiscoveryError	If discovery fails and skip_errors is False

get_global_actions_registry()

Get the global actions registry

init_global_actions_registry()

Initialize the global actions registry

register_field_type(type_name)

Decorator to register a field class with a given type name.