wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: handle Optional websocket/client receivers across services

Browse Source 
Pyright flagged 19 sites where `await self._<connection>.send/recv/...`
was called on a receiver typed `X | None`. Each kind of call site
needed a slightly different fix to be both type-safe and behaviour-
preserving:

Streaming/user-facing paths (early return + warn — drop and warn is
the right runtime fail-safe when reconnect didn't succeed):

- cartesia/stt.py (run_stt)
- soniox/stt.py (_send_keepalive)
- elevenlabs/tts.py (run_tts — yields ErrorFrame and returns)
- deepgram/sagemaker/tts.py (run_tts)
- transports/lemonslice/transport.py (send_message)
- transports/tavus/transport.py (send_message)

"Should never happen" cases (early return with comment, no warn —
caller already gated on a separate `_is_*` check, so a warn would be
noise):

- deepgram/flux/stt.py (transport methods, gated by _transport_is_active)
- deepgram/flux/sagemaker/stt.py (same)
- stt_service.py (_send_keepalive, gated by _is_keepalive_ready)
- elevenlabs/stt.py (_send_keepalive, same)
- llm_service.py (_ws_recv — raises ConnectionError to match
  _ensure_connected's contract)
- heygen/client.py (receive loop, gated by self._connected)

Just-assigned-above (use a local variable so pyright keeps the
narrowing across statements):

- lmnt/tts.py
- gradium/stt.py
- fish/tts.py

Other:

- transports/websocket/server.py — used the existing local `websocket`
  parameter in scope instead of `self._websocket` for the close call.
- websocket_service.py — `send_with_retry` raises ConnectionError when
  `self._websocket` is None inside the existing try-block, so the
  broad `except Exception` triggers reconnect just as it would on a
  real send failure (preserving the prior behaviour where None
  silently fell through to the AttributeError-driven reconnect path).

Drops three now-clean files from the pyright ignore list: cartesia/stt.py,
elevenlabs/stt.py, and soniox/stt.py.
This commit is contained in:
Paul Kompfner
2026-04-28 11:57:54 -04:00
parent 1cd73b1ef8
commit bec407ce3a
18 changed files with 72 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pyrightconfig.json
View File

@@ -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",

5
src/pipecat/services/cartesia/stt.py
View File

@@ -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:

8
src/pipecat/services/deepgram/flux/sagemaker/stt.py
View File

@@ -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:

8
src/pipecat/services/deepgram/flux/stt.py
View File

@@ -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:

4
src/pipecat/services/deepgram/sagemaker/tts.py
View File

@@ -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

4
src/pipecat/services/elevenlabs/stt.py
View File

@@ -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",

5
src/pipecat/services/elevenlabs/tts.py
View File

@@ -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)

5
src/pipecat/services/fish/tts.py
View File

@@ -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")

7
src/pipecat/services/gradium/stt.py
View File

@@ -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']}")

2
src/pipecat/services/heygen/client.py
View File

@@ -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)

5
src/pipecat/services/llm_service.py
View File

@@ -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)

5
src/pipecat/services/lmnt/tts.py
View File

@@ -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:

3
src/pipecat/services/soniox/stt.py
View File

@@ -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)

4
src/pipecat/services/stt_service.py
View File

@@ -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):

7
src/pipecat/services/websocket_service.py
View File

@@ -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)

3
src/pipecat/transports/lemonslice/transport.py
View File

@@ -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

4
src/pipecat/transports/tavus/transport.py
View File

@@ -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):

2
src/pipecat/transports/websocket/server.py
View File

@@ -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")