diff --git a/pyrightconfig.json b/pyrightconfig.json index f77cf5e8c..9370d4387 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -40,14 +40,12 @@ "src/pipecat/services/aws/utils.py", "src/pipecat/services/azure/stt.py", "src/pipecat/services/azure/tts.py", - "src/pipecat/services/cartesia/stt.py", "src/pipecat/services/deepgram/flux/base.py", "src/pipecat/services/deepgram/flux/sagemaker/stt.py", "src/pipecat/services/deepgram/flux/stt.py", "src/pipecat/services/deepgram/sagemaker/stt.py", "src/pipecat/services/deepgram/sagemaker/tts.py", "src/pipecat/services/deepgram/tts.py", - "src/pipecat/services/elevenlabs/stt.py", "src/pipecat/services/elevenlabs/tts.py", "src/pipecat/services/fish/tts.py", "src/pipecat/services/gladia/stt.py", @@ -92,7 +90,6 @@ "src/pipecat/services/sarvam/tts.py", "src/pipecat/services/simli/video.py", "src/pipecat/services/smallest/tts.py", - "src/pipecat/services/soniox/stt.py", "src/pipecat/services/speechmatics/stt.py", "src/pipecat/services/stt_service.py", "src/pipecat/services/tavus/video.py", diff --git a/src/pipecat/services/cartesia/stt.py b/src/pipecat/services/cartesia/stt.py index 018d95e6a..af3418edd 100644 --- a/src/pipecat/services/cartesia/stt.py +++ b/src/pipecat/services/cartesia/stt.py @@ -290,6 +290,11 @@ class CartesiaSTTService(WebsocketSTTService): if not self._websocket or self._websocket.state is not State.OPEN: await self._connect() + if self._websocket is None: + logger.warning(f"{self}: websocket unavailable after reconnect, dropping audio") + yield None + return + try: await self._websocket.send(audio) except Exception as e: diff --git a/src/pipecat/services/deepgram/flux/sagemaker/stt.py b/src/pipecat/services/deepgram/flux/sagemaker/stt.py index 1b110cf1e..0f7a6fded 100644 --- a/src/pipecat/services/deepgram/flux/sagemaker/stt.py +++ b/src/pipecat/services/deepgram/flux/sagemaker/stt.py @@ -145,9 +145,17 @@ class DeepgramFluxSageMakerSTTService(DeepgramFluxSTTBase): # ------------------------------------------------------------------ async def _transport_send_audio(self, audio: bytes): + if ( + self._client is None + ): # should never happen — caller should gate on _transport_is_active() + return await self._client.send_audio_chunk(audio) async def _transport_send_json(self, message: dict): + if ( + self._client is None + ): # should never happen — caller should gate on _transport_is_active() + return await self._client.send_json(message) def _transport_is_active(self) -> bool: diff --git a/src/pipecat/services/deepgram/flux/stt.py b/src/pipecat/services/deepgram/flux/stt.py index df14442eb..98cf17ec2 100644 --- a/src/pipecat/services/deepgram/flux/stt.py +++ b/src/pipecat/services/deepgram/flux/stt.py @@ -240,9 +240,17 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService): # ------------------------------------------------------------------ async def _transport_send_audio(self, audio: bytes): + if ( + self._websocket is None + ): # should never happen — caller should gate on _transport_is_active() + return await self._websocket.send(audio) async def _transport_send_json(self, message: dict): + if ( + self._websocket is None + ): # should never happen — caller should gate on _transport_is_active() + return await self._websocket.send(json.dumps(message)) def _transport_is_active(self) -> bool: diff --git a/src/pipecat/services/deepgram/sagemaker/tts.py b/src/pipecat/services/deepgram/sagemaker/tts.py index 171da06cf..b0c24e117 100644 --- a/src/pipecat/services/deepgram/sagemaker/tts.py +++ b/src/pipecat/services/deepgram/sagemaker/tts.py @@ -337,6 +337,10 @@ class DeepgramSageMakerTTSService(TTSService): the response processor). """ logger.debug(f"{self}: Generating TTS [{text}]") + if self._client is None: + logger.warning(f"{self}: client unavailable, skipping TTS") + yield ErrorFrame(error="client unavailable") + return try: await self._client.send_json({"type": "Speak", "text": text}) yield None diff --git a/src/pipecat/services/elevenlabs/stt.py b/src/pipecat/services/elevenlabs/stt.py index 7f65fd90b..5571f8700 100644 --- a/src/pipecat/services/elevenlabs/stt.py +++ b/src/pipecat/services/elevenlabs/stt.py @@ -739,6 +739,10 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): Args: silence: Silent 16-bit mono PCM audio bytes. """ + if ( + self._websocket is None + ): # should never happen — caller should gate on _is_keepalive_ready() + return audio_base64 = base64.b64encode(silence).decode("utf-8") message = { "message_type": "input_audio_chunk", diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 88254c0c3..4da94643c 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -905,6 +905,11 @@ class ElevenLabsTTSService(WebsocketTTSService): if not self._websocket or self._websocket.state is State.CLOSED: await self._connect() + if self._websocket is None: + logger.warning(f"{self}: websocket unavailable after reconnect, skipping TTS") + yield ErrorFrame(error="websocket unavailable") + return + try: if not self.audio_context_available(context_id): await self.create_audio_context(context_id) diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index 83512da45..327e650ec 100644 --- a/src/pipecat/services/fish/tts.py +++ b/src/pipecat/services/fish/tts.py @@ -281,7 +281,8 @@ class FishAudioTTSService(InterruptibleTTSService): model = assert_given(self._settings.model) if model is not None: headers["model"] = model - self._websocket = await websocket_connect(self._base_url, additional_headers=headers) + websocket = await websocket_connect(self._base_url, additional_headers=headers) + self._websocket = websocket # Send initial start message with ormsgpack request_settings = { @@ -300,7 +301,7 @@ class FishAudioTTSService(InterruptibleTTSService): if self._settings.top_p is not None: request_settings["top_p"] = self._settings.top_p start_message = {"event": "start", "request": {"text": "", **request_settings}} - await self._websocket.send(ormsgpack.packb(start_message)) + await websocket.send(ormsgpack.packb(start_message)) logger.debug("Sent start message to Fish Audio") await self._call_event_handler("on_connected") diff --git a/src/pipecat/services/gradium/stt.py b/src/pipecat/services/gradium/stt.py index 0b6a81cdd..6f1a48f60 100644 --- a/src/pipecat/services/gradium/stt.py +++ b/src/pipecat/services/gradium/stt.py @@ -383,10 +383,11 @@ class GradiumSTTService(WebsocketSTTService): "x-api-key": self._api_key, "x-api-source": "pipecat", } - self._websocket = await websocket_connect( + websocket = await websocket_connect( ws_url, additional_headers=headers, ) + self._websocket = websocket await self._call_event_handler("on_connected") setup_msg = { "type": "setup", @@ -406,8 +407,8 @@ class GradiumSTTService(WebsocketSTTService): json_config["delay_in_frames"] = self._settings.delay_in_frames if json_config: setup_msg["json_config"] = json_config - await self._websocket.send(json.dumps(setup_msg)) - ready_msg = await self._websocket.recv() + await websocket.send(json.dumps(setup_msg)) + ready_msg = await websocket.recv() ready_msg = json.loads(ready_msg) if ready_msg["type"] == "error": raise Exception(f"received error {ready_msg['message']}") diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py index a02d515a7..80e5adb3f 100644 --- a/src/pipecat/services/heygen/client.py +++ b/src/pipecat/services/heygen/client.py @@ -291,6 +291,8 @@ class HeyGenClient: """Handle incoming WebSocket messages.""" try: while self._connected: + if self._websocket is None: # should never happen while _connected is True + break try: message = await self._websocket.recv() parsed_message = json.loads(message) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 77c1412a5..95bf2c762 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -1265,6 +1265,11 @@ class WebsocketLLMService(LLMService[TAdapter], WebsocketService, Generic[TAdapt Returns: The parsed JSON message as a dict. """ + # Should never happen — `_ensure_connected` (which callers must invoke + # first) raises ConnectionError if it can't establish a websocket. + # Match that contract here. + if self._websocket is None: + raise ConnectionError(f"{self} _ws_recv called without a websocket") try: raw = await self._websocket.recv() return json.loads(raw) diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py index b098b43c2..81dffac7e 100644 --- a/src/pipecat/services/lmnt/tts.py +++ b/src/pipecat/services/lmnt/tts.py @@ -267,10 +267,11 @@ class LmntTTSService(InterruptibleTTSService): } # Connect to LMNT's websocket directly - self._websocket = await websocket_connect("wss://api.lmnt.com/v1/ai/speech/stream") + websocket = await websocket_connect("wss://api.lmnt.com/v1/ai/speech/stream") + self._websocket = websocket # Send initialization message - await self._websocket.send(json.dumps(init_msg)) + await websocket.send(json.dumps(init_msg)) await self._call_event_handler("on_connected") except Exception as e: diff --git a/src/pipecat/services/soniox/stt.py b/src/pipecat/services/soniox/stt.py index 1732b3f91..ca99e69c2 100644 --- a/src/pipecat/services/soniox/stt.py +++ b/src/pipecat/services/soniox/stt.py @@ -648,4 +648,7 @@ class SonioxSTTService(WebsocketSTTService): Args: silence: Silent PCM audio bytes (unused, Soniox uses a protocol message). """ + if self._websocket is None: + logger.warning(f"{self}: websocket unavailable, skipping keepalive") + return await self._websocket.send(KEEPALIVE_MESSAGE) diff --git a/src/pipecat/services/stt_service.py b/src/pipecat/services/stt_service.py index b72547e1a..7c6028ea1 100644 --- a/src/pipecat/services/stt_service.py +++ b/src/pipecat/services/stt_service.py @@ -859,6 +859,10 @@ class WebsocketSTTService(STTService, WebsocketService): Args: silence: Silent 16-bit mono PCM audio bytes. """ + if ( + self._websocket is None + ): # should never happen — caller should gate on _is_keepalive_ready() + return await self._websocket.send(silence) async def _report_error(self, error: ErrorFrame): diff --git a/src/pipecat/services/websocket_service.py b/src/pipecat/services/websocket_service.py index 83ddb3746..2e6040ead 100644 --- a/src/pipecat/services/websocket_service.py +++ b/src/pipecat/services/websocket_service.py @@ -120,12 +120,17 @@ class WebsocketService(ABC): async def send_with_retry(self, message, report_error: Callable[[ErrorFrame], Awaitable[None]]): """Attempt to send a message, retrying after reconnect if necessary.""" try: + # If websocket isn't connected/present, treat as a send failure — + # the broad `except Exception` below will trigger a reconnect + # attempt. + if self._websocket is None: + raise ConnectionError(f"{self} no websocket connected") await self._websocket.send(message) except Exception as e: logger.error(f"{self} send failed: {e}, will try to reconnect") # Try to reconnect before retrying success = await self._try_reconnect(report_error=report_error) - if success: + if success and self._websocket is not None: logger.info(f"{self} reconnected successfully, will retry send the message") # trying to send the message one more time await self._websocket.send(message) diff --git a/src/pipecat/transports/lemonslice/transport.py b/src/pipecat/transports/lemonslice/transport.py index 90966ade2..83d8c0e9e 100644 --- a/src/pipecat/transports/lemonslice/transport.py +++ b/src/pipecat/transports/lemonslice/transport.py @@ -312,6 +312,9 @@ class LemonSliceTransportClient: Args: frame: The message frame to send. """ + if self._daily_transport_client is None: + return + await self._daily_transport_client.send_message(frame) @property diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index 9ad28979d..698d0dc95 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -360,6 +360,9 @@ class TavusTransportClient: Args: frame: The message frame to send. """ + if self._client is None: + return + await self._client.send_message(frame) @property @@ -416,6 +419,7 @@ class TavusTransportClient: """ if not self._client: return False + return await self._client.write_audio_frame(frame) async def register_audio_destination(self, destination: str, auto_silence: bool | None = True): diff --git a/src/pipecat/transports/websocket/server.py b/src/pipecat/transports/websocket/server.py index 82b899f71..0a78f7504 100644 --- a/src/pipecat/transports/websocket/server.py +++ b/src/pipecat/transports/websocket/server.py @@ -226,7 +226,7 @@ class WebsocketServerInputTransport(BaseInputTransport): # Notify disconnection await self._callbacks.on_client_disconnected(websocket) - await self._websocket.close() + await websocket.close() self._websocket = None logger.info(f"Client {websocket.remote_address} disconnected")