Add support for sending contextual_update events (elevenlabs#553)

Author: IvanLuchkin

src/elevenlabs/conversational_ai/conversation.py

@@ -6,7 +6,7 @@
 import asyncio
 from concurrent.futures import ThreadPoolExecutor
 
-from websockets.sync.client import connect
+from websockets.sync.client import connect, ClientConnection
 from websockets.exceptions import ConnectionClosedOK
 
 from ..base_client import BaseElevenLabs
@@ -240,6 +240,7 @@ def __init__(
         self.client_tools.start()
 
         self._thread = None
+        self._ws: Optional[ClientConnection] = None
         self._should_stop = threading.Event()
         self._conversation_id = None
         self._last_interrupt_id = 0
@@ -257,6 +258,7 @@ def end_session(self):
         """Ends the conversation session and cleans up resources."""
         self.audio_interface.stop()
         self.client_tools.stop()
+        self._ws = None
         self._should_stop.set()
 
     def wait_for_session_end(self) -> Optional[str]:
@@ -283,6 +285,7 @@ def _run(self, ws_url: str):
                     }
                 )
             )
+            self._ws = ws
 
             def input_callback(audio):
                 try:
@@ -369,6 +372,16 @@ def send_response(response):
         else:
             pass  # Ignore all other message types.
 
+    def send_contextual_update(self, text: str):
+        if not self._ws:
+            raise RuntimeError("WebSocket is not connected")
+
+        payload = {
+                "type": "contextual_update",
+                "text": text,
+        }
+        self._ws.send(json.dumps(payload))
+
     def _get_wss_url(self):
         base_ws_url = self.client._client_wrapper.get_environment().wss
         return f"{base_ws_url}/v1/convai/conversation?agent_id={self.agent_id}"

tests/test_convai.py

@@ -163,3 +163,33 @@ def test_conversation_with_dynamic_variables():
     mock_ws.send.assert_any_call(json.dumps(expected_init_message))
     agent_response_callback.assert_called_once_with("Hello there!")
     assert conversation._conversation_id == TEST_CONVERSATION_ID
+
+def test_conversation_with_contextual_update():
+    # Mock setup
+    mock_ws = create_mock_websocket([])
+    mock_client = MagicMock()
+
+    # Setup the conversation
+    conversation = Conversation(
+        client=mock_client,
+        agent_id=TEST_AGENT_ID,
+        requires_auth=False,
+        audio_interface=MockAudioInterface(),
+    )
+
+    # Run the test
+    with patch("elevenlabs.conversational_ai.conversation.connect") as mock_connect:
+        mock_connect.return_value.__enter__.return_value = mock_ws
+
+        conversation.start_session()
+        time.sleep(0.1)
+
+        conversation.send_contextual_update("User appears to be looking at pricing page")
+
+        # Teardown
+        conversation.end_session()
+        conversation.wait_for_session_end()
+
+    # Assertions
+    expected = json.dumps({"type": "contextual_update", "text": "User appears to be looking at pricing page"})
+    mock_ws.send.assert_any_call(expected)