Skip to content

5 · Creating actions

Build worlds · Chapter 5 of 12 · Path home

This is the heart of the path. A behavior is a graph of states; the actions are what actually happen at each step. UNaIVERSE gives you many built-in actions, but the interesting worlds are made of the actions you write.

What an action really is

An action is the unit of what any agent can do. Not "a step in an LLM pipeline", just a thing an agent does: move a motor, read a sensor, check a rule, call a service, ask a person, run a model. If an agent can do it, it is an action.

By the end of this chapter you will know exactly what @action requires of you, what it does for you, how arguments reach your code, how to read and write data, and how to wire your action into a behavior, with a real example pulled straight from a shipped world.

The @action decorator

You add an action by writing a method on an Agent subclass and decorating it with @action:

from unaiverse.agent import Agent, action


class MyAgent(Agent):

    @action
    async def do_something(self) -> bool:
        ...           # whatever this agent does
        return True   # success: advance the state machine

That is the whole shape. Three rules define a valid action, and the rest of this chapter is about using them well:

  • It is async (async def), the framework awaits it on each clock tick.
  • It returns a bool: True for success, False to be retried (next section).
  • It lives on a class that subclasses Agent, so the behavior that owns it can find it by name and check at build time that it really exists.

Behind the scenes the decorator tags the method with _is_action = True, which is how the framework discovers every action on your class by introspection. You never set that yourself.

The success / retry contract

The return value is a contract with the state machine, and it is small and total:

You return What happens
True (or any non-None, non-False value) The action succeeded; the transition fires and the behavior moves to the next state.
False The action is not done yet; the agent stays put and the action is retried on the next clock tick.
raises an exception The decorator catches it, logs it, and treats it as False, the agent retries instead of crashing.
None A mistake. The framework logs a critical warning: an action must decide True/False, and returning nothing means you forgot a return.

The retry is your friend, not an error path

Returning False is the normal way to wait. The agent's clock ticks many times per second; an action that is "still waiting for the arm to arrive", "still waiting for the human to click", or "no new sensor reading yet" simply returns False and gets called again. You never write a wait loop, a sleep, or a callback yourself, returning False is the loop. Combine this with blocking states from Chapter 4 to control the pacing.

Do not block the tick

Because the framework awaits your action on every tick, a slow synchronous call (a long HTTP request, a heavy file read, a time.sleep) freezes the agent until it returns. For anything that can take a while, prefer the "start it, then poll with False until it is ready" shape, or await a proper async call. Long blocking work belongs behind a check, not inline.

What the decorator does for you

@action is not just a tag. Every time your action is called it quietly does three useful things, so your code can read in plain terms. Here is each one, on its own, with the simplest code that shows it.

1 · Friendly agent names to peer IDs

Any argument named agent, agents, partner, partners, target, or targets is turned from a human-readable name (or "email@node_name" notation) into the underlying peer ID before your code runs. You name people the way you think about them; the framework hands you the real identifier.

# In the behavior, you name the agent in plain language:
behav.add_transit("idle", "greeting", action="greet", args={"agent": "Alice"})


class GreeterAgent(Agent):

    @action
    async def greet(self, agent: str) -> bool:
        # You wrote "Alice" in the behavior.
        # Here 'agent' is ALREADY the resolved peer ID, no lookup to do yourself.
        await self.send(action_name="show_message", target=agent)
        return True

2 · Friendly stream names to canonical hashes

Any argument named stream or streams is resolved from a stream's friendly name to its canonical network hash, and the framework automatically picks the public or the in-world stream depending on where the agent is currently behaving.

# In the behavior, refer to the stream by its plain name:
behav.add_transit("idle", "watching", action="watch", args={"stream": "temperature"})


class WatcherAgent(Agent):

    @action
    async def watch(self, stream: str) -> bool:
        # You wrote "temperature".
        # Here 'stream' is ALREADY the canonical network hash for the right stream.
        reading = self.stdin.get(requested_by="watch")
        return reading is not None

3 · Exception isolation

If your body raises, the decorator logs the error and returns False for you (see the contract table above). One buggy action never takes the whole agent down, it just gets retried on the next tick.

class SensorAgent(Agent):

    @action
    async def read_sensor(self) -> bool:
        value = read_hardware()      # if this line raises (cable unplugged?)...
        self.stdout.set(value)
        return True
        # ...you write NO try/except here. @action catches it, logs it,
        # and returns False. The agent stays alive and retries next tick.

send resolves its own arguments

The built-in send is a special action: it does its own reference resolution internally, so the generic name-resolution step is skipped for it. You will almost never notice, it is mentioned only so the mechanism is not a surprise when you read the framework code.

Where arguments come from

An action's parameters are filled from three places:

  • The behavior. Whatever you put in a transition's args={...} (or a state's args={...}) is passed to your action as keyword arguments, this is the main channel. Defaults in your signature cover anything the behavior omits.
  • The framework-injected interaction. If your signature declares an interaction parameter, the framework hands you the live Interaction object that triggered the action (the request, its streams, its uuid). Declaring the parameter is how you opt in; you never pass interaction yourself in args.
  • Auto-resolved references. The agent/stream arguments above arrive already resolved, as just described.

behav.add_transit("idle", "answering", action="answer_from_db",
                  args={"table": "customers"})   # 'table' -> your kwarg

@action
async def answer_from_db(self, table: str = "default",
                         interaction=None) -> bool:   # 'interaction' injected
    query = self.stdin.get(uuid=interaction.uuid, requested_by="answer")
    self.stdout.set(my_db.search(table, query), uuid=interaction.uuid)
    return True

A few argument names are reserved for timing, not your body: max_duration, retry_timeout, and delay are consumed by the transition itself and stripped before your action sees them. These are the same three timing controls you set on a transition in Chapter 4, where add_transit exposes them as the keyword parameters total_time, timeout, and delay. So you normally pass them as add_transit(..., total_time=..., timeout=...), and the reserved names above are simply what they are called if they ever appear inside an action's args dict.

Reading input, doing work, producing output

Inside an action you can do three kinds of thing:

  • Read what arrived, with self.stdin.get().
  • Write a result, with self.stdout.set(value).
  • Ask others to do something, with self.send(...) (the whole of Chapter 6).

Start from the bare forms: get() returns the agent's input, set(value) writes its output, no slot name needed. You only reach for a name (get("proc_input_0")) when an agent has more than one input or output, and for the uuid / requested_by kwargs below when a read must be scoped to one specific request. New to self.stdin / self.stdout? They are an agent's fixed input/output slots, and An agent's own streams walks the whole convention, simplest form first.

That is the entire surface. Now the point of this chapter: those three primitives have nothing to do with AI. Here is the same idea across very different agents.

Focus · how stdin.get and stdout.set really move data

These two look trivial but carry two kwargs that decide which data you touch:

  • uuid scopes a read or write to one interaction. When your action was triggered by a request (it received an interaction), pass uuid=interaction.uuid so you read the data that request carried and write your answer back under the same id, like answer_from_db above. Leave it out for a plain, interaction-free read or write, like the sensor publishing.
  • requested_by gives single-delivery: each fresh sample is handed to a given reader once. Call get again with the same requested_by and you get None until a new value is written (pass requested_by=None to always re-read the latest). That is why water_if_dry and await_approval guard with is not None: None just means "nothing new since I last looked", so the action returns and is retried next tick.

stdout.set(value) publishes one sample on your output stream for anyone to read; add uuid=interaction.uuid and you are answering one specific request instead of broadcasting. Pass requested_by=None to get when you want the latest value every time (e.g. a controller re-reading a sensor each tick).

Examples that are not about AI

An agent that controls a robot arm. The action commands the hardware and reports True only once the arm has arrived, so the behavior waits for the move to finish before the next step. This is the start-then-poll shape: the False return is the wait.

class ArmAgent(Agent):

    @action
    async def move_arm_to(self, x: float, y: float, z: float) -> bool:
        arrived = my_robot.move_to(x, y, z)   # your hardware/driver call
        return bool(arrived)                  # False -> retry next tick

An input agent that reads the physical world and publishes the value as a stream others can use. No model, no inputs.

class MoistureAgent(Agent):

    @action
    async def read_moisture(self) -> bool:
        value = read_soil_sensor()            # your hardware function
        self.stdout.set(value)                # publish the reading
        return True

A plain rule, no learning. Read the latest value, decide, act. This is a complete agent: condition in, action out.

class IrrigationAgent(Agent):

    @action
    async def water_if_dry(self, threshold: float = 0.3) -> bool:
        moisture = self.stdin.get(requested_by="water_if_dry")
        if moisture is not None and moisture[0] < threshold:
            open_valve()                      # your actuator
        return True

An agent whose job is to look something up: a database, an HTTP API, a knowledge base. It answers one request, so it scopes every read and write to interaction.uuid.

class LookupAgent(Agent):

    @action
    async def answer_from_db(self, interaction=None) -> bool:
        query = self.stdin.get(uuid=interaction.uuid, requested_by="answer")
        self.stdout.set(my_database.search(query), uuid=interaction.uuid)
        return True

An action that defers to a person and returns once they have decided. Human agents are first-class here. While the person has not answered, get returns None, the action returns False, and the agent simply waits.

class ApproverAgent(Agent):

    @action
    async def await_approval(self) -> bool:
        decision = self.stdin.get(requested_by="approval")  # what the person typed
        return decision is not None                         # wait until they answer

And, of course, an agent whose action is to run a model. Even this is just an action, and you usually get it for free as the built-in process.

class ModelAgent(Agent):

    @action
    async def classify(self, interaction=None) -> bool:
        return await self.process(interaction)   # stdin -> model -> stdout

The point

A sensor, a valve, a rule, a database, a human, a neural network: in a world they are all agents, and each thing they do is an action. UNaIVERSE is not "a model plus a runner". It is a place where anything that can act is a peer, and your job as a world builder is to describe what each kind of peer does.

A real action, end to end

The examples above are deliberately tiny. Real actions do more, and the framework lets them, because an action is just a method on your class. Two patterns appear in almost every shipped world.

An action can keep state on self and call ordinary helper methods. It does not have to be self-contained. Here is a trimmed action from the info_extraction world: it runs at the start of every cycle, keeps accumulated results on the instance, and persists them to disk, plain Python, no special API.

class WAgent(Agent):

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self._extracted_info = {}       # state lives on the instance
        self._got_new_info = False

    @action
    async def check_status(self) -> bool:
        await self.disengage_all()                  # built-in helpers...
        await self.disconnect_by_role("extractor")  # ...called from your action

        if self._got_new_info:                      # ordinary Python state + I/O
            self._got_new_info = False
            with open("extracted_info.json", "w") as f:
                json.dump(self._extracted_info, f, indent=4)
        return True

The instance fields (self._extracted_info, self._got_new_info) survive across ticks, so an action can build something up over time. It can also call non-action helper methods on the same class (plain def, no decorator) for parsing, filtering, or formatting, only the methods you decorate become behavior steps.

Idiom · stage the input now, run the model next

A common two-step pattern: one action prepares a model prompt by writing it into a named input stream, and the next transition runs the model on it. From the chat world's user role:

@action
async def check_messages(self, ...) -> bool:
    # ...decide there is something worth replying to, then:
    self.stdin.set("proc_input_0", augmented_msg)   # stage the prompt
    return True

@action
async def generate_and_send(self, ...) -> bool:
    ok = await self.process()                       # stdin -> model -> stdout
    if not ok:
        return False
    return await self.send(action_name="broadcast_message", ...)  # Chapter 6

Splitting "decide + stage" from "run the model + send" keeps each state's job small and lets the behavior interleave other transitions between them.

Using your action in a behavior

A custom action is referenced from a transition by name, exactly like a built-in:

behav.add_state("idle", blocking=True)
behav.add_transit("idle", "moving", action="move_arm_to",
                  args={"x": 0.2, "y": 0.0, "z": 0.1})
behav.add_transit("moving", "idle", action="nop")

Because the behavior is built against the agent class that owns the action (Chapter 2), the machine can check at build time that move_arm_to really exists, a typo in the action name fails fast, not at runtime. The keys of args must match your method's parameters (minus the injected interaction and the timing names).

Pitfalls and a quick checklist

The mistakes everyone makes once

  • Forgetting return True. An action that falls off the end returns None and the framework flags it as a critical error. Always return a bool.
  • Returning True while still waiting. If you advance before the work is actually done (arm not arrived, human not answered), the behavior races ahead. Return False until you are sure.
  • Blocking the tick. A synchronous slow call freezes the agent. Poll with False, or await an async call.
  • requested_by confusion. With a fixed requested_by, a second get returns None until new data arrives, that is single-delivery, not a bug. Use requested_by=None when you want the latest value every tick.
  • Putting interaction in args. You never pass it; you declare it in the signature and the framework injects it.

Before you ship an action, check: it is async; it returns True/False on every path; long work is non-blocking; reads/writes use the right uuid and requested_by; and the name in add_transit(action=...) matches the method.

The actions you already have

Before writing an action, check whether a built-in already does it. Discovery, engagement, sending, processing, learning, evaluating, awarding badges, and more are provided, catalogued with every parameter in Built-in actions. Write your own for the parts that are unique to your world.

Where next