Skip to main content

​
Reactor Class

The core class for connecting to and interacting with Reactor models in Python.

​
Constructor

Reactor(
    model_name: str,
    api_key: str | None = None,
    coordinator_url: str = "https://api.reactor.inc",
    local: bool = False,
)

​
Parameters

ParameterTypeRequiredDescription
model_namestrYesThe name of the Reactor model to connect to
api_keystrNoYour Reactor API key. The SDK auto-fetches a JWT token using this key. Required unless local=True.
coordinator_urlstrNoThe coordinator URL (default: https://api.reactor.inc)
localboolNoConnect to localhost:8080 for local development (default: False)
Unlike the JavaScript SDK, you pass the API key directly to the constructor and the SDK handles JWT token exchange automatically during connect().

​
Example

from reactor_sdk import Reactor

# Production
reactor = Reactor(model_name="livecore", api_key="rk_your_key")

# Local development
reactor = Reactor(model_name="my-model", local=True)

​
Methods

​
connect()

Connects to the Reactor coordinator and waits for GPU assignment. If api_key was provided, automatically fetches a JWT token first. 
await reactor.connect() -> None
Raises: RuntimeError if authentication fails or already connected. Example: 
try:
    await reactor.connect()
    print("Connected successfully")
except Exception as error:
    print(f"Connection failed: {error}")

​
disconnect()

Disconnects from both the coordinator and GPU machine. 
await reactor.disconnect(recoverable: bool = False) -> None
Parameters:
  • recoverable — If True, keeps the session alive for potential reconnection. If False (default), terminates the session completely.
Example: 
# Full disconnect — terminates session
await reactor.disconnect()

# Recoverable disconnect — session can be resumed
await reactor.disconnect(recoverable=True)

​
send_command()

Sends a command to the model. Commands are the primary way to control model behavior. 
await reactor.send_command(command: str, data: Any) -> None
Parameters:
  • command — The command type string (e.g., "schedule_prompt", "start")
  • data — The command payload (typically a dictionary)
Example: 
await reactor.send_command("schedule_prompt", {
    "new_prompt": "A sunset over the ocean",
    "timestamp": 0,
})

​
publish_track()

Publishes a video track to the model for processing by video-to-video models. 
await reactor.publish_track(track: MediaStreamTrack) -> None
Parameters:
  • track — An aiortc.MediaStreamTrack object

​
unpublish_track()

Stops publishing the video track to the model. 
await reactor.unpublish_track() -> None

​
set_frame_callback()

Sets a callback to receive individual video frames as NumPy arrays. 
reactor.set_frame_callback(callback: FrameCallback | None) -> None
Parameters:
  • callback — A function that receives a NumPy array (H, W, 3) with dtype uint8 in RGB format. Pass None to stop receiving frames.
Example: 
def process_frame(frame):
    print(f"Frame shape: {frame.shape}")  # e.g. (720, 1280, 3)

reactor.set_frame_callback(process_frame)
You can also use the @reactor.on_frame decorator for the same effect. See Decorators.

​
get_remote_track()

Returns the remote video track from the GPU machine, if available. 
reactor.get_remote_track() -> MediaStreamTrack | None

​
on()

Registers an event handler. 
reactor.on(event: ReactorEvent, handler: Callable) -> None
Parameters:
  • event — The event type to listen for (see Events)
  • handler — Callback function
Example: 
def on_status(status):
    print(f"Status: {status}")

reactor.on("status_changed", on_status)

​
off()

Removes an event handler. 
reactor.off(event: ReactorEvent, handler: Callable) -> None
Example: 
reactor.off("status_changed", on_status)

​
get_status()

Returns the current connection status. 
reactor.get_status() -> ReactorStatus
Returns: ReactorStatus.DISCONNECTED, ReactorStatus.CONNECTING, ReactorStatus.WAITING, or ReactorStatus.READY Example: 
from reactor_sdk import ReactorStatus

status = reactor.get_status()
if status == ReactorStatus.READY:
    await reactor.send_command("start", {})

​
get_state()

Returns the complete current state including status and error information. 
reactor.get_state() -> ReactorState
Returns: ReactorState with status and last_error fields. Example: 
state = reactor.get_state()
print(f"Status: {state.status}")
if state.last_error:
    print(f"Last error: {state.last_error.message}")

​
get_session_id()

Returns the current session ID if connected. 
reactor.get_session_id() -> str | None

​
get_last_error()

Returns the most recent error that occurred. 
reactor.get_last_error() -> ReactorError | None

​
reconnect()

Reconnects to an existing session that may have been interrupted. 
await reactor.reconnect() -> None
Requirements: An active session_id must exist from a previous connection. Example: 
await reactor.disconnect(recoverable=True)

# Later, reconnect to the same session
await reactor.reconnect()

​
Context Manager

The Reactor class supports async with for automatic connection cleanup: 
async with Reactor(model_name="livecore", api_key=api_key) as reactor:
    await reactor.send_command("start", {})
    # Automatically disconnects when leaving the block

​
Events

The following event types can be used with on() and off():
EventHandler SignatureDescription
"status_changed"(status: ReactorStatus)Connection status changed
"session_id_changed"(session_id: str | None)Session ID changed
"new_message"(message: Any)Message received from the model
"stream_changed"(track: MediaStreamTrack)Video stream changed
"error"(error: ReactorError)Error occurred
"session_expiration_changed"(expiration: float | None)Session expiration time changed

​
Next Steps