Skip to main content
Beyond auto-generated set_<field> events, you can define your own events, send messages back to the client, and run logic when clients connect or disconnect.

​
Custom event handlers

Use @event to define actions the client can trigger that aren’t tied to a single state field. The method’s signature becomes the event schema. 
from reactor_runtime.interface import event

@event(name="get_current_prompt", description="Log the current prompt")
async def get_current_prompt(self):
    print(f"Current prompt: {self.state.prompt}")
The client sends {"type": "get_current_prompt", "data": {}} and the runtime calls your method. Events can also take parameters: 
@event(name="set_style", description="Change the rendering style")
async def set_style(self, style: str = InputField(default="realistic")):
    self.state._style_embedding = self.encoder.encode(style)
In ReactorPipeline, events only dispatch between yield points, so self.state is always consistent within a single iteration.

​
Deduplicating expensive handlers

Pass dedupe=True to collapse duplicate events that queue up while the model is busy. Only the latest payload is processed; earlier ones are discarded. 
@event(
    name="set_reference_image",
    description="Upload a reference image",
    dedupe=True,
)
async def set_reference_image(self, image: str = ""):
    self.state._ref_embedding = self.encoder.encode(image)
Use this for handlers that do heavy compute (encoding) where only the latest value matters.

​
Lifecycle hooks

​
@connected

Runs once when a client connects, before inference() starts. 
from reactor_runtime.interface import connected

@connected
async def on_connect(self):
    print("Client connected, session starting")

​
@disconnected

Runs once when a client disconnects, after inference() is closed. 
from reactor_runtime.interface import disconnected

@disconnected
async def on_disconnect(self):
    print("Client disconnected, session ended")

​
Outbound messages

Earlier we printed the prompt to stdout. What if we want to send it to the client instead? Define a ModelMessage subclass: 
from dataclasses import dataclass
from reactor_runtime.interface import ModelMessage

@dataclass
class CurrentPrompt(ModelMessage):
    prompt: str
Now replace the print with self.send(): 
@event(name="get_current_prompt", description="Return the current prompt")
async def get_current_prompt(self):
    await self.send(CurrentPrompt(prompt=self.state.prompt))
The client receives this as JSON: {"type": "current_prompt", "data": {"prompt": "..."}}. The type on the wire is the snake_case version of the class name. You can also send messages from inside inference(). If you need to, make it async: 
@dataclass
class Progress(ModelMessage):
    step: int
    total: int

async def inference(self):
    step = 0
    while True:
        frame = self.pipe.forward(prompt=self.state.prompt, step=step)
        # ✅ Send a message to the client
        await self.send(Progress(step=step, total=100))
        step += 1
        yield MyOutput(main_video=frame)
Note that inference() is async def here instead of def because we need await to send messages. Both sync and async generators work; use whichever fits your model.

​
File uploads

When your model needs binary files from the client (reference images, style assets, etc.), declare an UploadedFile-typed field on your InputState. The runtime generates a set_<field> event and handles the upload automatically. UploadedFile has four fields: name, mime_type, size, and data (the raw bytes). You can use them to filter by type, decode images, or pass the content to your pipeline. 
from reactor_runtime.interface import InputState, InputField, UploadedFile

@dataclass
class MyState(InputState):
    prompt: str = InputField(default="a sunny meadow")
    reference_image: UploadedFile = InputField(
        default=None,
        description="Reference image for conditioning",
    )
    _img_embedding: Any = None
To run expensive post-processing when a file arrives, override the auto-generated event with a custom @event handler. Type the parameter as UploadedFile: 
@event(name="set_reference_image", description="Upload and encode reference image")
def set_reference_image(self, reference_image: UploadedFile):
    self.state.reference_image = reference_image
    pil_image = Image.open(io.BytesIO(reference_image.data)).convert("RGB")
    self.state._img_embedding = self.encoder.encode(pil_image)

​
Putting it together

A model that combines everything from this page: custom events, lifecycle hooks, and outbound messages. First, the types — state, events, and messages: 
@dataclass
class MyState(InputState):
    prompt: str = InputField(default="a sunny meadow")

# Sent back to the client when they ask for the current prompt
@dataclass
class CurrentPrompt(ModelMessage):
    prompt: str

# Sent every frame so the client can show a progress indicator
@dataclass
class Progress(ModelMessage):
    step: int
    total: int
Then the model class: 
class MyModel(ReactorPipeline):
    state: MyState

    @event(name="get_current_prompt", description="Return the current prompt")
    async def get_current_prompt(self):
        await self.send(CurrentPrompt(prompt=self.state.prompt))

    async def inference(self):
        step = 0
        while True:
            frame = self.pipe.forward(prompt=self.state.prompt, step=step)
            await self.send(Progress(step=step, total=100))
            step += 1
            yield MyOutput(main_video=frame)
The client can:
  • Send set_prompt (auto-generated from InputState).
  • Send get_current_prompt (custom event) and receive a current_prompt message back.
  • Receive progress messages each frame.

​
Next

Video Input

Read webcam frames from the client.

Bring Your Own Model

Convert an existing research pipeline to Reactor.