Reactor

Reactor is the base class for every model on the platform. It speaks raw JSON over the wire: open a session, send commands by name, and receive generic message events. Typed SDKs like HeliosModel and other model-specific packages extend Reactor:

Inheritance

class HeliosModel(Reactor): ...

The typed SDKs add named methods that mirror the model’s published commands (set_prompt(), send_image(), …). Under the hood they call send_command() on this base class. Same wire protocol, just typed. You can always drop down to Reactor directly to send any command as raw JSON; see the SDK Reference overview for when each path makes sense. For the list of commands and events a specific model accepts, see the Model API Reference.

Constructor

Signature

Reactor(
    model_name: str,
    api_key: str | None = None,
    *,
    jwt: str | None = None,
    api_url: str = "https://api.reactor.inc",
    local: bool = False,
    auto_resume_tracks: bool = True,
    max_session_duration_seconds: int | None = None,
)

Argument	Type	Description
`model_name`	`str`	Required. The name of the model to connect to. Required — the constructor raises `TypeError` without it.
`api_key`	`str`	Your Reactor API key. The SDK automatically exchanges this for a token during `connect()`. Use this or `jwt`, not both — `jwt` wins if both are given.
`jwt`	`str`	A token to authenticate with directly, skipping the API key exchange.
`api_url`	`str`	The API URL. Ignored if `local=True`. Default: `https://api.reactor.inc`.
`local`	`bool`	If `True`, connects to a local runtime at `http://localhost:8080` and relaxes TLS verification. No API key required. Default: `False`.
`auto_resume_tracks`	`bool`	If `True`, the client resumes every `recvonly` track on connect and on reconnect. Pass `False` to start with them paused, then `Track.resume()` the ones you want. A client keeps this choice for its whole life. Default: `True`.
`max_session_duration_seconds`	`int`	Applies to the token this client mints from `api_key`. Every session that token creates ends after this many seconds (1 to 86 400), whatever the token’s own expiry says. Has no effect when you pass `jwt`: that token was already minted. Same constraint as `fetch_jwt()`’s argument of the same name.

Example

from reactor_sdk import Reactor

reactor = Reactor(model_name="your-model-name", api_key="rk_...")

model_name and api_key may also be given positionally — Reactor("your-model-name", "rk_...") — matching the old py-sdk’s own constructor order exactly, so a call site ported from there needs no rewrite. api_url is keyword-only, so it can never be confused with either.

The Reactor class also supports async with for automatic cleanup:

Context manager

from reactor_sdk import Reactor

async with Reactor(model_name="your-model-name", api_key="rk_...") as reactor:
    await reactor.connect()
    await reactor.send_command("start", {})

No audio device is ever opened. A sendonly audio track carries only the PCM you push into it with Track.push_frame() — nothing is captured from your microphone. A model’s audio arrives on its Track, via on_frame, for you to play through whatever you like — nothing is played through your speakers on your behalf.

Methods

`connect()`

Establishes a connection to the Reactor Platform, waits for GPU assignment, and opens a WebRTC connection to the model. If api_key was provided, fetches a token first.

Signature

await reactor.connect(*, session_id: str | None = None, connection_id: int | None = None) -> None

Argument	Type	Description
`session_id`	`str`	Attach to an existing session instead of creating a new one — the session this reactor adopts does not get terminated on `disconnect()`.
`connection_id`	`int`	Adopt a WebRTC connection slot a backend already registered for this session, instead of registering a new one. The connection-level analogue of `session_id` — most callers building a single connection per session leave this unset. See Multiple connections per session.

`disconnect()`

Disconnects and ends the session on the server. Not recoverable — there is no parameter to keep the session alive instead. To disconnect and later resume the same session, call reconnect() directly rather than disconnect() followed by connect().

Signature

await reactor.disconnect() -> None

A no-op if not connected. Only actually ends the session when this client created it — adopting one via session_id leaves it running for its owner.

`reconnect()`

Reconnects using the same session — after a transient failure, or called from ready to deliberately cycle the connection. Tears down the live connection itself first if there is one, without ending the session server-side, so there is no need to call disconnect() beforehand (doing so would end the session this is about to reuse).

Signature

await reactor.reconnect() -> None

Raises InvalidStateError if there is no session to reconnect to — nothing has connected yet, or a previous disconnect() already ended it.

recvonly tracks resume automatically, unless you created the client with auto_resume_tracks=False. sendonly tracks do not — a track published before the reconnect is not published after it, so publish again for anything you were sending:

await reactor.reconnect()
await camera.publish()

Track.published is False until that happens, and push_frame() raises rather than sending into a slot with nothing behind it.

`close()`

Synchronously tears down the connection. Prefer disconnect() in async code; close() exists for cleanup paths (like __del__) that cannot await.

Signature

reactor.close() -> None

An operation still in flight does not stay pending: close() settles it with AbortedError. A send_command() whose reply had not arrived yet raises it in the coroutine that awaits it.

`send_command()`

Sends a command to the model and waits for its correlated reply.

Signature

await reactor.send_command(command: str, data: Any) -> dict[str, Any] | None

Argument	Type	Description
`command`	`str`	Required. The command name. Must match a command defined on the model.
`data`	`dict`	Required. The command payload. Shape depends on the command.

The SDK carries a payload over 256 KiB, up to 64 MiB, when the runtime accepts data-channel chunking. A runtime that does not accept it keeps the 256 KiB limit, and a larger payload raises MessageTooLargeError. Returns {"type": ..., "data": ...} — the model’s typed reply to this specific command. Returns None if the handler ran and acknowledged the command but returned no message, as an auto-generated set_<field> setter does.

Example

reply = await reactor.send_command("set_prompt", {"prompt": "a mountain landscape"})

To fire a command without waiting on the reply, schedule the call instead of awaiting it directly: asyncio.create_task(reactor.send_command(...)) — keep a reference to the task so it isn’t garbage-collected before it completes.

You can also pass FileRef values from upload_file() as parameters. A FileRef nested inside a list or a dict in data works too, for a parameter that takes several files. The SDK sends each one as an upload reference. See File Uploads for details.

`upload_file()`

Uploads a file and returns a FileRef that can be passed into send_command().

Signature

await reactor.upload_file(
    file: BinaryIO | bytes | str | os.PathLike,
    *,
    name: str | None = None,
    mime_type: str | None = None,
) -> FileRef

Argument	Type	Description
`file`	`BinaryIO \| bytes \| str \| PathLike`	Required. The file to upload. Accepts a file path, raw bytes, or a file-like object opened in binary mode.
`name`	`str`	Custom filename. Inferred from the file path or file object when not given. Defaults to `"upload"` for raw bytes.
`mime_type`	`str`	MIME type. Guessed from the filename when not given. Defaults to `"application/octet-stream"`.

Example

ref = await reactor.upload_file("photo.jpg")
await reactor.send_command("set_image", {"image": ref})

Only callable once the connection status is "ready" — raises InvalidStateError otherwise, the same guard request_clip() and Track.pause() use.

`publish_track()`

Activates a named sendonly track slot, so frames pushed with Track.push_frame() start reaching the model.

Signature

await reactor.publish_track(name: str) -> Track

Argument	Type	Description
`name`	`str`	Required. Track name. Must match a `sendonly` track name the model declares.

Returns the slot as a Track, so frames can go straight into it:

Example

camera = await reactor.publish_track("camera")
camera.push_frame(frame)

A failed publish_track() does not clean up the session by itself. If you call it outside a try/finally (or before entering the block that calls disconnect()), a failure here can leave the session running with nothing attached to it — the next connect() attempt then fails with ConflictError against the orphaned session. Wrap the whole connected lifetime, publish included, in one try/finally: await reactor.disconnect().

`unpublish_track()`

Deactivates a sendonly track. Synchronous, unlike the other track methods — it never touches the network, only a local status check and a fire-and-forget notification.

Signature

reactor.unpublish_track(name: str) -> None

Argument	Type	Description
`name`	`str`	Required. The name of the track to stop publishing.

A failure here is logged, not raised. Unpublish is commonly the last call in a finally block, and raising there would replace whatever exception was already propagating instead of adding to it. Check the logs (reactor_sdk at WARNING) if a track seems to have stayed published.

`set_bitrate()`

Bounds what the whole connection may allocate, in bits per second. Needs a connected client.

Signature

await reactor.set_bitrate(
    *, min_bps: int | None = None, start_bps: int | None = None, max_bps: int | None = None
) -> None

Argument	Type	Description
`min_bps`	`int \| None`	Minimum connection bitrate in bits per second. Default: `None`.
`start_bps`	`int \| None`	Initial encoder target in bits per second; has no effect on an already-live connection. Default: `None`.
`max_bps`	`int \| None`	Maximum connection bitrate in bits per second. Default: `None`.

None leaves a bound at the WebRTC default. The SDK reapplies the bounds after a reconnect. This ceiling and Track.set_bitrate()’s apply together, so the lower of the two wins. The 2.5 Mbps default that caps a video encoder is the track-level one, so raise that as well.

Example

await reactor.set_bitrate(max_bps=12_000_000)
await reactor.track("camera").set_bitrate(max_bps=8_000_000)

start_bps names the initial encoder target. It has no effect from here: this call needs a live connection, and the ramp that start_bps skips is over by then.

`track()`

The named track, as a Track — the only way to push frames into a track, receive its frames, or pause/resume it. Its publish()/push_frame()/on_frame()/ pause()/resume() refuse whatever the track’s direction doesn’t allow instead of failing silently.

Signature

reactor.track(name: str) -> Track

Argument	Type	Description
`name`	`str`	Required. The track name.

Raises ValueError for a name the session doesn’t declare, once it has declared any tracks — before that (early in connect(), before the model’s capabilities arrive) any name is accepted, so handlers can be registered ahead of time. The same object comes back every time for a given name, including across a reconnect.

Example

camera = reactor.track("camera")
await camera.publish()
camera.push_frame(frame)

Push a frame, receive one, or pause/resume a track through this Track object. publish_track() / unpublish_track() also stay on Reactor directly — publish_track() hands back this same Track.

See the full Track reference for every method and property.

`tracks`

Every track the session declares, in declaration order, as a TrackList — a list[Track] with with_kind()/with_direction()/one() filters chained on top, for finding a track without hardcoding its name.

Signature

reactor.tracks -> TrackList

Empty until the model’s capabilities arrive, shortly after connect().

Example

output = reactor.tracks.with_direction("recvonly").with_kind("video").one()

@output.on_frame
def render(frame): ...

Media is delivered per track: register on_frame() for decoded frames, or on_raw_frame() for the same bytes with no NumPy conversion — found by name via track() or by filtering here. on("frame", ...) / on("audio", ...) raise ValueError at registration, since a single handler fed every recvonly track of a kind at once couldn’t tell them apart.

`paused_tracks`

The names of the recvonly tracks currently paused.

Signature

reactor.paused_tracks -> frozenset[str]

Recvonly tracks are resumed automatically once connected, so this is empty on a healthy session until something is paused with Track.pause(). A client created with auto_resume_tracks=False sees its recvonly tracks here until it resumes them.

`request_schema()`

Requests the model’s command schema, as an OpenAPI document — the same schema published on the Model API Reference pages, fetched at runtime.

Signature

await reactor.request_schema() -> dict[str, Any]

`request_clip()`

Requests a clip covering the last duration_seconds of the session.

Signature

await reactor.request_clip(duration_seconds: float) -> Clip

Argument	Type	Description
`duration_seconds`	`float`	Clip length in seconds, counting back from the current point in the session.

Returns a Clip. See Recordings for the full flow (the clip streams as HLS segments; this call returns the manifest, not the video itself).

`request_recording()`

Requests a clip covering the entire session up to now.

Signature

await reactor.request_recording() -> Clip

`download_clip()`

request_clip() and download_clip() in one call — for when the file is the only thing you want.

Signature

await reactor.download_clip(
    duration_seconds: float,
    path: str | os.PathLike | None = None,
    *,
    on_progress: Callable[[int, int], None] | None = None,
) -> bytes | None

Argument	Type	Description
`duration_seconds`	`float`	Clip length in seconds, counting back from the current point in the session.
`path`	`str \| os.PathLike \| None`	Destination file. Defaults to `None`, returning the assembled bytes instead.
`on_progress`	`Callable[[int, int], None] \| None`	Optional callback receiving `(done, total)` segments. Default: `None`.

Given path, streams straight to it and returns None. Without one, returns the assembled bytes instead. See download_clip() for the full contract — same function, request_clip() called for you first.

Example

data = await reactor.download_clip(10)
await reactor.download_clip(10, "clip.ts")

Reach for request_clip() directly instead if you also want the Clip’s session_id, its start_marker / end_marker, or predicted_ready_at_ms — or want to decide whether to download it at all based on what it says.

`download_recording()`

request_recording() and download_clip() in one call.

Signature

await reactor.download_recording(
    path: str | os.PathLike | None = None,
    *,
    on_progress: Callable[[int, int], None] | None = None,
) -> bytes | None

Argument	Type	Description
`path`	`str \| os.PathLike \| None`	Destination file. Defaults to `None`, returning the assembled bytes instead.
`on_progress`	`Callable[[int, int], None] \| None`	Optional callback receiving `(done, total)` segments. Default: `None`.

Example

await reactor.download_recording("full-session.ts")

Pass path: a full-session recording has no upper bound on length, and only the streamed-to-disk form (returns None) avoids holding the whole thing in memory. Omitting path returns the assembled bytes, which does mean holding the whole recording in memory to do it.

`on()`

Registers a handler for an event name.

Signature

reactor.on(event: str, handler: Callable) -> None

Argument	Type	Description
`event`	`str`	Event name from the Events table.
`handler`	`Callable`	Callback to register.

See Events below for the full list of event names and their payloads.

"frame" and "audio" are not among them — media is delivered per track, not client-wide. Registering either raises ValueError at registration time, naming the Track method to use instead (on_frame() / on_raw_frame()).

`off()`

Removes a handler previously registered with on().

Signature

reactor.off(event: str, handler: Callable) -> None

Argument	Type	Description
`event`	`str`	Event name from the Events table.
`handler`	`Callable`	The same callback previously registered for this event.

`status` / `get_status()`

The current connection status. get_status() is an older method form of the same thing — both exist, use whichever reads better.

Signature

reactor.status -> ReactorStatus
reactor.get_status() -> ReactorStatus

Returns ReactorStatus.DISCONNECTED, CONNECTING, WAITING, or READY. ReactorStatus is a str subclass, so it compares equal to the plain string too (reactor.status == "ready").

`session_id` / `get_session_id()`

The current session ID, or None if not connected. get_session_id() is the older method form.

Signature

reactor.session_id -> str | None
reactor.get_session_id() -> str | None

`get_stats()`

A snapshot of the live connection: RTT, jitter, packet loss, bitrates, the transport ICE selected, and the WebRTC engine’s own per-stream counters.

Signature

await reactor.get_stats() -> ConnectionStats

Returns a ConnectionStats. Raises InvalidStateError unless the session is ready.

Example

stats = await reactor.get_stats()
print(stats.rtt_ms, stats.incoming_bitrate_bps, stats.candidate_type)

Events

Event	Payload	Description
`"status_changed"`	`str` (a `ReactorStatus` value)	Connection status changed
`"session_id_changed"`	`str \| None`	Session ID changed
`"message"`	`Any` (parsed JSON)	Application message from the model
`"runtime_message"`	`Any` (parsed JSON)	Internal platform message
`"track_received"`	`(name: str, mid: str \| None)`	Named `recvonly` track became available
`"error"`	`ReactorError`	Error occurred
`"capabilities_received"`	`Any` (parsed JSON)	Model capabilities received after session creation

on()/off() take the event name as a plain str — there is no separate ReactorEvent type to import.

Media is delivered per track. See Track.on_frame() / Track.on_raw_frame().

Python events use snake_case (e.g., "status_changed") while the JavaScript SDK uses camelCase (e.g., "statusChanged").

Get Started

Concepts

SDK Reference

Resources

Constructor

Methods

`connect()`

`disconnect()`

`reconnect()`

`close()`

`send_command()`

`upload_file()`

`publish_track()`

`unpublish_track()`

`set_bitrate()`

`track()`

`tracks`

`paused_tracks`

`request_schema()`

`request_clip()`

`request_recording()`

`download_clip()`

`download_recording()`

`on()`

`off()`

`status` / `get_status()`

`session_id` / `get_session_id()`

`get_stats()`

Events

​Constructor

​Methods

​connect()

​disconnect()

​reconnect()

​close()

​send_command()

​upload_file()

​publish_track()

​unpublish_track()

​set_bitrate()

​track()

​tracks

​paused_tracks

​request_schema()

​request_clip()

​request_recording()

​download_clip()

​download_recording()

​on()

​off()

​status / get_status()

​session_id / get_session_id()

​get_stats()

​Events

Constructor

Methods

`connect()`

`disconnect()`

`reconnect()`

`close()`

`send_command()`

`upload_file()`

`publish_track()`

`unpublish_track()`

`set_bitrate()`

`track()`

`tracks`

`paused_tracks`

`request_schema()`

`request_clip()`

`request_recording()`

`download_clip()`

`download_recording()`

`on()`

`off()`

`status` / `get_status()`

`session_id` / `get_session_id()`

`get_stats()`

Events