Add asyncio event loop

Author: PaulAsjes

README.md

@@ -156,6 +156,100 @@ async def print_models() -> None:
 asyncio.run(print_models())
 ```
 
+## Conversational AI
+
+Build interactive AI agents with real-time audio capabilities using ElevenLabs Conversational AI.
+
+### Basic Usage
+
+```python
+from elevenlabs.client import ElevenLabs
+from elevenlabs.conversational_ai.conversation import Conversation, ClientTools
+from elevenlabs.conversational_ai.default_audio_interface import DefaultAudioInterface
+
+client = ElevenLabs(api_key="YOUR_API_KEY")
+
+# Create audio interface for real-time audio input/output
+audio_interface = DefaultAudioInterface()
+
+# Create conversation
+conversation = Conversation(
+    client=client,
+    agent_id="your-agent-id",
+    requires_auth=True,
+    audio_interface=audio_interface,
+)
+
+# Start the conversation
+conversation.start_session()
+
+# The conversation runs in background until you call:
+conversation.end_session()
+```
+
+### Custom Event Loop Support
+
+For advanced use cases involving context propagation, resource reuse, or specific event loop management, `ClientTools` supports custom asyncio event loops:
+
+```python
+import asyncio
+from elevenlabs.conversational_ai.conversation import ClientTools
+
+async def main():
+    # Get the current event loop
+    custom_loop = asyncio.get_running_loop()
+    
+    # Create ClientTools with custom loop to prevent "different event loop" errors
+    client_tools = ClientTools(loop=custom_loop)
+    
+    # Register your tools
+    async def get_weather(params):
+        location = params.get("location", "Unknown")
+        # Your async logic here
+        return f"Weather in {location}: Sunny, 72°F"
+    
+    client_tools.register("get_weather", get_weather, is_async=True)
+    
+    # Use with conversation
+    conversation = Conversation(
+        client=client,
+        agent_id="your-agent-id", 
+        requires_auth=True,
+        audio_interface=audio_interface,
+        client_tools=client_tools
+    )
+
+asyncio.run(main())
+```
+
+**Benefits of Custom Event Loop:**
+- **Context Propagation**: Maintain request-scoped state across async operations
+- **Resource Reuse**: Share existing async resources like HTTP sessions or database pools
+- **Loop Management**: Prevent "Task got Future attached to a different event loop" errors
+- **Performance**: Better control over async task scheduling and execution
+
+### Tool Registration
+
+Register custom tools that the AI agent can call during conversations:
+
+```python
+client_tools = ClientTools()
+
+# Sync tool
+def calculate_sum(params):
+    numbers = params.get("numbers", [])
+    return sum(numbers)
+
+# Async tool  
+async def fetch_data(params):
+    url = params.get("url")
+    # Your async HTTP request logic
+    return {"data": "fetched"}
+
+client_tools.register("calculate_sum", calculate_sum, is_async=False)
+client_tools.register("fetch_data", fetch_data, is_async=True)
+```
+
 ## Languages Supported
 
 Explore [all models & languages](https://elevenlabs.io/docs/models).

src/elevenlabs/conversational_ai/conversation.py

@@ -155,11 +155,18 @@ class ClientTools:
 
     Supports both synchronous and asynchronous tools running in a dedicated event loop,
     ensuring non-blocking operation of the main conversation thread.
+    
+    Args:
+        loop: Optional custom asyncio event loop to use for tool execution. If not provided,
+              a new event loop will be created and run in a separate thread. Using a custom
+              loop prevents "different event loop" runtime errors and allows for better
+              context propagation and resource management.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
         self.tools: Dict[str, Tuple[Union[Callable[[dict], Any], Callable[[dict], Awaitable[Any]]], bool]] = {}
         self.lock = threading.Lock()
+        self._custom_loop = loop
         self._loop = None
         self._thread = None
         self._running = threading.Event()
@@ -170,27 +177,39 @@ def start(self):
         if self._running.is_set():
             return
 
-        def run_event_loop():
-            self._loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(self._loop)
+        if self._custom_loop is not None:
+            # Use the provided custom event loop
+            self._loop = self._custom_loop
             self._running.set()
-            try:
-                self._loop.run_forever()
-            finally:
-                self._running.clear()
-                self._loop.close()
-                self._loop = None
+        else:
+            # Create and run our own event loop in a separate thread
+            def run_event_loop():
+                self._loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(self._loop)
+                self._running.set()
+                try:
+                    self._loop.run_forever()
+                finally:
+                    self._running.clear()
+                    self._loop.close()
+                    self._loop = None
 
-        self._thread = threading.Thread(target=run_event_loop, daemon=True, name="ClientTools-EventLoop")
-        self._thread.start()
-        # Wait for loop to be ready
-        self._running.wait()
+            self._thread = threading.Thread(target=run_event_loop, daemon=True, name="ClientTools-EventLoop")
+            self._thread.start()
+            # Wait for loop to be ready
+            self._running.wait()
 
     def stop(self):
         """Gracefully stop the event loop and clean up resources."""
         if self._loop and self._running.is_set():
-            self._loop.call_soon_threadsafe(self._loop.stop)
-            self._thread.join()
+            if self._custom_loop is not None:
+                # For custom loops, we don't stop the loop itself, just clear our running flag
+                self._running.clear()
+            else:
+                # For our own loop, stop it and join the thread
+                self._loop.call_soon_threadsafe(self._loop.stop)
+                if self._thread:
+                    self._thread.join()
             self.thread_pool.shutdown(wait=False)
 
     def register(
@@ -257,7 +276,12 @@ async def _execute_and_callback():
                 }
             callback(response)
 
-        asyncio.run_coroutine_threadsafe(_execute_and_callback(), self._loop)
+        if self._custom_loop is not None:
+            # For custom loops, schedule the task on the custom loop
+            self._loop.create_task(_execute_and_callback())
+        else:
+            # For our own loop running in a separate thread, use run_coroutine_threadsafe
+            asyncio.run_coroutine_threadsafe(_execute_and_callback(), self._loop)
 
 
 class ConversationInitiationData: