nits

AngeloGiacco · AngeloGiacco · commit f9e8069af34f · 2025-09-18T12:49:35.000+01:00
diff --git a/src/elevenlabs/conversational_ai/base_connection.py b/src/elevenlabs/conversational_ai/base_connection.py
@@ -39,6 +39,22 @@ async def send_audio(self, audio_data: bytes) -> None:
         """Send audio data through the connection."""
         pass
 
+    def send_message_sync(self, message: dict) -> None:
+        """Send a message synchronously (for compatibility with sync code)."""
+        import asyncio
+        try:
+            # Try to get the current event loop
+            loop = asyncio.get_event_loop()
+            if loop.is_running():
+                # If loop is running, create a task
+                asyncio.create_task(self.send_message(message))
+            else:
+                # If loop is not running, run the coroutine
+                loop.run_until_complete(self.send_message(message))
+        except RuntimeError:
+            # No event loop, create new one
+            asyncio.run(self.send_message(message))
+
     def on_message(self, callback: Callable[[dict], Union[None, Awaitable[None]]]) -> None:
         """Set the message callback."""
         self._on_message_callback = callback
diff --git a/src/elevenlabs/conversational_ai/conversation.py b/src/elevenlabs/conversational_ai/conversation.py
@@ -622,17 +622,27 @@ def end_session(self):
         if self._connection:
             connection_type = self._determine_connection_type()
             if connection_type == ConnectionType.WEBRTC:
-                # For WebRTC, we need to close the connection in an async context
                 import asyncio
                 try:
-                    loop = asyncio.get_event_loop()
-                    if loop.is_running():
-                        asyncio.create_task(self._connection.close())
-                    else:
-                        loop.run_until_complete(self._connection.close())
-                except RuntimeError:
-                    # No event loop running, create a new one
-                    asyncio.run(self._connection.close())
+                    try:
+                        loop = asyncio.get_event_loop()
+                        if loop.is_running():
+                            task = asyncio.create_task(self._connection.close())
+                        else:
+                            asyncio.wait_for(
+                                loop.run_until_complete(self._connection.close()),
+                                timeout=5.0
+                            )
+                    except RuntimeError:
+                        async def cleanup():
+                            await asyncio.wait_for(self._connection.close(), timeout=5.0)
+
+                        asyncio.run(cleanup())
+
+                except asyncio.TimeoutError:
+                    print("Warning: WebRTC connection cleanup timed out")
+                except Exception as e:
+                    print(f"Warning: Error during WebRTC connection cleanup: {e}")
             self._connection = None
 
         if self.callback_end_session:
@@ -678,9 +688,7 @@ def send_user_message(self, text: str):
 
             event = UserMessageClientToOrchestratorEvent(text=text)
             try:
-                # Send through WebRTC connection
-                import asyncio
-                asyncio.create_task(self._connection.send_message(event.to_dict()))
+                self._connection.send_message_sync(event.to_dict())
             except Exception as e:
                 print(f"Error sending user message: {e}")
                 raise
@@ -768,8 +776,15 @@ def _run_webrtc(self):
         try:
             # Connect to WebRTC
             import asyncio
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
+
+            try:
+                loop = asyncio.get_event_loop()
+                if loop.is_closed():
+                    loop = asyncio.new_event_loop()
+                    asyncio.set_event_loop(loop)
+            except RuntimeError:
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
 
             async def webrtc_session():
                 await self._connection.connect()
diff --git a/src/elevenlabs/conversational_ai/webrtc_connection.py b/src/elevenlabs/conversational_ai/webrtc_connection.py
@@ -2,7 +2,14 @@
 import asyncio
 from typing import Optional, Dict, Any, Callable, Union, Awaitable
 import httpx
-from livekit.rtc import Room, TrackKind
+
+try:
+    from livekit.rtc import Room, TrackKind
+except ImportError:
+    raise ImportError(
+        "livekit package is required for WebRTC support. "
+        "Install with: pip install livekit"
+    )
 
 from .base_connection import BaseConnection
 
@@ -17,7 +24,7 @@ def __init__(
         api_origin: Optional[str] = None,
         overrides: Optional[Dict[str, Any]] = None,
         on_debug: Optional[Callable[[Dict[str, Any]], None]] = None,
-    ):
+    ) -> None:
         self.conversation_token = conversation_token
         self.agent_id = agent_id
         self.livekit_url = livekit_url
@@ -40,7 +47,7 @@ def __init__(
         api_origin: Optional[str] = None,
         overrides: Optional[Dict[str, Any]] = None,
         on_debug: Optional[Callable[[Dict[str, Any]], None]] = None,
-    ):
+    ) -> None:
         super().__init__()
         self.conversation_token = conversation_token
         self.agent_id = agent_id
@@ -49,7 +56,7 @@ def __init__(
         self.overrides = overrides or {}
         self.on_debug = on_debug
         self._room: Optional[Room] = None
-        self._is_connected = False
+        self._is_connected: bool = False
 
     @classmethod
     async def create(cls, config: WebRTCConnectionConfig) -> "WebRTCConnection":
@@ -68,46 +75,82 @@ async def create(cls, config: WebRTCConnectionConfig) -> "WebRTCConnection":
 
     async def connect(self) -> None:
         """Establish the WebRTC connection using LiveKit."""
-        # Get conversation token if not provided
-        if not self.conversation_token:
-            if not self.agent_id:
-                raise ValueError("Either conversation_token or agent_id is required for WebRTC connection")
-            self.conversation_token = await self._fetch_conversation_token()
-
-        # Create room and connect
-        self._room = Room()
-        self._setup_room_callbacks()
-
-        # Connect to LiveKit room using configurable URL
-        await self._room.connect(self.livekit_url, self.conversation_token)
-        self._is_connected = True
-
-        # Set conversation ID from room name if available
-        if self._room.name:
-            # Extract conversation ID from room name if it contains one
-            import re
-            match = re.search(r'(conv_[a-zA-Z0-9]+)', self._room.name)
-            self.conversation_id = match.group(0) if match else self._room.name
-        else:
-            self.conversation_id = f"webrtc-{id(self)}"
-
-        # Enable microphone
-        await self._room.local_participant.set_microphone_enabled(True)
-
-        # Send overrides if any
-        if self.overrides:
-            await self.send_message(self._construct_overrides())
+        try:
+            # Get conversation token if not provided
+            if not self.conversation_token:
+                if not self.agent_id:
+                    raise ValueError("Either conversation_token or agent_id is required for WebRTC connection")
+                self.conversation_token = await self._fetch_conversation_token()
 
-        self.debug({
-            "type": "conversation_initiation_client_data",
-            "message": self._construct_overrides()
-        })
+            # Create room and connect
+            self._room = Room()
+            self._setup_room_callbacks()
+
+            # Connect to LiveKit room using configurable URL
+            try:
+                await self._room.connect(self.livekit_url, self.conversation_token)
+                self._is_connected = True
+            except Exception as e:
+                self._is_connected = False
+                raise ConnectionError(f"Failed to connect to LiveKit room: {e}") from e
+
+            # Set conversation ID from room name if available
+            if self._room.name:
+                # Extract conversation ID from room name if it contains one
+                import re
+                match = re.search(r'(conv_[a-zA-Z0-9]+)', self._room.name)
+                self.conversation_id = match.group(0) if match else self._room.name
+            else:
+                self.conversation_id = f"webrtc-{id(self)}"
+
+            # Enable microphone
+            try:
+                await self._room.local_participant.set_microphone_enabled(True)
+            except Exception as e:
+                self.debug({
+                    "type": "microphone_enable_error",
+                    "error": str(e)
+                })
+                # Don't fail the connection for microphone issues
+
+            # Send overrides if any
+            if self.overrides:
+                try:
+                    await self.send_message(self._construct_overrides())
+                except Exception as e:
+                    self.debug({
+                        "type": "overrides_send_error",
+                        "error": str(e)
+                    })
+
+            self.debug({
+                "type": "conversation_initiation_client_data",
+                "message": self._construct_overrides()
+            })
+
+        except Exception as e:
+            # Ensure cleanup on connection failure
+            if self._room:
+                try:
+                    await self._room.disconnect()
+                except:
+                    pass
+                self._room = None
+            self._is_connected = False
+            raise
 
     async def close(self) -> None:
         """Close the WebRTC connection."""
         if self._room:
-            await self._room.disconnect()
-            self._room = None
+            try:
+                await self._room.disconnect()
+            except Exception as e:
+                self.debug({
+                    "type": "disconnect_error",
+                    "error": str(e)
+                })
+            finally:
+                self._room = None
         self._is_connected = False
 
     async def send_message(self, message: dict) -> None:
@@ -148,32 +191,54 @@ async def _fetch_conversation_token(self) -> str:
         if not self.agent_id:
             raise ValueError("Agent ID is required to fetch conversation token")
 
-        # Get version and source from overrides or use defaults
-        version = self.overrides.get("client", {}).get("version", "2.15.0")  # From pyproject.toml
-        source = self.overrides.get("client", {}).get("source", "python_sdk")
+        try:
+            # Get version and source from overrides or use defaults
+            version = self.overrides.get("client", {}).get("version", "2.15.0")  # From pyproject.toml
+            source = self.overrides.get("client", {}).get("source", "python_sdk")
+
+            # Convert WSS origin to HTTPS for API calls
+            api_origin = self._convert_wss_to_https(self.api_origin)
 
-        # Convert WSS origin to HTTPS for API calls
-        api_origin = self._convert_wss_to_https(self.api_origin)
+            url = f"{api_origin}/v1/convai/conversation/token?agent_id={self.agent_id}&source={source}&version={version}"
 
-        url = f"{api_origin}/v1/convai/conversation/token?agent_id={self.agent_id}&source={source}&version={version}"
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                try:
+                    response = await client.get(url)
+                except httpx.TimeoutException:
+                    raise ConnectionError(f"Timeout when fetching conversation token for agent {self.agent_id}")
+                except httpx.NetworkError as e:
+                    raise ConnectionError(f"Network error when fetching conversation token: {e}")
 
-        async with httpx.AsyncClient() as client:
-            response = await client.get(url)
+                if not response.is_success:
+                    error_msg = f"ElevenLabs API returned {response.status_code} {response.reason_phrase}"
+                    if response.status_code == 401:
+                        error_msg = "Your agent has authentication enabled, but no signed URL or conversation token was provided."
+                    elif response.status_code == 404:
+                        error_msg = f"Agent with ID {self.agent_id} not found"
+                    elif response.status_code == 429:
+                        error_msg = "Rate limit exceeded. Please try again later."
 
-            if not response.is_success:
-                error_msg = f"ElevenLabs API returned {response.status_code} {response.reason_phrase}"
-                if response.status_code == 401:
-                    error_msg = "Your agent has authentication enabled, but no signed URL or conversation token was provided."
+                    raise Exception(f"Failed to fetch conversation token for agent {self.agent_id}: {error_msg}")
 
-                raise Exception(f"Failed to fetch conversation token for agent {self.agent_id}: {error_msg}")
+                try:
+                    data = response.json()
+                except Exception as e:
+                    raise Exception(f"Invalid JSON response from API: {e}")
 
-            data = response.json()
-            token = data.get("token")
+                token = data.get("token")
 
-            if not token:
-                raise Exception("No conversation token received from API")
+                if not token:
+                    raise Exception("No conversation token received from API")
 
-            return token
+                return token
+
+        except Exception as e:
+            self.debug({
+                "type": "token_fetch_error",
+                "agent_id": self.agent_id,
+                "error": str(e)
+            })
+            raise
 
     def _convert_wss_to_https(self, origin: str) -> str:
         """Convert WSS origin to HTTPS for API calls."""
