Skip to main content
The Python SDK provides decorators as an idiomatic alternative to reactor.on(event, handler). Both approaches are equivalent. Use whichever you prefer.

​
@reactor.on_frame

Registers a callback to receive video frames as NumPy arrays.
Signature
@reactor.on_frame
def handler(frame: NDArray[np.uint8]) -> None: ...
frame is a NumPy array with shape (H, W, 3), dtype uint8, in RGB color order.
Example
@reactor.on_frame
def on_frame(frame):
    print(f"Frame: {frame.shape}")  # e.g. (768, 1280, 3)

    # PIL
    from PIL import Image
    Image.fromarray(frame).save("frame.png")

    # OpenCV (convert RGB to BGR first)
    import cv2
    cv2.imwrite("frame.png", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))

​
@reactor.on_message

Registers a handler for messages from the model.
Signature
@reactor.on_message
def handler(message: Any) -> None: ...
message is the payload sent by the model (typically a dictionary). Shape depends on the model.
Example
@reactor.on_message
def on_message(message):
    if message.get("type") == "state":
        print(f"Frame: {message['data']['current_frame']}")

​
@reactor.on_status

Registers a handler for connection status changes. Supports three usage patterns.
Signature
# All status changes
@reactor.on_status
def handler(status: ReactorStatus) -> None: ...

# One specific status
@reactor.on_status(ReactorStatus.READY)
def handler(status: ReactorStatus) -> None: ...

# Multiple statuses
@reactor.on_status([ReactorStatus.CONNECTING, ReactorStatus.WAITING])
def handler(status: ReactorStatus) -> None: ...
Example
from reactor_sdk import ReactorStatus

@reactor.on_status(ReactorStatus.READY)
async def on_ready(status):
    await reactor.send_command("start", {})

@reactor.on_status([ReactorStatus.CONNECTING, ReactorStatus.WAITING])
def on_pending(status):
    print(f"Connecting: {status}")

​
@reactor.on_error

Registers a handler for errors.
Signature
@reactor.on_error
def handler(error: ReactorError) -> None: ...
error is a ReactorError with fields code, message, timestamp, recoverable, component, and optional retry_after.
Example
@reactor.on_error
async def on_error(error: ReactorError):
    print(f"[{error.component}:{error.code}] {error.message}")
    if error.recoverable:
        await asyncio.sleep(error.retry_after or 3)
        await reactor.reconnect()

​
@reactor.on_track

Registers a handler for a named media track received from the model.
Signature
@reactor.on_track(name: str)
def handler(track: MediaStreamTrack) -> None: ...
​
name
str
required
The track name to listen for (e.g., "main_video").
Example
@reactor.on_track("main_video")
def on_main_video(track):
    print(f"Received track: {track.kind}")
For most use cases, @reactor.on_frame is simpler since it delivers decoded NumPy frames directly. Use @reactor.on_track when you need access to the raw MediaStreamTrack.