bug fixes and QoL improvements

Author: PaulAsjes

src/elevenlabs/client.py

@@ -73,7 +73,7 @@ class _AsyncSpeechEngineAccessor:
     def __init__(self, client_wrapper: typing.Any) -> None:
         self._client_wrapper = client_wrapper
 
-    async def get(self, engine_id: str) -> "SpeechEngineResource":
+    async def get(self, engine_id: str) -> typing.Any:
         from .speech_engine.resource import SpeechEngineResource  # noqa: E402
 
         return SpeechEngineResource(

src/elevenlabs/speech_engine/server.py

@@ -1,15 +1,12 @@
 """SpeechEngineServer — standalone WebSocket server for Speech Engine."""
 
 import asyncio
-import logging
 import os
 import typing
 
-from .session import SpeechEngineSession, _wire_handlers
+from .session import SpeechEngineSession, _make_log, _wire_handlers
 from .types import WebSocketLike
 
-logger = logging.getLogger("elevenlabs.speech_engine")
-
 
 class SpeechEngineServer:
     """Standalone WebSocket server that produces :class:`SpeechEngineSession`
@@ -44,17 +41,9 @@ def __init__(
         self._api_key = api_key
         self._debug = debug
         self._handlers = handlers
-        self._stop_event = asyncio.Event()
+        self._stop_event = None  # type: typing.Optional[asyncio.Event]
         self._server = None  # type: typing.Any
-
-        if debug:
-            logger.setLevel(logging.DEBUG)
-            if not logger.handlers:
-                handler = logging.StreamHandler()
-                handler.setFormatter(
-                    logging.Formatter("[SpeechEngine] %(message)s")
-                )
-                logger.addHandler(handler)
+        self._log = _make_log(debug)
 
     def handle_connection(self, ws: WebSocketLike) -> SpeechEngineSession:
         """Wrap *ws* in a :class:`SpeechEngineSession` with the server's
@@ -64,7 +53,7 @@ def handle_connection(self, ws: WebSocketLike) -> SpeechEngineSession:
         individual connections.  The returned session's :meth:`run` must
         still be awaited by the caller.
         """
-        logger.debug("creating new session")
+        self._log("creating new session")
         session = SpeechEngineSession(ws, debug=self._debug)
         _wire_handlers(session, self._handlers)
         return session
@@ -83,6 +72,8 @@ async def serve(self) -> None:
                 "environment variable."
             )
 
+        self._stop_event = asyncio.Event()
+
         async def _handler(websocket: typing.Any, *_args: typing.Any) -> None:
             if self._path is not None and websocket.request.path != self._path:
                 await websocket.close(4000, "not found")
@@ -92,7 +83,7 @@ async def _handler(websocket: typing.Any, *_args: typing.Any) -> None:
                 "x-elevenlabs-speech-engine-authorization"
             )
             if not header_value:
-                logger.debug(
+                self._log(
                     "rejected connection — missing "
                     "X-Elevenlabs-Speech-Engine-Authorization header"
                 )
@@ -104,11 +95,11 @@ async def _handler(websocket: typing.Any, *_args: typing.Any) -> None:
             try:
                 verify_speech_engine_jwt(header_value, api_key)
             except ValueError as e:
-                logger.debug("rejected connection — %s", e)
+                self._log("rejected connection — %s", e)
                 await websocket.close(4001, str(e))
                 return
 
-            logger.debug("verified connection, accepting WebSocket")
+            self._log("verified connection, accepting WebSocket")
             session = self.handle_connection(websocket)
             await session.run()
 
@@ -117,7 +108,7 @@ async def _handler(websocket: typing.Any, *_args: typing.Any) -> None:
             "",
             self._port,
         )
-        logger.debug("speech engine server listening on port %d", self._port)
+        self._log("speech engine server listening on port %d", self._port)
         try:
             await self._stop_event.wait()
         finally:
@@ -126,4 +117,5 @@ async def _handler(websocket: typing.Any, *_args: typing.Any) -> None:
 
     async def stop(self) -> None:
         """Signal the server to shut down gracefully."""
-        self._stop_event.set()
+        if self._stop_event is not None:
+            self._stop_event.set()

src/elevenlabs/speech_engine/session.py

@@ -9,6 +9,17 @@
 
 logger = logging.getLogger("elevenlabs.speech_engine")
 
+
+def _make_log(
+    debug: bool,
+) -> typing.Callable[..., None]:
+    """Return a per-instance log function, mirroring the JS SDK pattern."""
+    if debug:
+        def _log(msg: str, *args: typing.Any) -> None:
+            print("[SpeechEngine]", msg % args if args else msg)
+        return _log
+    return lambda *_args, **_kw: None
+
 Callback = typing.Callable[..., typing.Any]
 
 
@@ -170,15 +181,7 @@ def __init__(
         self._closed = False
         self._event_handlers = {}  # type: typing.Dict[str, typing.List[Callback]]
         self._once_handlers = {}  # type: typing.Dict[str, typing.List[Callback]]
-
-        if debug:
-            logger.setLevel(logging.DEBUG)
-            if not logger.handlers:
-                handler = logging.StreamHandler()
-                handler.setFormatter(
-                    logging.Formatter("[SpeechEngine] %(message)s")
-                )
-                logger.addHandler(handler)
+        self._log = _make_log(debug)
 
     # ------------------------------------------------------------------
     # Event emitter interface
@@ -238,7 +241,7 @@ async def run(self) -> None:
                 except asyncio.CancelledError:
                     raise
                 except Exception:
-                    logger.debug("WebSocket connection lost")
+                    self._log("WebSocket connection lost")
                     break
 
                 try:
@@ -290,15 +293,15 @@ async def send_response(
             return
 
         if isinstance(response, str):
-            logger.debug(
+            self._log(
                 'sending string response: "%s", event_id=%s',
                 response,
                 self._current_event_id,
             )
             await self._send_agent_response(response, False)
             await self._send_agent_response("", True)
         else:
-            logger.debug(
+            self._log(
                 "starting streamed response, event_id=%s",
                 self._current_event_id,
             )
@@ -325,7 +328,7 @@ async def _handle_message(self, msg: typing.Dict[str, typing.Any]) -> None:
 
         if msg_type == "init":
             self._conversation_id = msg.get("conversation_id")
-            logger.debug(
+            self._log(
                 "session initialized, conversation_id=%s",
                 self._conversation_id,
             )
@@ -339,7 +342,7 @@ async def _handle_message(self, msg: typing.Dict[str, typing.Any]) -> None:
                 and self._current_task is not None
                 and not self._current_task.done()
             ):
-                logger.debug(
+                self._log(
                     "skipping duplicate transcript, event_id=%s",
                     incoming_event_id,
                 )
@@ -351,7 +354,7 @@ async def _handle_message(self, msg: typing.Dict[str, typing.Any]) -> None:
             )
             await self._cancel_current_and_wait()
             if was_active:
-                logger.debug(
+                self._log(
                     "interrupted: cancelling previous response "
                     "(event_id=%s) for new transcript (event_id=%s)",
                     self._current_event_id,
@@ -360,16 +363,20 @@ async def _handle_message(self, msg: typing.Dict[str, typing.Any]) -> None:
 
             self._current_event_id = incoming_event_id
             transcript_data = msg.get("user_transcript", [])
-            logger.debug(
+            self._log(
                 "received transcript, event_id=%s, messages=%d",
                 self._current_event_id,
                 len(transcript_data),
             )
 
-            transcript = [
-                ConversationMessage(role=m["role"], content=m["content"])
-                for m in transcript_data
-            ]
+            try:
+                transcript = [
+                    ConversationMessage(role=m["role"], content=m["content"])
+                    for m in transcript_data
+                ]
+            except (KeyError, TypeError) as e:
+                await self._emit("error", e)
+                return
 
             handlers = list(
                 self._event_handlers.get("user_transcript", [])
@@ -419,7 +426,7 @@ async def _stream_response(self, stream: typing.Any) -> None:
         try:
             async for chunk in stream:
                 if self._closed:
-                    logger.debug(
+                    self._log(
                         "stream stopped: session closed after %d chunks, "
                         "event_id=%s",
                         chunks,
@@ -431,7 +438,7 @@ async def _stream_response(self, stream: typing.Any) -> None:
                     chunks += 1
                     await self._send_agent_response(text, False, event_id)
             if not self._closed:
-                logger.debug(
+                self._log(
                     "stream complete: %d chunks sent, event_id=%s",
                     chunks,
                     event_id,

tests/test_speech_engine_resource.py

@@ -18,7 +18,7 @@
 
 class MockWebSocket:
     def __init__(self) -> None:
-        self._inbox = asyncio.Queue()  # type: asyncio.Queue[str]
+        self._inbox = asyncio.Queue()  # type: asyncio.Queue[typing.Any]
         self.sent = []  # type: typing.List[str]
         self.closed = False
 

tests/test_speech_engine_session.py

@@ -26,7 +26,7 @@ class MockWebSocket:
     """In-memory WebSocket stand-in backed by an asyncio.Queue."""
 
     def __init__(self) -> None:
-        self._inbox = asyncio.Queue()  # type: asyncio.Queue[str]
+        self._inbox = asyncio.Queue()  # type: asyncio.Queue[typing.Any]
         self.sent = []  # type: typing.List[str]
         self.closed = False