services: only read from TTS websocket if websocket connection established

src/pipecat/services/cartesia/tts.py

@@ -185,7 +185,8 @@ class CartesiaTTSService(AudioContextWordTTSService):
 
     async def _connect(self):
         await self._connect_websocket()
-        if not self._receive_task:
+
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):

src/pipecat/services/elevenlabs/tts.py

@@ -308,10 +308,10 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
-        if not self._keepalive_task:
+        if self._websocket and not self._keepalive_task:
             self._keepalive_task = self.create_task(self._keepalive_task_handler())
 
     async def _disconnect(self):

src/pipecat/services/fish/tts.py

@@ -104,7 +104,8 @@ class FishAudioTTSService(InterruptibleTTSService):
 
     async def _connect(self):
         await self._connect_websocket()
-        if not self._receive_task:
+
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):

src/pipecat/services/gladia/stt.py

@@ -285,7 +285,7 @@ class GladiaSTTService(STTService):
         settings = self._prepare_settings()
         response = await self._setup_gladia(settings)
         self._websocket = await websockets.connect(response["url"])
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler())
 
     async def stop(self, frame: EndFrame):

src/pipecat/services/lmnt/tts.py

@@ -109,7 +109,7 @@ class LmntTTSService(InterruptibleTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):

src/pipecat/services/neuphonic/tts.py

@@ -106,6 +106,9 @@ class NeuphonicTTSService(InterruptibleTTSService):
         self._started = False
         self._cumulative_time = 0
 
+        self._receive_task = None
+        self._keepalive_task = None
+
     def can_generate_metrics(self) -> bool:
         return True
 
@@ -159,8 +162,11 @@ class NeuphonicTTSService(InterruptibleTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+        if self._websocket and not self._receive_task:
-        self._keepalive_task = self.create_task(self._keepalive_task_handler())
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+        if self._websocket and not self._keepalive_task:
+            self._keepalive_task = self.create_task(self._keepalive_task_handler())
 
     async def _disconnect(self):
         if self._receive_task:
@@ -193,7 +199,6 @@ class NeuphonicTTSService(InterruptibleTTSService):
             url = f"{self._url}/speak/{self._settings['lang_code']}?{'&'.join(query_params)}"
 
             self._websocket = await websockets.connect(url)
-
         except Exception as e:
             logger.error(f"{self} initialization error: {e}")
             self._websocket = None

src/pipecat/services/playht/tts.py

@@ -157,7 +157,7 @@ class PlayHTTTSService(InterruptibleTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):

src/pipecat/services/rime/tts.py

@@ -168,7 +168,7 @@ class RimeTTSService(AudioContextWordTTSService):
         """Establish websocket connection and start receive task."""
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):