add location

Author: AngeloGiacco

src/elevenlabs/conversational_ai/conversation.py

@@ -15,6 +15,7 @@
 from ..version import __version__
 from .base_connection import ConnectionType, BaseConnection
 from .connection_factory import create_connection, determine_connection_type
+from .location_utils import Location, get_origin_for_location
 
 
 class ClientToOrchestratorEvent(str, Enum):
@@ -299,6 +300,7 @@ def __init__(
         user_id: Optional[str] = None,
         connection_type: Optional[ConnectionType] = None,
         conversation_token: Optional[str] = None,
+        location: Optional[Location] = None,
         livekit_url: Optional[str] = None,
         api_origin: Optional[str] = None,
         webrtc_overrides: Optional[dict] = None,
@@ -310,6 +312,7 @@ def __init__(
         self.user_id = user_id
         self.connection_type = connection_type
         self.conversation_token = conversation_token
+        self.location = location
         self.livekit_url = livekit_url
         self.api_origin = api_origin
         self.webrtc_overrides = webrtc_overrides or {}
@@ -343,8 +346,12 @@ def __init__(
         self._connection: Optional[BaseConnection] = None
 
     def _get_wss_url(self):
-        base_http_url = self.client._client_wrapper.get_base_url()
-        base_ws_url = base_http_url.replace("https://", "wss://").replace("http://", "ws://")
+        # Use location-based URL if location is specified
+        if self.config.location is not None:
+            base_ws_url = get_origin_for_location(self.config.location)
+        else:
+            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__}"
 
     def _get_signed_url(self):

src/elevenlabs/conversational_ai/conversation_factory.py

@@ -11,6 +11,7 @@
 )
 from .webrtc_conversation import WebRTCConversation
 from .base_connection import ConnectionType
+from .location_utils import Location, get_origin_for_location, get_livekit_url_for_location
 
 
 def create_conversation(
@@ -21,6 +22,7 @@ def create_conversation(
     connection_type: ConnectionType = ConnectionType.WEBSOCKET,
     conversation_token: Optional[str] = None,
     requires_auth: bool = True,
+    location: Optional[Location] = None,
     audio_interface: Optional[Union[AudioInterface, AsyncAudioInterface]] = None,
     config: Optional[ConversationInitiationData] = None,
     client_tools: Optional[ClientTools] = None,
@@ -46,6 +48,7 @@ def create_conversation(
         connection_type: Type of connection (websocket or webrtc)
         conversation_token: Token for WebRTC authentication
         requires_auth: Whether authentication is required
+        location: Data residency location (us, eu-residency, in-residency, global)
         audio_interface: Audio interface for the conversation
         config: Conversation configuration
         client_tools: Client tools for handling agent calls
@@ -63,23 +66,22 @@ def create_conversation(
             audio_interface=your_audio_interface
         )
 
-        # WebRTC conversation
+        # WebRTC conversation with EU residency
         conversation = create_conversation(
             client=client,
             agent_id="your-agent-id",
             connection_type=ConnectionType.WEBRTC,
-            conversation_token="your-token",  # Optional, will fetch if not provided
+            location=Location.EU_RESIDENCY,
             audio_interface=your_async_audio_interface,
             async_callback_agent_response=your_response_handler
         )
 
-        # Public agent (no auth required)
+        # WebSocket conversation with specific location
         conversation = create_conversation(
             client=client,
-            agent_id="public-agent-id",
-            connection_type=ConnectionType.WEBRTC,
-            requires_auth=False,
-            audio_interface=your_async_audio_interface
+            agent_id="your-agent-id",
+            location=Location.IN_RESIDENCY,
+            audio_interface=your_audio_interface
         )
     """
 
@@ -90,17 +92,29 @@ def create_conversation(
     config.connection_type = connection_type
     if conversation_token:
         config.conversation_token = conversation_token
+    if location is not None:
+        config.location = location
 
     if connection_type == ConnectionType.WEBRTC:
         # Create WebRTC conversation
         if not isinstance(audio_interface, AsyncAudioInterface) and audio_interface is not None:
             raise ValueError("WebRTC conversations require an AsyncAudioInterface")
 
+        # Determine URLs based on location
+        livekit_url = None
+        api_origin = None
+        if location is not None:
+            livekit_url = get_livekit_url_for_location(location)
+            # Convert WSS to HTTPS for API origin
+            api_origin = get_origin_for_location(location).replace("wss://", "https://")
+
         return WebRTCConversation(
             client=client,
             agent_id=agent_id,
             user_id=user_id,
             conversation_token=conversation_token,
+            livekit_url=livekit_url,
+            api_origin=api_origin,
             audio_interface=audio_interface,
             config=config,
             client_tools=client_tools,
@@ -169,6 +183,7 @@ def create_webrtc_conversation(
     user_id: Optional[str] = None,
     *,
     conversation_token: Optional[str] = None,
+    location: Optional[Location] = None,
     livekit_url: Optional[str] = None,
     api_origin: Optional[str] = None,
     webrtc_overrides: Optional[dict] = None,
@@ -185,7 +200,17 @@ def create_webrtc_conversation(
     """Create a WebRTC conversation.
 
     Convenience function for creating WebRTC conversations with type safety.
+
+    Args:
+        location: Data residency location. If provided, overrides livekit_url and api_origin.
+        livekit_url: Custom LiveKit URL (overridden by location if provided).
+        api_origin: Custom API origin (overridden by location if provided).
     """
+    # Determine URLs based on location if provided
+    if location is not None:
+        livekit_url = get_livekit_url_for_location(location)
+        api_origin = get_origin_for_location(location).replace("wss://", "https://")
+
     return WebRTCConversation(
         client=client,
         agent_id=agent_id,
@@ -212,6 +237,7 @@ def create_websocket_conversation(
     user_id: Optional[str] = None,
     *,
     requires_auth: bool = True,
+    location: Optional[Location] = None,
     audio_interface: Optional[AudioInterface] = None,
     config: Optional[ConversationInitiationData] = None,
     client_tools: Optional[ClientTools] = None,
@@ -224,13 +250,17 @@ def create_websocket_conversation(
     """Create a WebSocket conversation.
 
     Convenience function for creating WebSocket conversations with type safety.
+
+    Args:
+        location: Data residency location (us, eu-residency, in-residency, global)
     """
     result = create_conversation(
         client=client,
         agent_id=agent_id,
         user_id=user_id,
         connection_type=ConnectionType.WEBSOCKET,
         requires_auth=requires_auth,
+        location=location,
         audio_interface=audio_interface,
         config=config,
         client_tools=client_tools,

src/elevenlabs/conversational_ai/location_utils.py

@@ -0,0 +1,50 @@
+from enum import Enum
+from typing import Dict
+
+
+class Location(Enum):
+    """Location enum for data residency and region selection."""
+    US = "us"
+    EU_RESIDENCY = "eu-residency"
+    IN_RESIDENCY = "in-residency"
+    GLOBAL = "global"
+
+
+def get_origin_for_location(location: Location) -> str:
+    """
+    Get the WebSocket API origin URL for a given location.
+
+    Args:
+        location: The location enum value
+
+    Returns:
+        The WebSocket URL for the specified location
+    """
+    origin_map: Dict[Location, str] = {
+        Location.US: "wss://api.elevenlabs.io",
+        Location.EU_RESIDENCY: "wss://api.eu.residency.elevenlabs.io",
+        Location.IN_RESIDENCY: "wss://api.in.residency.elevenlabs.io",
+        Location.GLOBAL: "wss://api.elevenlabs.io",
+    }
+
+    return origin_map[location]
+
+
+def get_livekit_url_for_location(location: Location) -> str:
+    """
+    Get the LiveKit WebRTC URL for a given location.
+
+    Args:
+        location: The location enum value
+
+    Returns:
+        The LiveKit URL for the specified location
+    """
+    livekit_url_map: Dict[Location, str] = {
+        Location.US: "wss://livekit.rtc.elevenlabs.io",
+        Location.EU_RESIDENCY: "wss://livekit.rtc.eu.residency.elevenlabs.io",
+        Location.IN_RESIDENCY: "wss://livekit.rtc.in.residency.elevenlabs.io",
+        Location.GLOBAL: "wss://livekit.rtc.elevenlabs.io",
+    }
+
+    return livekit_url_map[location]