fix url generation

Author: PaulAsjes

src/elevenlabs/conversational_ai/conversation.py

@@ -328,9 +328,13 @@ def __init__(
         self._last_interrupt_id = 0
 
     def _get_wss_url(self):
+        import urllib.parse
         base_http_url = self.client._client_wrapper.get_base_url()
-        base_ws_url = base_http_url.replace("https://", "wss://").replace("http://", "ws://")
-        return f"{base_ws_url}/v1/convai/conversation?agent_id={self.agent_id}&source=python_sdk&version={__version__}"
+        base_ws_url = urllib.parse.urlparse(base_http_url)._replace(scheme="wss" if base_http_url.startswith("https") else "ws").geturl()
+        # Ensure base URL ends with '/' for proper joining
+        if not base_ws_url.endswith('/'):
+            base_ws_url += '/'
+        return f"{base_ws_url}v1/convai/conversation?agent_id={self.agent_id}&source=python_sdk&version={__version__}"
 
     def _get_signed_url(self):
         response = self.client.conversational_ai.conversations.get_signed_url(agent_id=self.agent_id)

tests/test_async_convai.py

@@ -414,3 +414,48 @@ async def test_async_conversation_wss_url_generation_without_get_environment():
 
         except Exception as e:
             assert False, f"Unexpected error generating WebSocket URL: {e}"
+
+
+@pytest.mark.asyncio
+async def test_async_websocket_url_construction_edge_cases():
+    """Test WebSocket URL construction edge cases for async conversation, specifically for trailing slash handling."""
+    from elevenlabs.core.client_wrapper import SyncClientWrapper
+    from elevenlabs.conversational_ai.conversation import AsyncConversation
+
+    # Test cases with various base URL formats
+    test_cases = [
+        # Base URLs without trailing slashes (the main edge case)
+        ("https://api.eu.residency.elevenlabs.io", "wss://api.eu.residency.elevenlabs.io"),
+        ("https://api.elevenlabs.io", "wss://api.elevenlabs.io"),
+        ("http://localhost:8000", "ws://localhost:8000"),
+        # Base URLs with trailing slashes (should still work)
+        ("https://api.eu.residency.elevenlabs.io/", "wss://api.eu.residency.elevenlabs.io"),
+        ("https://api.elevenlabs.io/", "wss://api.elevenlabs.io"),
+        ("http://localhost:8000/", "ws://localhost:8000"),
+    ]
+
+    for base_url, expected_ws_base in test_cases:
+        # Test async conversation WebSocket URL construction
+        mock_client = MagicMock()
+        mock_client._client_wrapper = SyncClientWrapper(
+            base_url=base_url,
+            api_key="test_key",
+            httpx_client=MagicMock(),
+            timeout=30.0
+        )
+
+        conversation = AsyncConversation(
+            client=mock_client,
+            agent_id=TEST_AGENT_ID,
+            requires_auth=False,
+            audio_interface=MockAsyncAudioInterface()
+        )
+
+        # Test conversation URL generation
+        conv_url = conversation._get_wss_url()
+        expected_conv_url = f"{expected_ws_base}/v1/convai/conversation"
+        assert expected_conv_url in conv_url, f"Async conversation URL should contain {expected_conv_url}, got {conv_url}"
+
+        # Ensure no double slashes in the path (except after the protocol)
+        url_path = conv_url.split("://", 1)[1]  # Remove protocol
+        assert "//" not in url_path, f"Async conversation URL should not contain double slashes in path: {conv_url}"

tests/test_convai.py

@@ -3,6 +3,9 @@
 import json
 import time
 
+from elevenlabs.core.client_wrapper import SyncClientWrapper
+from elevenlabs.conversational_ai.conversation import Conversation
+from elevenlabs.realtime_tts import RealtimeTextToSpeechClient
 
 class MockAudioInterface(AudioInterface):
     def start(self, input_callback):
@@ -256,3 +259,68 @@ def test_conversation_wss_url_generation_without_get_environment():
 
         except Exception as e:
             assert False, f"Unexpected error generating WebSocket URL: {e}"
+
+
+def test_websocket_url_construction_edge_cases():
+    """Test WebSocket URL construction edge cases, specifically for trailing slash handling."""
+
+    # Test cases with various base URL formats
+    test_cases = [
+        # Base URLs without trailing slashes (the main edge case)
+        ("https://api.eu.residency.elevenlabs.io", "wss://api.eu.residency.elevenlabs.io", "wss://api.eu.residency.elevenlabs.io"),
+        ("https://api.elevenlabs.io", "wss://api.elevenlabs.io", "wss://api.elevenlabs.io"),
+        ("http://localhost:8000", "ws://localhost:8000", "wss://localhost:8000"),
+        # Base URLs with trailing slashes (should still work)
+        ("https://api.eu.residency.elevenlabs.io/", "wss://api.eu.residency.elevenlabs.io", "wss://api.eu.residency.elevenlabs.io/"),
+        ("https://api.elevenlabs.io/", "wss://api.elevenlabs.io", "wss://api.elevenlabs.io/"),
+        ("http://localhost:8000/", "ws://localhost:8000", "wss://localhost:8000/"),
+    ]
+
+    for base_url, expected_ws_base, expected_tts_ws_base in test_cases:
+        # Test conversation WebSocket URL construction
+        mock_client = MagicMock()
+        mock_client._client_wrapper = SyncClientWrapper(
+            base_url=base_url,
+            api_key="test_key",
+            httpx_client=MagicMock(),
+            timeout=30.0
+        )
+
+        conversation = Conversation(
+            client=mock_client,
+            agent_id=TEST_AGENT_ID,
+            requires_auth=False,
+            audio_interface=MockAudioInterface()
+        )
+
+        # Test conversation URL generation
+        conv_url = conversation._get_wss_url()
+        expected_conv_url = f"{expected_ws_base}/v1/convai/conversation"
+        assert expected_conv_url in conv_url, f"Conversation URL should contain {expected_conv_url}, got {conv_url}"
+
+        # Ensure no double slashes in the path (except after the protocol)
+        url_path = conv_url.split("://", 1)[1]  # Remove protocol
+        assert "//" not in url_path, f"URL should not contain double slashes in path: {conv_url}"
+
+        # Test realtime TTS WebSocket URL construction
+        realtime_client = RealtimeTextToSpeechClient(client_wrapper=mock_client._client_wrapper)
+
+        # Test the WebSocket base URL construction
+        # Note: realtime TTS always uses wss scheme, not ws
+        assert realtime_client._ws_base_url == expected_tts_ws_base, f"TTS WebSocket base URL should be {expected_tts_ws_base}, got {realtime_client._ws_base_url}"
+
+        # Test full URL construction using urljoin (simulating the actual method)
+        import urllib.parse
+        test_voice_id = "test_voice_123"
+        test_model = "eleven_turbo_v2_5"
+        test_format = "mp3_44100_128"
+        relative_path = f"v1/text-to-speech/{test_voice_id}/stream-input?model_id={test_model}&output_format={test_format}"
+
+        full_tts_url = urllib.parse.urljoin(realtime_client._ws_base_url, relative_path)
+        # For URLs with trailing slash, expect it to be preserved in the joined URL
+        expected_tts_url_base = expected_tts_ws_base.rstrip('/') + "/v1/text-to-speech/" + test_voice_id + "/stream-input"
+        assert expected_tts_url_base in full_tts_url, f"TTS URL should contain {expected_tts_url_base}, got {full_tts_url}"
+
+        # Ensure no double slashes in the path
+        tts_url_path = full_tts_url.split("://", 1)[1]
+        assert "//" not in tts_url_path, f"TTS URL should not contain double slashes in path: {full_tts_url}"