From 080f70d91ce65d1c93574f5a12fdcb98a69c1cea Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 29 Apr 2025 10:32:42 -0300 Subject: [PATCH] Allowing to define the username and credential for the ice servers. --- .../transports/network/webrtc_connection.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py index 81ece3af7..d21b57cad 100644 --- a/src/pipecat/transports/network/webrtc_connection.py +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -7,7 +7,7 @@ import asyncio import json import time -from typing import Any, Literal, Optional, Union +from typing import Any, List, Literal, Optional, Union from av.frame import Frame from loguru import logger @@ -87,13 +87,21 @@ class SmallWebRTCTrack: return getattr(self._track, name) +# Alias so we don't need to expose RTCIceServer +IceServer = RTCIceServer + + class SmallWebRTCConnection(BaseObject): - def __init__(self, ice_servers=None): + def __init__(self, ice_servers: Optional[Union[List[str], List[IceServer]]] = None): super().__init__() - if ice_servers: - self.ice_servers = [RTCIceServer(urls=server) for server in ice_servers] + if not ice_servers: + self.ice_servers: List[IceServer] = [] + elif all(isinstance(s, IceServer) for s in ice_servers): + self.ice_servers = ice_servers + elif all(isinstance(s, str) for s in ice_servers): + self.ice_servers = [IceServer(urls=s) for s in ice_servers] else: - self.ice_servers = [] + raise TypeError("ice_servers must be either List[str] or List[RTCIceServer]") self._connect_invoked = False self._track_map = {} self._track_getters = {