Merge pull request #2853 from pipecat-ai/filipi/trickle_ice

Adding support for trickle ice.

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added support for trickle ICE to the `SmallWebRTCTransport`.
+
 - Added support for updating `OpenAITTSService` settings (`instructions` and
   `speed`) at runtime via `TTSUpdateSettingsFrame`.
 

src/pipecat/runner/run.py

@@ -204,6 +204,7 @@ def _setup_webrtc_routes(
 
         from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
         from pipecat.transports.smallwebrtc.request_handler import (
+            SmallWebRTCPatchRequest,
             SmallWebRTCRequest,
             SmallWebRTCRequestHandler,
         )
@@ -256,6 +257,13 @@ def _setup_webrtc_routes(
         )
         return answer
 
+    @app.patch("/api/offer")
+    async def ice_candidate(request: SmallWebRTCPatchRequest):
+        """Handle WebRTC new ice candidate requests."""
+        logger.debug(f"Received patch request: {request}")
+        await small_webrtc_handler.handle_patch_request(request)
+        return {"status": "success"}
+
     @asynccontextmanager
     async def smallwebrtc_lifespan(app: FastAPI):
         """Manage FastAPI application lifecycle and cleanup connections."""

src/pipecat/transports/smallwebrtc/connection.py

@@ -689,3 +689,8 @@ class SmallWebRTCConnection(BaseObject):
                 )()
                 if track:
                     track.set_enabled(signalling_message.enabled)
+
+    async def add_ice_candidate(self, candidate):
+        """Handle incoming ICE candidates."""
+        logger.debug(f"Adding remote candidate: {candidate}")
+        await self.pc.addIceCandidate(candidate)

src/pipecat/transports/smallwebrtc/request_handler.py

@@ -14,6 +14,7 @@ from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Awaitable, Callable, Dict, List, Optional
 
+from aiortc.sdp import candidate_from_sdp
 from fastapi import HTTPException
 from loguru import logger
 
@@ -39,6 +40,34 @@ class SmallWebRTCRequest:
     request_data: Optional[Any] = None
 
 
+@dataclass
+class IceCandidate:
+    """The remote ice candidate object received from the peer connection.
+
+    Parameters:
+        candidate: The ice candidate patch SDP string (Session Description Protocol).
+        sdp_mid: The SDP mid for the candidate patch.
+        sdp_mline_index: The SDP mline index for the candidate patch.
+    """
+
+    candidate: str
+    sdp_mid: str
+    sdp_mline_index: int
+
+
+@dataclass
+class SmallWebRTCPatchRequest:
+    """Small WebRTC transport session arguments for the runner.
+
+    Parameters:
+        pc_id: Identifier for the peer connection.
+        candidates: A list of ICE candidate patches.
+    """
+
+    pc_id: str
+    candidates: List[IceCandidate]
+
+
 class ConnectionMode(Enum):
     """Enum defining the connection handling modes."""
 
@@ -197,6 +226,19 @@ class SmallWebRTCRequestHandler:
             logger.debug(f"SmallWebRTC request details: {request}")
             raise
 
+    async def handle_patch_request(self, request: SmallWebRTCPatchRequest):
+        """Handle a SmallWebRTC patch candidate request."""
+        peer_connection = self._pcs_map.get(request.pc_id)
+
+        if not peer_connection:
+            raise HTTPException(status_code=404, detail="Peer connection not found")
+
+        for c in request.candidates:
+            candidate = candidate_from_sdp(c.candidate)
+            candidate.sdpMid = c.sdp_mid
+            candidate.sdpMLineIndex = c.sdp_mline_index
+            await peer_connection.add_ice_candidate(candidate)
+
     async def close(self):
         """Clear the connection map."""
         coros = [pc.disconnect() for pc in self._pcs_map.values()]