Skip to main content

​
Types

​
ReactorStatus

The current connection status of the Reactor instance. This is a Python Enum. 
from reactor_sdk import ReactorStatus

class ReactorStatus(Enum):
    DISCONNECTED = "disconnected"  # Not connected
    CONNECTING = "connecting"      # Establishing coordinator connection
    WAITING = "waiting"            # In queue, waiting for GPU assignment
    READY = "ready"                # Connected to GPU, can send/receive
Example: 
status = reactor.get_status()
if status == ReactorStatus.READY:
    await reactor.send_command("start", {})

​
ReactorState

Complete state information including status and errors. This is a Python dataclass. 
@dataclass
class ReactorState:
    status: ReactorStatus
    last_error: ReactorError | None = None
Example: 
state = reactor.get_state()
print(f"Status: {state.status}")
if state.last_error:
    print(f"Error: {state.last_error.message}")

​
ReactorError

Error information when issues occur. This is a Python dataclass. 
@dataclass
class ReactorError:
    code: str                              # Error code identifier
    message: str                           # Human-readable error message
    timestamp: float                       # When the error occurred (Unix timestamp)
    recoverable: bool                      # Whether the error can be recovered from
    component: Literal["coordinator", "gpu"]  # Which component errored
    retry_after: float | None = None       # Suggested retry delay in seconds
ReactorError also supports string conversion: 
error = reactor.get_last_error()
if error:
    print(error)  # "[coordinator:CONNECTION_FAILED] Connection failed: ..."

​
Common Error Codes

CodeComponentRecoverableDescription
AUTHENTICATION_FAILEDcoordinatorNoInvalid API key or JWT token
COORDINATOR_CONNECTION_ERRORcoordinatorYesFailed to connect to coordinator
GPU_CONNECTION_FAILEDgpuYesFailed to connect to assigned GPU
GPU_CONNECTION_ERRORgpuYesGPU connection error
MESSAGE_SEND_FAILEDgpuYesFailed to send message to GPU

​
ReactorEvent

The event types that can be used with reactor.on() and reactor.off(). 
ReactorEvent = Literal[
    "status_changed",
    "session_id_changed",
    "new_message",
    "stream_changed",
    "error",
    "session_expiration_changed",
]
Python events use snake_case (e.g., "status_changed") while the JavaScript SDK uses camelCase (e.g., "statusChanged").

​
FrameCallback

Type alias for functions that receive video frames. 
FrameCallback = Callable[[NDArray[np.uint8]], None]
The callback receives a NumPy array with:
  • Shape: (H, W, 3) — height, width, 3 color channels
  • Dtype: uint8 — values 0-255
  • Color order: RGB
Example: 
from reactor_sdk import FrameCallback

def my_callback(frame):
    print(f"Shape: {frame.shape}, dtype: {frame.dtype}")

reactor.set_frame_callback(my_callback)

​
ConflictError

Exception raised when a connection conflict occurs (e.g., an existing connection was superseded by a newer request). 
class ConflictError(Exception):
    pass

​
fetch_jwt_token()

Fetches a JWT token from the coordinator using an API key. This is an async utility function. 
async def fetch_jwt_token(
    api_key: str,
    coordinator_url: str = "https://api.reactor.inc",
) -> str

​
Parameters

ParameterTypeDescription
api_keystrYour Reactor API key (e.g., rk_...)
coordinator_urlstrOptional. The coordinator URL (default: https://api.reactor.inc)

​
Returns

str — The JWT token.

​
Raises

RuntimeError if authentication fails (invalid API key, network error, etc.)

​
Example

from reactor_sdk import fetch_jwt_token

token = await fetch_jwt_token("rk_your_api_key_here")
In most cases, you don’t need to call this directly. Pass api_key to the Reactor constructor and the SDK handles token exchange automatically during connect().