Events & Messages

Communication runs both ways. Clients send commands, declared with @event. Your model sends messages, declared as ModelMessage subclasses. Both are typed, and both end up in the schema that generates your clients’ SDKs.

Commands

@event marks a method as the handler for a named command. The method’s parameters become the command’s payload:

from reactor_runtime import InputField, event


@event(name="set_style", description="Rendering style applied to each frame")
async def set_style(
    self, style: str = InputField(default="realistic", choices=["realistic", "anime"])
) -> None:
    self._style_embedding = self.encoder.encode(style)

The client sends {"type": "set_style", "data": {"style": "anime"}} and the runtime validates the payload against the signature before calling you. Handlers can be async def or plain def. A command taking no parameters is fine, and useful for actions rather than settings:

@event(name="reset", description="Clear the scene and start over")
async def reset(self) -> None:
    self._step = 0

Constraining a parameter

InputField carries the constraints the runtime checks before your handler runs:

brightness: float = InputField(default=1.0, ge=0.0, le=2.0)
prompt: str = InputField(default="hello", min_length=1, max_length=500)
style: str = InputField(default="none", choices=["none", "oil_paint", "sketch"])

A value outside the constraint is rejected, and the client is answered with invalid_command. The runtime checks your defaults too, when the class is declared, so a default that violates its own constraint fails at startup rather than on the first request. Constraints also travel into the schema, so a generated SDK shows a client the range before it sends anything.

Replying to a command

A handler that returns a ModelMessage sends it back as that command’s correlated reply, so a client awaiting the command resolves with the result:

class StyleChanged(ModelMessage):
    style: str
    intensity: float


@event(name="set_style", description="Rendering style applied to each frame")
async def set_style(self, style: str = InputField(default="realistic")) -> StyleChanged:
    self._style = style
    # ✅ Returned, so it becomes the reply to this command
    return StyleChanged(style=self._style, intensity=self._intensity)

Return None when the client needs no answer. Use a reply when the client should confirm what actually took effect — clamped values, resolved defaults, or the full state after a change. The return annotation must be a single ModelMessage subclass or None. A union — including StyleChanged | None — is rejected when the class is declared, because the schema publishes one response shape and a client generated from it would expect no body and receive one.

Reporting a failure

Raise CommandError to answer a command with a failure the client can act on. The code and message cross the wire unchanged, correlated with the command, so an awaiting caller rejects with a reason instead of waiting for a reply that never comes:

from reactor_runtime import CommandError


@event(name="set_style", description="Rendering style applied to each frame")
async def set_style(self, style: str = InputField(default="realistic")) -> StyleChanged:
    if style not in self._available_styles:
        raise CommandError("unknown_style", f"No style named {style!r} is loaded.")
    self._style = style
    return StyleChanged(style=self._style, intensity=self._intensity)

Write the message for the client, not for a log. Any other exception is a fault the client cannot act on: the runtime logs it with its traceback and answers with a generic internal_error, keeping the detail — which can name paths, queries, or credentials — out of the reply. The runtime sends a few codes of its own on the same channel: invalid_command when a payload fails the model’s contract and no handler runs, and unresolved_upload when a command references an upload that cannot be fetched.

Outbound messages

A ModelMessage subclass declares a typed payload your model can send at any time. Fields are declared like a dataclass:

from reactor_runtime import MessageField, ModelMessage


class Progress(ModelMessage):
    """Emitted each frame while a generation is in flight."""

    step: int
    total: int = MessageField(default=100, description="Steps in the full run")

The client receives {"type": "progress", "data": {"step": 3, "total": 100}}. The wire type is the snake_case form of the class name. MessageField is optional — it exists to attach a default or a description that shows up in the generated schema. The class docstring documents the message itself. Send one with self.send():

async def run(self) -> None:
    while True:
        await self.connected.wait()
        while self.connected.is_set():
            frame = self.pipe.forward(prompt=self.prompt, step=self._step)
            await self.send(Progress(step=self._step))
            await self.emit(MyOutput(main_video=frame))
            self._step += 1

A message with no fields is valid and is the right shape for a pure signal, like GenerationComplete.

Per-client messages

self.send() broadcasts to every client in the session. To reach one client, take a client: ClientInfo parameter and use client.send(). The runtime injects the client that triggered the handler, so it never appears in your schema.

from reactor_runtime import ClientInfo


@event(name="get_state", description="Current scene parameters")
async def get_state(self, client: ClientInfo) -> None:
    # ✅ Reply to the client that asked, not to everyone watching
    await client.send(SceneState(prompt=self.prompt, step=self._step))

client is a handle you can keep. Store it in @connected and message that client later from anywhere:

@connected
async def on_connect(self, client: ClientInfo) -> None:
    self._clients[client.id] = client

# ✅ Later, from run() or another handler
await self._clients[client_id].send(Progress(step=self._step))

For a single-client model, self.send() covers everything. Reach for client.send() when a session has an audience and the message is meant for one of them.

File uploads

To receive a binary file with a command, annotate a parameter as UploadedFile. The runtime resolves the upload and hands your handler the bytes:

from reactor_runtime import UploadedFile


@event(name="set_reference_image", description="Image the style is drawn from")
async def set_reference_image(
    self,
    reference_image: UploadedFile,
    strength: float = InputField(default=0.5, ge=0.0, le=1.0),
) -> None:
    if not reference_image.mime_type.startswith("image/"):
        return
    self._reference = decode(reference_image.data)
    self._strength = strength

UploadedFile carries name, mime_type, data (the raw bytes), and size (their length). Check mime_type before decoding — a client can upload anything. A file that arrives with a command reaches that handler alone. The runtime resolves it first, so a handler runs with the bytes in hand, and an upload it cannot fetch answers the client with an unresolved_upload failure.

Uploads outside a command

A client can also send a file on its own. @file_uploaded receives those:

from reactor_runtime import ClientInfo, UploadedFile, file_uploaded


@file_uploaded
async def on_file(self, uploaded_file: UploadedFile, client: ClientInfo) -> None:
    if uploaded_file.mime_type.startswith("image/"):
        self._reference = decode(uploaded_file.data)

The handler takes exactly one parameter named uploaded_file, plus the optional client. The runtime fetches an upload’s bytes only when the model declares this hook, so it is what turns a bare notification into a file. Reach for it when a file has no natural command to ride along with — a drag-and-drop surface, or a client that uploads first and decides what to do with the result later. When the file belongs to an action, put it on that action’s @event instead, so the schema shows a client what the file is for.

Putting it together

class PromptChanged(ModelMessage):
    """Confirms the prompt the model is now rendering."""

    prompt: str


class Progress(ModelMessage):
    """Emitted each frame while a generation is in flight."""

    step: int
    total: int


class MyModel(ReactorModel):
    fps = 24

    def load(self, config_path: Path | None) -> None:
        self.pipe = load_checkpoint()
        self.prompt = "a sunny meadow"
        self._step = 0

    @event(name="set_prompt", description="Scene the model renders")
    async def set_prompt(
        self, prompt: str = InputField(default="", max_length=500)
    ) -> PromptChanged:
        self.prompt = prompt
        return PromptChanged(prompt=self.prompt)

    @event(name="get_prompt", description="Scene currently being rendered")
    async def get_prompt(self, client: ClientInfo) -> None:
        await client.send(PromptChanged(prompt=self.prompt))

    async def run(self) -> None:
        while True:
            await self.connected.wait()
            while self.connected.is_set():
                prompt = self.prompt
                frame = self.pipe.forward(prompt=prompt, step=self._step)
                await self.send(Progress(step=self._step, total=100))
                await self.emit(MyOutput(main_video=frame))
                self._step += 1

From the client’s side, this model:

accepts set_prompt and replies with the applied prompt_changed,
accepts get_prompt and answers the asking client alone,
broadcasts a progress message on every frame.

Managing State

One event loop, shared by your loop and your handlers.

Media Input

Read the client’s camera and microphone.

Get Started

Build a Model

Media & Transport

Releases

Events & Messages

Commands

Constraining a parameter

Replying to a command

Reporting a failure

Outbound messages

Per-client messages

File uploads

Uploads outside a command

Putting it together

Next

Managing State

Media Input

​Commands

​Constraining a parameter

​Replying to a command

​Reporting a failure

​Outbound messages

​Per-client messages

​File uploads

​Uploads outside a command

​Putting it together

​Next

Managing State

Media Input

Commands

Constraining a parameter

Replying to a command

Reporting a failure

Outbound messages

Per-client messages

File uploads

Uploads outside a command

Putting it together

Next