roboto.domain.actions.action

Create a new action in the Roboto platform.

Creates a new action with the specified configuration. The action will be owned by the caller’s organization and can be invoked to process data.

Usage

Create a simple action:

action = Action.create(
    name="hello_world", uri="ubuntu:latest", description="A simple hello world action"
)

Create an action with parameters and compute requirements:

from roboto.domain.actions import ComputeRequirements, ActionParameter
action = Action.create(
    name="data_processor",
    uri="my-registry.com/processor:v1.0",
    description="Processes sensor data with configurable parameters",
    compute_requirements=ComputeRequirements(vCPU=4096, memory=8192),
    parameters=[
        ActionParameter(name="threshold", required=True, description="Processing threshold"),
        ActionParameter(name="output_format", default="json", description="Output format"),
    ],
    tags=["data-processing", "sensors"],
    timeout=60,
)

Create an action that inherits from another:

base_action = Action.from_name("base_processor", owner_org_id="roboto-public")
derived_action = Action.create(
    name="custom_processor",
    inherits=base_action.record.reference,
    description="Custom processor based on base_processor",
    metadata={"version": "2.0", "team": "data-science"},
)

Delete this action from the Roboto platform.

Permanently removes this action and all its versions. This operation cannot be undone.

Usage

Delete an action:

action = Action.from_name("old_action")
action.delete()

Load an existing action by name.

Retrieves an action from the Roboto platform by its name and optionally a specific version digest. Action names are unique within an organization, so a name + org_id combination always provides a fully qualified reference to a specific action.

Usage

Load the latest version of an action:

action = Action.from_name("data_processor")

Load a specific version of an action:

action = Action.from_name("data_processor", digest="abc123def456")

Load an action from another organization:

action = Action.from_name("public_processor", owner_org_id="roboto-public")

Invokes this action using any inputs and options provided.

Executes this action with the specified parameters and returns an Invocation object that can be used to track progress and retrieve results.

Usage

Basic invocation with a dataset:

from roboto import Action, InvocationSource
action = Action.from_name("ros_ingestion", owner_org_id="roboto-public")
iv = action.invoke(
    invocation_source=InvocationSource.Manual,
    data_source_id="ds_12345",
    data_source_type=InvocationDataSourceType.Dataset,
    input_data=["**/*.bag"],
    upload_destination=InvocationUploadDestination.dataset("ds_12345"),
)
iv.wait_for_terminal_status()

Invocation with compute requirement overrides:

from roboto import Action, InvocationSource, ComputeRequirements
action = Action.from_name("image_processing", owner_org_id="roboto-public")
compute_reqs = ComputeRequirements(vCPU=4096, memory=8192)
iv = action.invoke(
    invocation_source=InvocationSource.Manual,
    compute_requirement_overrides=compute_reqs,
    parameter_values={"threshold": 0.75},
)
status = iv.wait_for_terminal_status()
print(status)
# 'COMPLETED'

Query actions with optional filtering and pagination.

Searches for actions based on the provided query specification. Can search within an organization or across the public Action Hub.

Usage

Query all actions in your organization:

for action in Action.query():
    print(f"Action: {action.name}")

Query actions with specific tags:

from roboto.query import QuerySpecification
spec = QuerySpecification().where("tags").contains("ml")
for action in Action.query(spec):
    print(f"ML Action: {action.name}")

Query public actions in the Action Hub:

from roboto.domain.actions import Accessibility
spec = QuerySpecification().where("name").contains("ros")
for action in Action.query(spec, accessibility=Accessibility.ActionHub):
    print(f"Public ROS Action: {action.name}")

Query with pagination:

spec = QuerySpecification().limit(10).order_by("created", ascending=False)
recent_actions = list(Action.query(spec))
print(f"Found {len(recent_actions)} recent actions")

Set the accessibility level of this action.

Changes whether this action is private to the organization or published to the public Action Hub.

Usage

Make an action public in the Action Hub:

from roboto.domain.actions import Accessibility
action = Action.from_name("my_action")
action.set_accessibility(Accessibility.ActionHub)

Make an action private to the organization:

action.set_accessibility(Accessibility.Organization)

Convert this action to a dictionary representation.

Usage

Get action as dictionary:

action = Action.from_name("my_action")
action_dict = action.to_dict()
print(action_dict["name"])
# 'my_action'

Update this action with new configuration.

Updates the action with the provided changes. Only specified parameters will be modified; others remain unchanged. This creates a new version of the action.

Usage

Update action description and timeout:

action = Action.from_name("my_action")
action.update(description="Updated description of what this action does", timeout=45)

Update compute requirements:

from roboto.domain.actions import ComputeRequirements
action.update(compute_requirements=ComputeRequirements(vCPU=4096, memory=8192))

Add metadata using changeset:

from roboto.updates import MetadataChangeset
changeset = MetadataChangeset().set("version", "2.0").set("team", "ml")
action.update(metadata_changeset=changeset)