BYOC Overview

Livepeer Gateway
  ↓ trickle protocol
PyTrickle StreamServer (inside your container)
  ↓ VideoFrame / AudioFrame tensors
Your FrameProcessor (your model logic here)
  ↓ processed tensors
PyTrickle StreamServer
  ↓ trickle protocol
Livepeer Gateway

from pytrickle import FrameProcessor, StreamServer
from pytrickle.frames import VideoFrame, AudioFrame
from typing import Optional, List
import torch


class MyAIProcessor(FrameProcessor):
    """Custom AI video processor."""

    async def initialize(self):
        """Load your model. Called once on container start."""
        self.model = load_my_model()

    async def process_video_async(self, frame: VideoFrame) -> Optional[VideoFrame]:
        """Process a single video frame. Called once per frame."""
        tensor = frame.tensor  # PyTorch tensor: (H, W, C) or (C, H, W)
        with torch.no_grad():
            processed = self.model(tensor)
        return frame.replace_tensor(processed)

    async def process_audio_async(self, frame: AudioFrame) -> Optional[List[AudioFrame]]:
        """Process audio. Return None to drop; return frame list to pass through."""
        return [frame]

    def update_params(self, params: dict):
        """Handle real-time parameter updates from the gateway or client."""
        pass

import { BYOCClient } from '@muxionlabs/byoc-sdk';

const client = new BYOCClient({
  gatewayUrl: 'https://<livepeer-gateway-url>',
  capability: 'live-video-to-video',
});

await client.startStream({ videoElement, params: { /* pipeline params */ } });

pip install git+https://github.com/livepeer/pytrickle.git
python processor.py