Streamlit WebRTC

https://github.com/whitphx/streamlit-webrtc
Admin
Streamlit WebRTC enables real-time video and audio streaming and processing in Streamlit...
Tokens:11,183
Snippets:179
Trust Score:9.7
License:MIT
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install streamlit-webrtc

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/index.md

Use pip to install the necessary packages for real-time media processing in Streamlit.

```bash
pip install streamlit streamlit-webrtc
```

--------------------------------

### Install project dependencies

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/CONTRIBUTING.md

Use uv to synchronize the project environment with required dependencies.

```shell
$ uv sync
```

--------------------------------

### Run the frontend development server

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/CONTRIBUTING.md

Navigate to the frontend directory and start the development server using pnpm.

```shell
$ cd streamlit_webrtc/frontend
$ pnpm dev
```

--------------------------------

### Install pre-commit hooks

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/CONTRIBUTING.md

Initialize pre-commit hooks to ensure code quality before commits.

```shell
$ pre-commit install
```

--------------------------------

### Install and Run streamlit-remote

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/deployment.md

Commands to install the streamlit-remote utility and execute it with different providers or local HTTPS configurations.

```bash
pip install streamlit-remote
st-remote your_app.py --provider cloudflare  # Remote HTTPS URL via cloudflared.
st-remote your_app.py --provider ngrok  # Remote HTTPS URL via ngrok.
st-remote your_app.py --https self-signed --no-remote  # Local HTTPS only.
# Then access the HTTPS URL printed by st-remote.
```

--------------------------------

### Initialize WebRTC Streamer

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/README.md

Basic setup for a WebRTC streamer component in a Streamlit application.

```py
from streamlit_webrtc import webrtc_streamer

webrtc_streamer(key="sample")
```

--------------------------------

### Implement SENDRECV mode orchestrator

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/06-refactor-process-offer-coro.md

Example of the refactored SENDRECV mode branch using the new helper functions.

```python
if mode == WebRtcMode.SENDRECV:
    @pc.listens_to("track")
    def on_track(input_track):
        _notify_track_created(on_track_created, "input", input_track)
        output_track = _build_output_track(
            kind=input_track.kind,
            input_track=input_track,
            source_track=source_audio_track if input_track.kind == "audio" else source_video_track,
            processor=audio_processor if input_track.kind == "audio" else video_processor,
            async_processing=async_processing,
            relay=relay,
        )
        if _should_sendback(output_track.kind, sendback_video, sendback_audio):
            pc.addTrack(relay.subscribe(output_track))
        if out_recorder:
            out_recorder.addTrack(relay.subscribe(output_track))
        if in_recorder:
            in_recorder.addTrack(relay.subscribe(input_track))
        _notify_track_created(on_track_created, "output", output_track)
        # ...ended handler unchanged
```

--------------------------------

### Build the package locally

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/CONTRIBUTING.md

Run the build process manually using the Makefile command.

```shell
$ make build
```

--------------------------------

### Run Streamlit Application

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/tutorial.md

Command to execute the Streamlit application after creating the app.py file.

```bash
streamlit run app.py
```

--------------------------------

### Run the Streamlit application

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/CONTRIBUTING.md

Execute the main application script using the Streamlit CLI.

```shell
$ streamlit run home.py
```

--------------------------------

### Configure STUN server for remote deployment

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/README.md

Pass an iceServers configuration to the rtc_configuration argument in webrtc_streamer to enable WebRTC connectivity on remote hosts.

```python
webrtc_streamer(
    # ...
    rtc_configuration={  # Add this config
        "iceServers": [{"urls": ["stun:stun.l.google.com:19302"]}]
    }
    # ...
)
```

--------------------------------

### Create a video source track with default session-based lifecycle

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/tutorial.md

Uses the default lifecycle scope where the track is tied to the active WebRTC session and cleared upon disconnection.

```python
from streamlit_webrtc import create_video_source_track

video_track = create_video_source_track(
    callback=video_source_callback,
    key="video-source",
)
```

--------------------------------

### Create a changelog fragment

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/AGENTS.md

Generates a new timestamped changelog file in the changelog.d/ directory using the scriv tool.

```bash
uv run scriv create --edit
```

--------------------------------

### Create a video source track with Streamlit-session lifecycle

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/tutorial.md

Overrides the default behavior to persist the track across multiple WebRTC sessions within the same Streamlit session.

```python
video_track = create_video_source_track(
    callback=video_source_callback,
    key="video-source",
    lifecycle_scope="streamlit-session",
)
```

--------------------------------

### Run local tests and linters

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/AGENTS.md

Executes the test suite and pre-commit hooks to ensure code quality before submitting a pull request.

```bash
uv run pytest
```

```bash
pre-commit run --all-files
```

--------------------------------

### Configure Twilio TURN server for webrtc_streamer

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/deployment.md

Uses the Twilio Python SDK to generate ICE server credentials and pass them to the rtc_configuration argument.

```python
## This sample code is from https://www.twilio.com/docs/stun-turn/api
# Download the helper library from https://www.twilio.com/docs/python/install
import os
from twilio.rest import Client

# Find your Account SID and Auth Token at twilio.com/console
# and set the environment variables. See http://twil.io/secure
account_sid = os.environ['TWILIO_ACCOUNT_SID']
auth_token = os.environ['TWILIO_AUTH_TOKEN']
client = Client(account_sid, auth_token)

token = client.tokens.create()

# Then, pass the ICE server information to webrtc_streamer().
webrtc_streamer(
    # ...
    rtc_configuration={
        "iceServers": token.ice_servers
    }
    # ...
)
```

--------------------------------

### Initialize attribute in __init__

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/03-fix-ice-candidate-mutable-class-attr.md

Move the initialization of the tracking set to the instance level within the constructor.

```python
def __init__(self, ...):
    ...
    self._added_ice_candidate_ids: Set[str] = set()
```

--------------------------------

### Verify instance-level tracking with a unit test

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/03-fix-ice-candidate-mutable-class-attr.md

A test case to ensure that multiple worker instances maintain independent tracking of ICE candidate IDs.

```python
def test_each_worker_tracks_its_own_added_candidate_ids(monkeypatch):
    monkeypatch.setattr(
        "streamlit_webrtc.webrtc.candidate_from_sdp",
        lambda sdp: object(),
    )
    w1 = make_worker_without_starting()
    w2 = make_worker_without_starting()
    calls1, calls2 = [], []
    w1.add_ice_candidate = lambda c: calls1.append(c)
    w2.add_ice_candidate = lambda c: calls2.append(c)
    cand = {"id1": {"candidate": "candidate:...", "sdpMid": "0", "sdpMLineIndex": 0}}
    w1.set_ice_candidates_from_offerer(cand)
    w2.set_ice_candidates_from_offerer(cand)
    assert len(calls1) == 1
    assert len(calls2) == 1  # would be 0 with the bug
```

--------------------------------

### Configure Media Toggle Controls

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/tutorial.md

Disables the camera and microphone toggle buttons in the UI by setting media_toggle_controls to False.

```python
from streamlit_webrtc import webrtc_streamer

webrtc_streamer(key="example", media_toggle_controls=False)
```

--------------------------------

### Standard Property Implementation

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/08-cleanup-webrtc-streamer-context.md

The existing pattern for pass-through properties in WebRtcStreamerContext, which is repeated for 11 different attributes.

```python
@property
def output_audio_track(self) -> Optional[MediaStreamTrack]:
    worker = self._get_worker()
    return worker.output_audio_track if worker else None
```

--------------------------------

### Implement loopback integration test for WebRTC worker

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/05-add-pytest-layers.md

Uses aiortc to validate signaling and frame flow without a network by creating a loopback connection.

```python
import asyncio
from aiortc import RTCPeerConnection
from aiortc.contrib.media import MediaRelay
from streamlit_webrtc.webrtc import WebRtcWorker, WebRtcMode
from streamlit_webrtc.source import VideoSourceTrack

class _NoopShutdownObserver:
    def stop(self, timeout=1.0): pass

def _noop_observer_factory(_cb):
    return _NoopShutdownObserver()

async def test_worker_processes_frames_in_sendonly_mode(make_video_frame):
    loop = asyncio.get_running_loop()
    relay = MediaRelay()

    client = RTCPeerConnection()
    client.addTrack(VideoSourceTrack(make_video_frame, fps=10))
    offer = await client.createOffer()
    await client.setLocalDescription(offer)

    received: list = []
    def cb(frame):
        received.append(frame)
        return frame

    worker = WebRtcWorker(
        mode=WebRtcMode.SENDONLY,
        rtc_configuration=None,
        ...,
        video_frame_callback=cb,
        loop=loop,
        relay=relay,
        session_shutdown_observer_factory=_noop_observer_factory,
    )
    answer = worker.process_offer(client.localDescription.sdp, client.localDescription.type, timeout=5)
    await client.setRemoteDescription(answer)

    # Pump frames for a few seconds
    deadline = loop.time() + 5
    while loop.time() < deadline and len(received) < 3:
        await asyncio.sleep(0.1)
    assert len(received) >= 3

    worker.stop()
    await client.close()
```

--------------------------------

### Configure Logging Levels

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/README.md

Adjust the logging verbosity for the library and its internal dependencies like aiortc.

```python
st_webrtc_logger = logging.getLogger("streamlit_webrtc")
st_webrtc_logger.setLevel(logging.WARNING)
```

```python
aioice_logger = logging.getLogger("aioice")
aioice_logger.setLevel(logging.WARNING)
```

--------------------------------

### Provision Issue Labels

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/ai-issue-workflow.md

Executes a bash script to idempotently create the required labels for the issue workflow.

```bash
bash scripts/setup-issue-labels.sh
```

--------------------------------

### Define test file layout

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/05-add-pytest-layers.md

The directory structure for organizing tests into distinct layers.

```text
tests/
  conftest.py               # existing — keep
  import_test.py            # existing — keep
  session_info_test.py      # existing — keep (touched by item 01)
  source_test.py            # existing — keep
  config_test.py            # NEW — layer 1
  models_test.py            # NEW — layer 1 (CallbackAttachableProcessor)
  component_pure_test.py    # NEW — layer 1 (compile_state, generate_frontend_component_key)
  process_test.py           # NEW — layer 2
  mix_test.py               # NEW — layer 2
  webrtc_loopback_test.py   # NEW — layer 3
```

--------------------------------

### Define _build_output_track helper

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/06-refactor-process-offer-coro.md

Helper function to determine the output track based on input, source, and processor availability.

```python
def _build_output_track(
    *,
    kind: Literal["video", "audio"],
    input_track: Optional[MediaStreamTrack],
    source_track: Optional[MediaStreamTrack],
    processor: Optional[ProcessorBase],
    async_processing: bool,
    relay: MediaRelay,
) -> Optional[MediaStreamTrack]:
    """Return the track that should be sent back / handed to the receiver / recorded.

    Precedence: source_track > processor-wrapped input > raw input > None.
    Raises if neither input_track nor source_track is provided when a processor exists.
    """
```

--------------------------------

### Implement Session Cleanup with on_video_ended

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/README.md

Use the on_video_ended callback to clean up session-specific resources when a media track stops. Note that this callback runs on an asyncio loop, so standard Streamlit commands are not supported.

```python
import streamlit as st
from streamlit_webrtc import webrtc_streamer


def video_frame_callback(frame):
    # ... process the frame, possibly initializing per-session state on first call ...
    return frame


def on_video_ended():
    st.session_state.pop("my_session_resource", None)


webrtc_streamer(
    key="example",
    video_frame_callback=video_frame_callback,
    on_video_ended=on_video_ended,
)
```

--------------------------------

### Serve Streamlit apps via HTTPS with streamlit-remote

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/README.md

Use the st-remote command to serve applications over HTTPS, either via a tunnel provider or using self-signed certificates for local development.

```shell
$ pip install streamlit-remote
$ st-remote your_app.py --provider cloudflare  # Remote HTTPS URL via cloudflared.
$ st-remote your_app.py --provider ngrok  # Remote HTTPS URL via ngrok.
$ st-remote your_app.py --https self-signed --no-remote  # Local HTTPS only.
# Then access the HTTPS URL printed by st-remote.
```

--------------------------------

### Define _notify_track_created helper

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/06-refactor-process-offer-coro.md

Helper function to dispatch track creation events based on track kind and role.

```python
def _notify_track_created(
    callback: Callable[[TrackType, MediaStreamTrack], None],
    role: Literal["input", "output"],
    track: MediaStreamTrack,
) -> None:
    if track.kind == "video":
        callback(f"{role}:video", track)
    elif track.kind == "audio":
        callback(f"{role}:audio", track)
```

--------------------------------

### Simplify event loop retrieval

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/01-bump-streamlit-minimum.md

Collapse the event loop retrieval logic to the modern implementation available in newer Streamlit versions.

```python
from streamlit.runtime.runtime import Runtime
def get_global_event_loop():
    return Runtime.instance()._get_async_objs().eventloop
```

--------------------------------

### Add Video Processing

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/tutorial.md

Use the video_frame_callback argument to receive and return PyAV VideoFrame objects for real-time image manipulation.

```python
--8<-- "./examples/tutorial/02_video_processing.py"
```

--------------------------------

### Issue Automation Workflow Diagram

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/ai-issue-workflow.md

Visual representation of the issue lifecycle from creation to implementation or closure.

```text
Issue created
    ↓
[claude-issue-triage] ──→ needs-info / triaged / out-of-scope
                                   ↓
                          (maintainer applies a label)
                                   ↓
                       ai-implement ─→ [implement] ─→ PR
                       wontfix      ─→ [wontfix]   ─→ closed
```

--------------------------------

### ICE candidate handling helper

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/07-refactor-webrtc-streamer-function.md

A wrapper function to process ICE candidates from the component value if a worker is active.

```python
def _handle_ice_candidates(context, component_value):
    worker = context._get_worker()
    if worker and component_value and component_value.get("iceCandidates"):
        worker.set_ice_candidates_from_offerer(component_value["iceCandidates"])
```

--------------------------------

### Inject dependencies into WebRtcWorker constructor

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/04-decouple-worker-from-streamlit-runtime.md

Modify the constructor to accept optional event loop, relay, and shutdown observer factory arguments, defaulting to Streamlit-resolved instances to maintain backward compatibility.

```python
class WebRtcWorker(Generic[...]):
    def __init__(
        self,
        mode: WebRtcMode,
        ...,
        # NEW — keyword-only, defaults to Streamlit-resolved
        loop: Optional[asyncio.AbstractEventLoop] = None,
        relay: Optional[MediaRelay] = None,
        session_shutdown_observer_factory: Optional[
            Callable[[Callable[[], None]], "SessionShutdownObserver"]
        ] = None,
    ) -> None:
        self._loop = loop or get_global_event_loop()
        self._relay = relay or get_global_relay()
        ...
        if session_shutdown_observer_factory is None:
            self._session_shutdown_observer = SessionShutdownObserver(self.stop)
        else:
            self._session_shutdown_observer = session_shutdown_observer_factory(self.stop)
```

--------------------------------

### Simplify always-truthy conditional in process.py

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/02-delete-dead-code.md

The check for self.processor.recv is redundant as it is always a bound method, making the fallback return branch unreachable.

```python
async def _fallback_recv_queued(self, frames: List[FrameT]) -> List[FrameT]:
    if len(frames) > 1:
        logger.warning(...)
    if self.processor.recv:        # ← always truthy (bound method)
        return [self.processor.recv(frames[-1])]
    return [frames[-1]]            # ← dead branch
```

--------------------------------

### Remove Streamlit 0.84.0 JSON hotfix

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/02-delete-dead-code.md

The legacy workaround for Streamlit 0.84.0 is no longer required as the minimum supported version has increased.

```python
# HOTFIX: The return value from _component_func()
#         is of type str with streamlit==0.84.0.
# See https://github.com/whitphx/streamlit-webrtc/issues/287
component_value: Union[Dict, None]
if isinstance(component_value_raw, str):
    LOGGER.warning("The component value is of type str")
    component_value = json.loads(component_value_raw)
else:
    component_value = component_value_raw
```

--------------------------------

### Update pyproject.toml dependencies

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/01-bump-streamlit-minimum.md

Update the minimum Streamlit version requirement in the project configuration.

```toml
dependencies = [
    "streamlit>=1.51.0",  # First Streamlit version requiring Python >=3.10
    ...
]
```

--------------------------------

### Refactored webrtc_streamer orchestration

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/07-refactor-webrtc-streamer-function.md

The main function body after extracting logic into specialized helper functions.

```python
def webrtc_streamer(key, mode=..., **kwargs) -> WebRtcStreamerContext:
    _apply_argument_defaults_and_deprecations(kwargs)
    context = _get_or_create_context(key)
    component_value = _render_frontend(key, context, **frontend_kwargs(kwargs))
    component_value = _restore_snapshot_if_needed(context, component_value)
    _handle_worker_lifecycle(context, component_value, key, **worker_kwargs(kwargs))
    _handle_ice_candidates(context, component_value)
    _update_callbacks(context, **callback_kwargs(kwargs))
    return context
```

--------------------------------

### Pass Parameters to Callback

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/tutorial.md

Pass custom arguments to the callback function to control processing logic dynamically.

```python
--8<-- "./examples/tutorial/03_parameters.py"
```

--------------------------------

### Descriptor-based Property Forwarding

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/08-cleanup-webrtc-streamer-context.md

A refactoring approach using a descriptor class to reduce boilerplate for forwarded properties. This significantly reduces line count but may impact IDE navigation.

```python
class _WorkerForwarded:
    def __init__(self, attr): self._attr = attr
    def __set_name__(self, owner, name): self._attr = self._attr or name
    def __get__(self, instance, owner=None):
        if instance is None: return self
        worker = instance._get_worker()
        return getattr(worker, self._attr) if worker else None

class WebRtcStreamerContext(Generic[VideoProcessorT, AudioProcessorT]):
    video_processor: Optional[Union[VideoProcessorT, CallbackAttachableProcessor]] = _WorkerForwarded("video_processor")
    audio_processor: Optional[Union[AudioProcessorT, CallbackAttachableProcessor]] = _WorkerForwarded("audio_processor")
    video_receiver: Optional[VideoReceiver] = _WorkerForwarded("video_receiver")
    # ...
```

--------------------------------

### Pull Values from Callback

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/docs/tutorial.md

Use thread-safe containers like threading.Lock to share data between the callback thread and the main script loop.

```python
--8<-- "./examples/tutorial/04_pull_values.py"
```

--------------------------------

### Pass image frames from callback to outer scope

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/README.md

Uses a threading.Lock and a shared dictionary to safely pass image frames from the callback to the main script for processing.

```python
import threading

import cv2
import streamlit as st
from matplotlib import pyplot as plt

from streamlit_webrtc import webrtc_streamer

lock = threading.Lock()
img_container = {"img": None}


def video_frame_callback(frame):
    img = frame.to_ndarray(format="bgr24")
    with lock:
        img_container["img"] = img

    return frame


ctx = webrtc_streamer(key="example", video_frame_callback=video_frame_callback)

fig_place = st.empty()
fig, ax = plt.subplots(1, 1)

while ctx.state.playing:
    with lock:
        img = img_container["img"]
    if img is None:
        continue
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    ax.cla()
    ax.hist(gray.ravel(), 256, [0, 256])
    fig_place.pyplot(fig)
```

--------------------------------

### Pass Parameters to Callback

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/README.md

Pass external state or parameters into the frame processing callback.

```python
import streamlit as st
from streamlit_webrtc import webrtc_streamer
import av


flip = st.checkbox("Flip")


def video_frame_callback(frame):
    img = frame.to_ndarray(format="bgr24")

    flipped = img[::-1,:,:] if flip else img

    return av.VideoFrame.from_ndarray(flipped, format="bgr24")


webrtc_streamer(key="example", video_frame_callback=video_frame_callback)
```

--------------------------------

### Remove _test() block in webrtc.py

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/02-delete-dead-code.md

The _test function and its __main__ guard are non-functional and unused by test runners.

```python
def _test():
    ...

if __name__ == "__main__":
    ...
    _test()
```

--------------------------------

### Identify the class-level mutable attribute

Source: https://github.com/whitphx/streamlit-webrtc/blob/main/refactoring/03-fix-ice-candidate-mutable-class-attr.md

The original implementation defined _added_ice_candidate_ids as a class attribute, causing it to be shared across all instances.

```python
class WebRtcWorker(Generic[VideoProcessorT, AudioProcessorT]):
    ...
    _added_ice_candidate_ids: Set[str] = set()   # ← class-level shared mutable

    def set_ice_candidates_from_offerer(self, candidates: Dict[str, Dict]):
        ...
        for candidate_id, candidate_dict in candidates.items():
            if candidate_id in self._added_ice_candidate_ids:
                continue
            ...
            self._added_ice_candidate_ids.add(candidate_id)
```
Tokens:
Raw