} ; }; All methods on FrameProcessor are async. The SDK calls them from an asyncio event loop; blocking calls in process\_video\_async will stall the pipeline. Run blocking model inference in a thread pool using asyncio.to\_thread. *** PyTrickle's processing surface is the `FrameProcessor` base class. Subclass it, implement the async methods below, and pass an instance to `StreamServer`. The SDK manages connection lifecycle, decoding, encoding, and transport. ## FrameProcessor ### Initialize Method ```python theme={"theme":{"light":"github-light","dark":"dark-plus"}} async def initialize(self) -> None ``` Called once before the first frame is delivered. Load AI models, allocate CUDA tensors, or warm up any resources that must be ready before processing begins. Raises are propagated and halt startup. *** ### Process Video Async Method ```python theme={"theme":{"light":"github-light","dark":"dark-plus"}} async def process_video_async(self, frame: VideoFrame) -> Optional[VideoFrame] ``` Called for each decoded video frame. Return a `VideoFrame` to include it in the output stream, or `None` to drop the frame. Blocking operations (model inference on CPU, disk I/O) stall the processing loop and increase output latency. Run them in a thread pool: ```python theme={"theme":{"light":"github-light","dark":"dark-plus"}} import asyncio async def process_video_async(self, frame: VideoFrame) -> Optional[VideoFrame]: tensor = frame.tensor.clone() result_tensor = await asyncio.to_thread(self._run_model, tensor) return frame.replace_tensor(result_tensor) def _run_model(self, tensor): # Blocking model inference here return self.model(tensor) ``` *** ### Process Audio Async Method ```python theme={"theme":{"light":"github-light","dark":"dark-plus"}} async def process_audio_async(self, frame: AudioFrame) -> Optional[List[AudioFrame]] ``` Called for each decoded audio frame. Return a list of `AudioFrame` objects (pass-through returns `[frame]`), or `None` to drop the frame. The SDK automatically detects and converts between mono, stereo, and multi-channel formats. *** ### Update Params Method ```python theme={"theme":{"light":"github-light","dark":"dark-plus"}} def update_params(self, params: dict) -> None ``` Called when the Livepeer Orchestrator forwards a parameter update from the Gateway. `params` is a plain dict; the schema is defined by your service. Updates are applied immediately to the running processor without interrupting the stream. Example: ```python theme={"theme":{"light":"github-light","dark":"dark-plus"}} def update_params(self, params: dict): if "prompt" in params: self.prompt = params["prompt"] if "strength" in params: self.strength = float(params["strength"]) ``` ## VideoFrame Represents a single decoded video frame. Attribute / Method Type Description `frame.tensor` `torch.Tensor` CPU tensor, shape \[H, W, C], float32, values in \[0, 1], RGB channel order `frame.pts` `int` Presentation timestamp in stream time base units `frame.width` `int` Frame width in pixels `frame.height` `int` Frame height in pixels `frame.replace_tensor(tensor)` `VideoFrame` Returns a new VideoFrame with the given tensor and the same metadata ## AudioFrame Represents a decoded audio buffer. Attribute Type Description `frame.tensor` `torch.Tensor` Audio samples as a float32 tensor, shape \[channels, samples] `frame.sample_rate` `int` Sample rate in Hz `frame.pts` `int` Presentation timestamp ## StreamServer ```python theme={"theme":{"light":"github-light","dark":"dark-plus"}} StreamServer( frame_processor: FrameProcessor, port: int = 8000, capability_name: str = "custom-processor", ) ``` Parameter Type Description `frame_processor` `FrameProcessor` The processor instance to call for each frame. Must be started before passing to StreamServer. `port` `int` HTTP port to listen on. Default: 8000. `capability_name` `str` Name advertised to the Orchestrator as the BYOC capability identifier. ### HTTP endpoints Endpoint Method Purpose `/api/stream/start` POST Start a processing session. Body: `subscribe_url`, `publish_url`, `gateway_request_id`, `params`. `/api/stream/stop` POST Stop the active session. Body: `gateway_request_id`. `/api/stream/update` POST Send parameter updates. Body: any JSON object; forwarded to `update_params`. `/api/stream/status` GET Returns current session status and frame counts. `/health` GET Returns `{"status": "ok"}`. Required for BYOC registration. ## Framerate and Throughput The `max_framerate` parameter in the `/api/stream/start` request body controls the maximum number of frames passed to `process_video_async` per second. Frames arriving faster than `max_framerate` are dropped before the processor sees them. This prevents backpressure from accumulating when model inference is slower than the input frame rate. Set `max_framerate` to the throughput your model can sustain. For StreamDiffusion workflows on an RTX 4090, this is typically 15-30 FPS depending on resolution. For lighter processing, match the input stream frame rate. The [PyTrickle quickstart](/v2/developers/build/ai-and-agents/realtime-ai/pytrickle/pytrickle-quickstart) walks through a complete FrameProcessor from scratch. The [PyTrickle reference](/v2/developers/resources/reference/pytrickle-reference) covers the full API surface. ## Related Pages Working example of a FrameProcessor running against a local test stream. How to register a trickle service with a Livepeer Orchestrator.