reactor-sdk 0.5.0

Python SDK for Reactor - Real-time AI video streaming

Reactor Python SDK

Python SDK for Reactor — real-time AI video streaming platform.

For full documentation, see the Python SDK Guide.

Installation

pip install reactor-sdk

Quick Start

import asyncio
from reactor_sdk import Reactor, ReactorStatus, video

async def main():
    reactor = Reactor(
        model_name="my-model",
        api_key="REACTOR_API_KEY",
        receive=[video("main_video")],
    )

    @reactor.on_frame
    def handle_frame(frame):
        print(f"Received frame: {frame.shape}")

    @reactor.on_status(ReactorStatus.READY)
    def handle_ready(status):
        print("Connected and ready!")

    await reactor.connect()
    await reactor.send_command("setParameter", {"value": 0.5})

    try:
        while reactor.get_status() == ReactorStatus.READY:
            await asyncio.sleep(0.1)
    finally:
        await reactor.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

Features

• WebRTC video streaming via aiortc
• Multi-track support — named video and audio tracks with video() / audio() helpers
• Event-driven API matching the JavaScript SDK
• Frame callbacks for single-frame access (numpy arrays)
• Video input support for sending video to models
• Local development mode for testing without auth
• Full type hints for IDE support
• Async context manager — async with Reactor(...) as r:

API Overview

Constructor

from reactor_sdk import Reactor, video, audio

reactor = Reactor(
    model_name="my-model",
    api_key="REACTOR_API_KEY",
    receive=[video("main_video")],
    send=[video("webcam")],
)

For local development, pass local=True instead of api_key.

Key Methods

• await reactor.connect() — Connect to the model (auto-fetches JWT from API key)
• await reactor.disconnect(recoverable=False) — Disconnect
• await reactor.reconnect() — Reconnect to an existing session
• await reactor.send_command(command, data) — Send a command
• await reactor.publish_track(name, track) — Publish a named media track
• await reactor.unpublish_track(name) — Unpublish a named track
• reactor.get_status() — Current ReactorStatus
• reactor.get_state() — ReactorState(status, last_error)
• reactor.get_session_id() — Session ID
• reactor.get_remote_tracks() — Dict of received tracks by name
• reactor.set_frame_callback(callback) — Set frame callback

Decorators

@reactor.on_frame
def handle_frame(frame): ...

@reactor.on_track("main_video")
def handle_video(track): ...

@reactor.on_message
def handle_message(message): ...

@reactor.on_status(ReactorStatus.READY)
def handle_ready(status): ...

@reactor.on_error
def handle_error(error): ...

See the full documentation for the complete API reference, event list, and usage patterns.

Examples

See the examples/ directory for complete examples:

• pygame_app/ — Pygame application with dynamic UI controls
• rtmp_app/ — Stream Reactor video to RTMP servers (Twitch, YouTube, etc.)

License

MIT License — Copyright (c) 2026 Reactor Technologies, Inc.