Skip to content

Client

WebRTC client for the Inference SDK.

WebRTCClient

Namespaced WebRTC API bound to an InferenceHTTPClient instance.

Provides a unified streaming interface for different video sources (webcam, RTSP, video files, manual frames).

Source code in inference_sdk/webrtc/client.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class WebRTCClient:
    """Namespaced WebRTC API bound to an InferenceHTTPClient instance.

    Provides a unified streaming interface for different video sources
    (webcam, RTSP, video files, manual frames).
    """

    @experimental(
        info="WebRTC SDK is experimental and under active development. "
        "API may change in future releases. Please report issues at "
        "https://github.com/roboflow/inference/issues"
    )
    def __init__(self, api_url: str, api_key: Optional[str]) -> None:
        """Initialize WebRTC client.

        Args:
            api_url: Base URL for the inference API
            api_key: API key for authentication (optional)
        """
        self._api_url = api_url
        self._api_key = api_key

    def stream(
        self,
        source: StreamSource,
        *,
        workflow: Union[str, dict],
        image_input: str = "image",
        workspace: Optional[str] = None,
        config: Optional[StreamConfig] = None,
    ) -> WebRTCSession:
        """Create a WebRTC streaming session.

        Args:
            source: Stream source (WebcamSource, RTSPSource, VideoFileSource, or ManualSource)
            workflow: Either a workflow ID (str) or workflow specification (dict)
            image_input: Name of the image input in the workflow
            workspace: Workspace name (required if workflow is an ID string)
            config: Stream configuration (output routing, FPS, TURN server, etc.)

        Returns:
            WebRTCSession context manager

        Raises:
            InvalidParameterError: If workflow/workspace parameters are invalid

        Examples:
            # Pattern 1: Using run() with decorators (recommended, auto-cleanup)
            from inference_sdk.webrtc import WebcamSource

            session = client.webrtc.stream(
                source=WebcamSource(resolution=(1920, 1080)),
                workflow="object-detection",
                workspace="my-workspace"
            )

            @session.on_frame
            def process_frame(frame, metadata):
                cv2.imshow("Frame", frame)
                if cv2.waitKey(1) & 0xFF == ord('q'):
                    session.close()

            session.run()  # Auto-closes on exception or stream end

            # Pattern 2: Using video() iterator (requires context manager or explicit close)
            from inference_sdk.webrtc import RTSPSource

            # Option A: With context manager (recommended)
            with client.webrtc.stream(
                source=RTSPSource("rtsp://camera.local/stream"),
                workflow=workflow_spec_dict
            ) as session:
                for frame, metadata in session.video():
                    cv2.imshow("Frame", frame)
                    if cv2.waitKey(1) & 0xFF == ord('q'):
                        break
            # Auto-cleanup on exit

            # Option B: Manual cleanup (not recommended)
            session = client.webrtc.stream(source=RTSPSource("rtsp://..."), ...)
            for frame, metadata in session.video():
                process(frame)
            session.close()  # Must call close() explicitly!
        """
        # Validate workflow configuration
        workflow_config = self._parse_workflow_config(workflow, workspace)

        # Use default config if not provided
        if config is None:
            config = StreamConfig()

        # Create session
        return WebRTCSession(
            api_url=self._api_url,
            api_key=self._api_key,
            source=source,
            image_input_name=image_input,
            workflow_config=workflow_config,
            stream_config=config,
        )

    def _parse_workflow_config(
        self, workflow: Union[str, dict], workspace: Optional[str]
    ) -> dict:
        """Parse workflow configuration from inputs.

        Args:
            workflow: Either workflow ID (str) or specification (dict)
            workspace: Workspace name (required for ID mode)

        Returns:
            Dictionary with workflow configuration

        Raises:
            InvalidParameterError: If configuration is invalid
        """
        if isinstance(workflow, str):
            # Workflow ID mode - requires workspace
            if not workspace:
                raise InvalidParameterError(
                    "workspace parameter required when workflow is an ID string"
                )
            return {"workflow_id": workflow, "workspace_name": workspace}
        elif isinstance(workflow, dict):
            # Workflow specification mode
            return {"workflow_specification": workflow}
        else:
            raise InvalidParameterError(
                f"workflow must be a string (ID) or dict (specification), got {type(workflow)}"
            )

__init__(api_url, api_key)

Initialize WebRTC client.

Parameters:

Name Type Description Default
api_url str

Base URL for the inference API

 required
api_key Optional[str]

API key for authentication (optional)

 required
Source code in inference_sdk/webrtc/client.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@experimental(
    info="WebRTC SDK is experimental and under active development. "
    "API may change in future releases. Please report issues at "
    "https://github.com/roboflow/inference/issues"
)
def __init__(self, api_url: str, api_key: Optional[str]) -> None:
    """Initialize WebRTC client.

    Args:
        api_url: Base URL for the inference API
        api_key: API key for authentication (optional)
    """
    self._api_url = api_url
    self._api_key = api_key

stream(source, *, workflow, image_input='image', workspace=None, config=None)

Create a WebRTC streaming session.

Parameters:

Name Type Description Default
source StreamSource

Stream source (WebcamSource, RTSPSource, VideoFileSource, or ManualSource)

 required
workflow Union[str, dict]

Either a workflow ID (str) or workflow specification (dict)

 required
image_input str

Name of the image input in the workflow

 'image'
workspace Optional[str]

Workspace name (required if workflow is an ID string)

 None
config Optional[StreamConfig]

Stream configuration (output routing, FPS, TURN server, etc.)

 None

Returns:

Type Description
WebRTCSession

WebRTCSession context manager

Raises:

Type Description
InvalidParameterError

If workflow/workspace parameters are invalid

Examples:

Pattern 1: Using run() with decorators (recommended, auto-cleanup)

from inference_sdk.webrtc import WebcamSource

session = client.webrtc.stream( source=WebcamSource(resolution=(1920, 1080)), workflow="object-detection", workspace="my-workspace" )

@session.on_frame def process_frame(frame, metadata): cv2.imshow("Frame", frame) if cv2.waitKey(1) & 0xFF == ord('q'): session.close()

session.run() # Auto-closes on exception or stream end

Pattern 2: Using video() iterator (requires context manager or explicit close)

from inference_sdk.webrtc import RTSPSource

Option A: With context manager (recommended)

with client.webrtc.stream( source=RTSPSource("rtsp://camera.local/stream"), workflow=workflow_spec_dict ) as session: for frame, metadata in session.video(): cv2.imshow("Frame", frame) if cv2.waitKey(1) & 0xFF == ord('q'): break

Auto-cleanup on exit

Option B: Manual cleanup (not recommended)

session = client.webrtc.stream(source=RTSPSource("rtsp://..."), ...) for frame, metadata in session.video(): process(frame) session.close() # Must call close() explicitly!

Source code in inference_sdk/webrtc/client.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def stream(
    self,
    source: StreamSource,
    *,
    workflow: Union[str, dict],
    image_input: str = "image",
    workspace: Optional[str] = None,
    config: Optional[StreamConfig] = None,
) -> WebRTCSession:
    """Create a WebRTC streaming session.

    Args:
        source: Stream source (WebcamSource, RTSPSource, VideoFileSource, or ManualSource)
        workflow: Either a workflow ID (str) or workflow specification (dict)
        image_input: Name of the image input in the workflow
        workspace: Workspace name (required if workflow is an ID string)
        config: Stream configuration (output routing, FPS, TURN server, etc.)

    Returns:
        WebRTCSession context manager

    Raises:
        InvalidParameterError: If workflow/workspace parameters are invalid

    Examples:
        # Pattern 1: Using run() with decorators (recommended, auto-cleanup)
        from inference_sdk.webrtc import WebcamSource

        session = client.webrtc.stream(
            source=WebcamSource(resolution=(1920, 1080)),
            workflow="object-detection",
            workspace="my-workspace"
        )

        @session.on_frame
        def process_frame(frame, metadata):
            cv2.imshow("Frame", frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                session.close()

        session.run()  # Auto-closes on exception or stream end

        # Pattern 2: Using video() iterator (requires context manager or explicit close)
        from inference_sdk.webrtc import RTSPSource

        # Option A: With context manager (recommended)
        with client.webrtc.stream(
            source=RTSPSource("rtsp://camera.local/stream"),
            workflow=workflow_spec_dict
        ) as session:
            for frame, metadata in session.video():
                cv2.imshow("Frame", frame)
                if cv2.waitKey(1) & 0xFF == ord('q'):
                    break
        # Auto-cleanup on exit

        # Option B: Manual cleanup (not recommended)
        session = client.webrtc.stream(source=RTSPSource("rtsp://..."), ...)
        for frame, metadata in session.video():
            process(frame)
        session.close()  # Must call close() explicitly!
    """
    # Validate workflow configuration
    workflow_config = self._parse_workflow_config(workflow, workspace)

    # Use default config if not provided
    if config is None:
        config = StreamConfig()

    # Create session
    return WebRTCSession(
        api_url=self._api_url,
        api_key=self._api_key,
        source=source,
        image_input_name=image_input,
        workflow_config=workflow_config,
        stream_config=config,
    )