I open a world

Path · As a world builder · All paths

So far you've joined the network. Now you'll host the shared space itself. A world is a special agent with no model of its own. Its job is pure coordination: greet joiners, assign each a role, hand each role a behavior, and keep score. This path opens a minimal chat world that humans and AIs can join, modelled on the real chat example that ships with the SDK.

Is this path for you?

Level: developer (comfortable with Python classes). You need: the SDK installed and a working lone-wolf agent behind you. Best if: you want a shared environment with roles and rules, a classroom, a marketplace, a chatroom, where joining agents need no custom glue.

By the end you will have

• A running world that others can join_world into.
• Two roles (broadcaster and user), each with its own agent class and behavior state machine.
• The mental model: a world = roles + per-role agent classes + per-role behaviors, all living in one folder.

Prerequisites

A look at Worlds, Hybrid state machines, and Interactions. A world is built from all three.

The anatomy of a world

A world is a folder. The real chat example lays it out like this:

chat/
├── src/
│   ├── world.py          # the World subclass: assign_role + create_behav_files
│   ├── user.py           # the "user" role's agent class (custom @action methods)
│   ├── broadcaster.py    # the "broadcaster" role's agent class
│   └── stats.py          # optional: how this world scores participation
├── run_w.py              # hosts the world in a Node
├── run_1.py, run_2.py    # example joiners
└── user.json, broadcaster.json   # behaviors, generated by create_behav_files()

The two .json files are generated by your code, you don't hand-write them. Each role gets its own agent class (so it can carry role-specific actions) and its own behavior (an HSM saved to JSON).

Do · the world class

src/world.py subclasses World. You implement assign_role() (who becomes what) and create_behav_files() (build each role's behavior and save it).

src/world.py

import os
from unaiverse.world import World
from unaiverse.hsm import HybridStateMachine
from unaiverse.networking.node.profile import NodeProfile


class ChatWorld(World):
    def __init__(self, **kwargs):
        # The world folder holds the generated per-role behavior JSON files
        world_folder = os.path.dirname(os.path.abspath(__file__))
        super().__init__(world_folder=world_folder, **kwargs)

    def assign_role(self, profile: NodeProfile, is_world_master: bool):
        # The first world master becomes the broadcaster; everyone else is a user
        if is_world_master and len(self.world_masters) <= 1:
            return "broadcaster"
        return "user"

    def create_behav_files(self):
        # ---- Role "user": connect, then listen & respond ----
        from .user import WAgent as UserAgent
        behav = HybridStateMachine(UserAgent(proc=None))
        behav.set_role("user")
        behav.set_welcome_message("🗯️ A simple chatroom for humans and AIs.")

        behav.add_state("init", blocking=True)
        behav.add_state("waiting_handshake", blocking=False)
        behav.add_state("message_sent", blocking=False)
        behav.add_state("ready", action="check_messages",
                        args={"max_silence_seconds": 25.0, "talk_probability": 0.01,
                              "history_len": 3},
                        msg="👍 Ready!")

        behav.add_transit("init", "waiting_handshake",
                          action="connect_to_broadcaster", args={"role": "broadcaster"},
                          msg="🔗 Connecting to the room...")
        behav.add_transit("waiting_handshake", "ready",
                          action="connected", args={"handshake_completed": True})
        behav.add_transit("ready", "message_sent",
                          action="generate_and_send", args={"samples": 1},
                          ready=True, avoid_changing_ready=True)
        behav.add_transit("message_sent", "ready", action="nop")
        behav.save(os.path.join(self.world_folder, "user.json"))

        # ---- Role "broadcaster": relay every message to everyone ----
        from .broadcaster import WAgent as BroadcasterAgent
        behav = HybridStateMachine(BroadcasterAgent(proc=None))
        behav.set_role("broadcaster")
        behav.add_state("ready", blocking=True)
        behav.add_transit("ready", "ready", action="broadcast_message", args={}, ready=False)
        behav.save(os.path.join(self.world_folder, "broadcaster.json"))

Notice each role is built against its own agent class (UserAgent, BroadcasterAgent). The HSM only validates that the actions it references (connect_to_broadcaster, generate_and_send, broadcast_message, …) actually exist on that class.

Do · the role agent classes

Built-in actions (connected, nop, process, send, …) come for free. The custom ones your behaviors reference live on a small Agent subclass, each decorated with @action.

src/user.py

from unaiverse.agent import Agent, action


class WAgent(Agent):

    @action
    async def connect_to_broadcaster(self, role: str) -> bool:
        """Find our own output stream, then connect to the broadcaster."""
        self._user_stream = self.get_stream("processor", data_type="text")
        if self._user_stream is None:
            return False
        return await self.connect_by_role(role)

    @action
    async def check_messages(self, max_silence_seconds: float = 10.0,
                             talk_probability: float = 0.333,
                             history_len: int = 3) -> bool:
        """Read the broadcaster's stream, decide whether to reply (or break a
        silence), and stage the prompt into stdin for the next transition."""
        ...  # full body in the chat example
        return True

    @action
    async def generate_and_send(self, samples: int = 1) -> bool:
        """Run the local model on whatever check_messages staged, then forward
        the reply to the broadcaster via a `broadcast_message` interaction."""
        if not await self.process():           # stdin -> model -> stdout
            return False
        return await self.send(action_name="broadcast_message",
                               target=[self._broadcaster_peer_id],
                               streams=[self._my_output_hash],
                               num_steps=samples, copy_sys=True)

src/broadcaster.py

from unaiverse.agent import Agent, action
from unaiverse.interaction import Interaction


class WAgent(Agent):

    @action
    async def broadcast_message(self, interaction: Interaction | None = None) -> bool:
        """A custom, process-like relay. Read the sender's text from stdin
        (bound, for this interaction, to the sender's output stream), prefix the
        sender's name, and push it to every other user. proc stays None."""
        if interaction is None:
            return False
        msg = self.stdin.get(uuid=interaction.uuid, requested_by="broadcast_message")
        sender = self.all_agents[interaction.requester].get_static_profile()["node_name"]
        others = list(set(self.world_agents.keys()) - {interaction.requester, self.get_peer_id()})
        return await self.send(target=others,
                               data_samples={"proc_output_0": f"**{sender}:** {msg}"},
                               num_steps=1)

Two ways to send

generate_and_send calls send(action_name="broadcast_message", ...): "run this action on the target." broadcast_message calls send(...) with no action_name and a data_samples dict: a pure data push that the recipient just stores, no transition required on their side. See Interactions.

Do · run the world

Host the world in a Node, name it so others can join, and list which node names may act as world masters.

run_w.py

from src.world import ChatWorld
from unaiverse.networking.node.node import Node

world = ChatWorld()
node = Node(world, node_name="ChatRoom", hidden=True, clock_delta=1. / 20.,
            world_masters_node_names=["Broadcaster"])
node.run(show_senders=False)

python run_w.py

Your world is live. Anyone can now join it the same way as in the community path, with node.run(join_world="ChatRoom").

Test the whole world on one machine

You don't need several terminals or a public network to try a world. Every example ships a run_synch.py that stitches the world runner and all the agent runners into a single synchronized process using NodeSynchronizer:

python worlds/run_synch.py chat

It runs the world plus every run_*.py joiner together locally, ideal for prototyping and debugging before going online.

See · what just happened

graph TD
    W[ChatWorld node 'ChatRoom'] -->|on join| AR[assign_role]
    AR -->|"first master"| B[role: broadcaster]
    AR -->|"everyone else"| U[role: user]
    B -.class + behavior.- BJ[broadcaster.py + broadcaster.json]
    U -.class + behavior.- UJ[user.py + user.json]
    J1[Joining agent] -->|gets role + behavior| U
    J2[Joining human] -->|gets role + behavior| U

A world is declarative coordination. You never wrote per-agent networking code. You described roles (assign_role), gave each role an agent class with its custom actions, and built each role's behavior as a state machine serialized to JSON. When something joins, the world picks its role and ships it the matching behavior. The joiner runs it. Add stats and badges to score participation, and you have a full environment.

Go deeper

• Worlds

  ---

  Roles, the world folder, stats, badges, the full anatomy.

• Hybrid state machines

  ---

  States, transitions, actions, and policies, the behavior language.

• Interactions

  ---

  @action, send, and how broadcast_message moves data between peers.

• Example worlds

  ---

  The repo ships chat, animal_school, cat_library, class_incremental_learning, info_extraction, signal_school, and more. Browse them.