> ## Documentation Index > Fetch the complete documentation index at: https://docs.reactor.inc/llms.txt > Use this file to discover all available pages before exploring further. > ## Agent Instructions > To build and serve your own model, start at /deploy/development/quickstart and /deploy/development/overview. Deploying is the default path: reactor init scaffolds a workspace, reactor auth login authenticates, and reactor model deploy registers the model, publishes the release with the weights/ folder, and activates it on Reactor's GPUs, in one command from that workspace. Docker must be running, because the publish step builds the image locally. Bump model.version in reactor.yaml before redeploying a change, because a release that already has an image is reactivated as it is. Deployment access is granted per account, so contact team@reactor.inc if a deploy is refused. Every key in reactor.yaml is documented at /deploy/platform/reactor-yaml. Model code imports reactor_runtime; Python client code imports reactor_sdk. The runtime overview explains the model interface. Running the model on your own machine with reactor run is optional and needs a GPU you attach with --gpus; /deploy/development/local-testing covers that loop and pairs a complete brightness model with a Python client test in a separate brightness-test workspace. > Reactor hosts multiple models, each with its own connect slug (modelName) and command/event schema. The catalog of every model — slug, typed SDK package, and links to its schema — is at /model-api-reference/overview. Some models expose one slug per experience (e.g. HappyOyster); always take the slug from the model's own pages, never guess it. > Fastest path to a working app: `npx create-reactor-app my-app --model=` scaffolds a complete app with secure auth wired up. Typed TypeScript SDKs are published as @reactor-models/; Python uses the base reactor-sdk package. > Auth: exchange an API key (rk_...) for a JWT via POST https://api.reactor.inc/tokens from your server. Never put the API key in client-side code. > Append .md to any docs URL for clean Markdown. Search these docs via the MCP server at https://docs.reactor.inc/mcp. # Video & Audio Tracks > Declare the video and audio your model sends and receives, shape the frames, pace playout, and keep audio in sync. Real-time media is streamed as it is produced, frame by frame, in both directions. It travels on named streams called **tracks**. Your model declares the tracks it sends on an `Output` class and the tracks it receives on a `MediaInput` class. The runtime handles the encoding and the transport at both ends; you deal in NumPy arrays. The examples on this page are a video-to-video model in the shape of SANA-Streaming: the client publishes its camera, the model restyles it according to a prompt, and the restyled video comes back. A client panel on the left publishes webcam and subscribes to main_video. A ReactorApp panel on the right declares webcam on MediaInput and main_video on Output. Lines join each name to its match on the other side. ## Output tracks Subclass `Output` with one field per track. The field name is the track name a client subscribes to, and the annotation says what it carries: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} from reactor_runtime import Audio, Output, Video class SanaStreamingOutput(Output): main_video: Video ``` A step returns an instance of this class, and the runtime puts each field on its track. Every declared track has to be present on every `Output`. If a step has no audio for an audio track, send an empty array rather than leaving the field out. A video frame is an RGB array of shape `(height, width, 3)` with dtype `uint8`. A step that produces several frames returns them as one `(frames, height, width, 3)` array; the runtime splits the batch and paces it. Audio is `int16` of shape `(1, samples)` at 48 kHz, mono; the [Audio](#audio) section below covers rates and sync. ### Several tracks A model can send more than one stream, and each client picks the ones it wants. Take a two-player world model where each player gets their own point of view. The `Output` declares one track per player, and every step fills both: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} class ArenaOutput(Output): # ✅ One track per point of view player_one: Video player_two: Video # ✅ Shared by both clients ambience: Audio class Arena(ReactorApp): def generate(self, input: ArenaState) -> ArenaOutput: views = self.world.step(input.moves) # both cameras return ArenaOutput( player_one=views[0], player_two=views[1], ambience=self.world.sound(), ) ``` The two clients connect to the same session and see the same world from different places. Player one's app subscribes to `player_one`, player two's to `player_two`, and both take `ambience`. Tracks returned in one `Output` are kept in step with each other, so the two views and the sound line up. ## Frame rate `generate()` hands frames over one chunk at a time, and a chunk starts playing the moment it lands. While it plays, the runtime is already asking for the next one. Generation runs one chunk ahead of what the client is playing, and the chunks play in the order they were made: A time axis from 0 to 3 seconds. On top, generate() returns chunk 1, chunk 2, and chunk 3 back to back. Below, each chunk's 24 frames start playing the moment that chunk lands, so playout runs exactly one chunk behind generation. To play at a fixed rate instead, declare `fps` on the class: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} class SanaStreaming(ReactorApp): fps = 24 ``` Now every chunk plays at 24 frames per second regardless of how long it took. A model that runs faster than that is held back: returning the next `Output` waits while frames already handed over are still playing. That is what stops a fast model from running ahead of the client, and it is why you never need to sleep in `generate()`. Two controls sit on `self.output`, and both reach every connected client: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} self.output.fps = 30 # re-pace what is queued, and what follows self.output.flush() # drop what is queued and cut playout ``` Call `flush()` when the scene changes, so nothing from the old scene plays after the cut. A recording keeps running across the cut. ## Frame metadata A frame can carry a small tag. Wrap the array in `TrackPayload` and pass `metadata`: a mapping travels as JSON, bytes travel as they are. On a batch, pass one value for every frame or a list with one value per frame. ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} tag = {"chunk": result.chunk_index} tags = [tag] * len(result.frames) video = TrackPayload(result.frames, metadata=tags) return SanaStreamingOutput(main_video=video) ``` The client reads the tag next to the frame it came with. A video-to-video model can copy the tag it received on an input frame onto the frame it produced from it, which lets the client pair the two without a side channel. ## Input tracks Subclass `MediaInput` with one field per track you expect the client to publish, and annotate it on your model. The runtime finds it by its type, so the attribute name is up to you; `media` is the convention: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} from reactor_runtime import MediaInput, ReactorApp, Video class SanaStreamingMedia(MediaInput): webcam: Video class SanaStreaming(ReactorApp): media: SanaStreamingMedia ``` The runtime negotiates each declared track with the client and buffers the frames that arrive, reachable as `self.media.webcam`. Each track holds the last 128 frames and drops the oldest when full, so a model that falls behind does not grow memory without bound. `Input` is the old name for `MediaInput`. It still imports, with a deprecation warning. ### Several tracks A client can publish more than one stream too. A robot policy driven from three cameras declares one track per camera and reads them all in `process_input()`: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} class RobotMedia(MediaInput): # ✅ One track per camera; the robot publishes all three head: Video left_wrist: Video right_wrist: Video class Policy(ReactorApp): media: RobotMedia async def process_input(self) -> PolicyInput: # ✅ The newest frame from each camera head = self.media.head.try_read(1) left = self.media.left_wrist.try_read(1) right = self.media.right_wrist.try_read(1) # ✅ Wait until every camera has delivered at least once if head is None or left is None or right is None: raise ApplicationError("waiting for the cameras") return PolicyInput( head=head[0].data, left_wrist=left[0].data, right_wrist=right[0].data, task=self.state.task, ) ``` Each track is its own buffer, so the cameras do not have to arrive in lockstep. Reading the newest frame from each gives the policy the most recent view from every angle, whatever order the frames came in. ## Reading frames Read tracks in `process_input()` with `try_read(n)`. It returns the `n` frames you asked for, or `None` if fewer than that have arrived, and it never waits. Too few frames is a reason to skip the step, so the model runs only with a full input: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} async def process_input(self) -> SanaStreamingInput: # ✅ Take the 4 newest camera frames frames = self.media.webcam.try_read(4) # ✅ Not enough yet: skip this step and try again shortly if frames is None: raise ApplicationError("waiting for 4 webcam frames") # ✅ Stack them into one array for the model batch = np.stack([f.data for f in frames]) prompt = self.state.prompt return SanaStreamingInput(prompt=prompt, frames=batch) ``` Each frame arrives as an `InputFrame`, whose `data` holds the array: `(height, width, 3)` `uint8` for video, `(1, samples)` `int16` for audio. A frame also carries the bytes the client tagged it with and, when the sender supplied one, a presentation timestamp. Reads take a `mode`. `ReadMode.LATEST`, the default, returns the newest frames and drops the backlog, which is what video wants: if the model fell behind, it should catch up to now rather than restyle stale frames. `ReadMode.FIFO` returns the oldest in order and leaves the rest queued, which is what audio wants, since dropped samples are audible gaps. ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} frames = self.media.webcam.try_read(4) # newest 4 mic = self.media.mic chunks = mic.try_read(1, mode=ReadMode.FIFO) # oldest 1 ``` `await read(n)` waits until `n` frames are available instead of returning `None`. Inside `process_input()` a wait holds the step and no command runs until a frame arrives, so prefer `try_read()` there and keep `read()` for a hand-written `run()`. When the last client leaves or the session ends, every input buffer is reset, so the next client starts from empty tracks. Every field on a frame, both read calls with their arguments, and the rest of the buffer surface. ## Audio Audio tracks work like video tracks. Declare them on the `Output`, return samples alongside the frames, and the runtime encodes and transports them: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} class MyOutput(Output): main_video: Video main_audio: Audio def generate(self, input: MyState) -> MyOutput: video = self.pipe.forward(prompt=input.prompt) audio = self.pipe.generate_audio() return MyOutput( main_video=video, # (H, W, 3) uint8 main_audio=audio, # (1, N) int16 ) ``` Samples are `int16`, shaped `(1, samples)` for mono, at 48 kHz. The runtime encodes them as Opus, whose native rate is 48 kHz, so audio already at that rate needs no resampling. When a step has no audio, send an empty `(1, 0)` array rather than leaving the field out. ### Keeping audio and video together When one `Output` carries both tracks, the runtime keeps them aligned: if the video is a batch, the audio is spread across those frames. So each `Output` should carry the audio that belongs to its video. At 30 fps one frame is `48000 / 30 = 1600` samples, so a batch of 3 frames carries about 4,800 samples. A frame's worth of video with a second's worth of audio drifts, however good the transport. ### Other sample rates Live audio always goes out at 48 kHz, and the runtime never resamples for you. If your model produces audio at another rate, resample to 48 kHz before it goes on the `Output`, or playback runs at the wrong pitch. If you would rather emit at the native rate, subclass `Audio` and declare it: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} class Audio16k(Audio): sample_rate = 16_000 class MyOutput(Output): main_video: Video main_audio: Audio16k ``` This does not change what goes out live, which stays 48 kHz. It matters for [recording](/deploy/development/recording): the recorder encodes the audio at the rate you declare, which is what keeps a recorded clip's pitch and duration right. ### Two producers A model whose audio and video come from separate producers has to pair them before emitting. That is a loop which blocks on input, so it keeps a hand-written `run()` (see [Controlling the loop](/deploy/development/reactor-app/step-loop#controlling-the-loop)). Buffer whichever stream runs ahead, and emit only when both have material for the same span: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} async def run(self) -> None: while True: await self.connected.wait() audio_pending: dict[int, np.ndarray] = {} video_pending: dict[int, np.ndarray] = {} next_block = 0 while self.connected.is_set(): kind, index, data = await self._next_block() if kind == "audio": audio_pending[index] = data else: video_pending[index] = data # ✅ Emit only once both tracks have the same block ready = lambda i: i in audio_pending and i in video_pending while ready(next_block): await self.emit( MyOutput( main_video=video_pending.pop(next_block), main_audio=audio_pending.pop(next_block), ) ) next_block += 1 ``` Pairing by block index rather than by arrival order is what keeps the two tracks locked when one producer is briefly slower than the other. ### Reading audio from the client Inbound audio arrives as `(1, samples)` `int16` at 48 kHz and is read like video, but in `FIFO` order so no samples are dropped: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} chunks = self.media.mic.try_read(1, mode=ReadMode.FIFO) ``` A common pattern is to drain everything queued each step: ```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} audio = [] while (chunks := self.media.mic.try_read(1, mode=ReadMode.FIFO)): audio.extend(chunks) ``` ## A video-to-video model Everything above, in one class. The client publishes `webcam` and subscribes to `main_video`. Each step takes the four newest camera frames and the current prompt, restyles them, and sends them back tagged with the step they came from. ```python sana_streaming.py theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}} class SanaStreamingState(InputState): ... # prompt and strength, as on Managing State # ✅ What the client sends: one video track named "webcam" class SanaStreamingMedia(MediaInput): webcam: Video # ✅ What the client receives: one video track named "main_video" class SanaStreamingOutput(Output): main_video: Video # ✅ Optional typing: what process_input() hands to generate() @dataclass(frozen=True) class SanaStreamingInput: prompt: str frames: np.ndarray # (4, H, W, 3) uint8 class SanaStreaming(ReactorApp): state: SanaStreamingState media: SanaStreamingMedia def load(self, config_path: Path | None) -> None: weights = get_weights_path() self.pipe = SanaStreamingPipeline.from_pretrained(weights) self.chunk = 0 async def process_input(self) -> SanaStreamingInput: # ✅ Take the 4 newest camera frames frames = self.media.webcam.try_read(4) # ✅ Not enough yet: skip this step and try again shortly if frames is None: raise ApplicationError("waiting for 4 webcam frames") # ✅ Stack them into one array for the model batch = np.stack([f.data for f in frames]) prompt = self.state.prompt return SanaStreamingInput(prompt=prompt, frames=batch) def generate(self, input: SanaStreamingInput) -> np.ndarray: return self.pipe.restyle( input.frames, prompt=input.prompt, strength=self.state.strength, ) async def process_output( self, outcome: StepOutcome ) -> SanaStreamingOutput | None: if outcome.error is not None: raise outcome.error # ✅ Tag every frame with the step it came from self.chunk += 1 tag = {"chunk": self.chunk} video = TrackPayload(outcome.result, metadata=[tag] * 4) return SanaStreamingOutput(main_video=video) ``` Before the first camera frame lands, every step is skipped and `generate()` never runs. Once frames flow, the model keeps up with the camera by design: `LATEST` reads always start from the newest frames, so a slow step costs a few dropped camera frames rather than growing lag. ## Next Record sessions and let clients capture clips. How process\_input() and process\_output() fit around generate().