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

Merge pull request #4349 from pipecat-ai/mb/serializer-pyright

Browse Source 
Fix type errors in serializers and add to pyright checked set
This commit is contained in:
Mark Backman
2026-04-22 09:43:31 -04:00
committed by GitHub
parent 75bd1b5b9b 263cad41f0
commit f425e946eb
9 changed files with 76 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4349.changed.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ `PlivoFrameSerializer` and `TelnyxFrameSerializer` now raise `ValueError` at construction when `auto_hang_up=True` (the default) but required credentials are missing, matching `TwilioFrameSerializer`. Previously they constructed successfully and the hangup failed silently at call-end, leaving phantom billable sessions on the provider. If you relied on the old silent behavior, pass `auto_hang_up=False` explicitly or provide the credentials. The specific fields checked are `call_id`/`auth_id`/`auth_token` for Plivo and `call_control_id`/`api_key` for Telnyx.

2
pyrightconfig.json
View File

@@ -11,6 +11,7 @@
"src/pipecat/observers",
"src/pipecat/pipeline",
"src/pipecat/runner",
"src/pipecat/serializers",
"src/pipecat/tests",
"src/pipecat/transcriptions",
"src/pipecat/turns",
@@ -21,7 +22,6 @@
"src/pipecat/adapters",
"src/pipecat/audio",
"src/pipecat/processors",
"src/pipecat/serializers",
"src/pipecat/services",
"src/pipecat/transports",
"tests"

4
src/pipecat/serializers/exotel.py
View File

@@ -59,7 +59,9 @@ class ExotelFrameSerializer(FrameSerializer):
call_sid: The associated Exotel Call SID (optional, not used in this implementation).
params: Configuration parameters.
"""
super().__init__(params or ExotelFrameSerializer.InputParams())
params = params or ExotelFrameSerializer.InputParams()
super().__init__(params)
self._params: ExotelFrameSerializer.InputParams = params
self._stream_sid = stream_sid
self._call_sid = call_sid

4
src/pipecat/serializers/genesys.py
View File

@@ -166,7 +166,9 @@ class GenesysAudioHookSerializer(FrameSerializer):
params: Configuration parameters.
**kwargs: Additional arguments passed to BaseObject (e.g., name).
"""
super().__init__(params or GenesysAudioHookSerializer.InputParams(), **kwargs)
params = params or GenesysAudioHookSerializer.InputParams()
super().__init__(params, **kwargs)
self._params: GenesysAudioHookSerializer.InputParams = params
self._genesys_sample_rate = self._params.genesys_sample_rate
self._sample_rate = 0 # Pipeline input rate, set in setup()

42
src/pipecat/serializers/plivo.py
View File

@@ -8,6 +8,7 @@
import base64
import json
from typing import cast
from loguru import logger
@@ -71,7 +72,24 @@ class PlivoFrameSerializer(FrameSerializer):
auth_token: Plivo auth token (required for auto hang-up).
params: Configuration parameters.
"""
super().__init__(params or PlivoFrameSerializer.InputParams())
params = params or PlivoFrameSerializer.InputParams()
super().__init__(params)
self._params: PlivoFrameSerializer.InputParams = params
# Validate hangup-related parameters if auto_hang_up is enabled
if self._params.auto_hang_up:
missing_credentials = []
if not call_id:
missing_credentials.append("call_id")
if not auth_id:
missing_credentials.append("auth_id")
if not auth_token:
missing_credentials.append("auth_token")
if missing_credentials:
raise ValueError(
f"auto_hang_up is enabled but missing required parameters: {', '.join(missing_credentials)}"
)
self._stream_id = stream_id
self._call_id = call_id
@@ -152,23 +170,11 @@ class PlivoFrameSerializer(FrameSerializer):
try:
import aiohttp
auth_id = self._auth_id
auth_token = self._auth_token
call_id = self._call_id
if not call_id or not auth_id or not auth_token:
missing = []
if not call_id:
missing.append("call_id")
if not auth_id:
missing.append("auth_id")
if not auth_token:
missing.append("auth_token")
logger.warning(
f"Cannot hang up Plivo call: missing required parameters: {', '.join(missing)}"
)
return
# __init__ guarantees these are non-None whenever auto_hang_up is True,
# which is the only path that reaches this method.
auth_id = cast(str, self._auth_id)
auth_token = cast(str, self._auth_token)
call_id = cast(str, self._call_id)
# Plivo API endpoint for hanging up calls
endpoint = f"https://api.plivo.com/v1/Account/{auth_id}/Call/{call_id}/"

17
src/pipecat/serializers/protobuf.py
View File

@@ -83,23 +83,24 @@ class ProtobufFrameSerializer(FrameSerializer):
Serialized frame as bytes, or None if frame type is not serializable.
"""
# Wrapping this messages as a JSONFrame to send
serializable: Frame | MessageFrame = frame
if isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
if self.should_ignore_frame(frame):
return None
frame = MessageFrame(
serializable = MessageFrame(
data=json.dumps(frame.message),
)
proto_frame = frame_protos.Frame()
if type(frame) not in self.SERIALIZABLE_TYPES:
logger.warning(f"Frame type {type(frame)} is not serializable")
proto_frame = frame_protos.Frame() # type: ignore[attr-defined]
if type(serializable) not in self.SERIALIZABLE_TYPES:
logger.warning(f"Frame type {type(serializable)} is not serializable")
return None
# ignoring linter errors; we check that type(frame) is in this dict above
proto_optional_name = self.SERIALIZABLE_TYPES[type(frame)] # type: ignore
proto_optional_name = self.SERIALIZABLE_TYPES[type(serializable)] # type: ignore
proto_attr = getattr(proto_frame, proto_optional_name)
for field in dataclasses.fields(frame): # type: ignore
value = getattr(frame, field.name)
for field in dataclasses.fields(serializable): # type: ignore
value = getattr(serializable, field.name)
if value and hasattr(proto_attr, field.name):
setattr(proto_attr, field.name, value)
@@ -114,7 +115,7 @@ class ProtobufFrameSerializer(FrameSerializer):
Returns:
Deserialized frame instance, or None if deserialization fails.
"""
proto = frame_protos.Frame.FromString(data)
proto = frame_protos.Frame.FromString(data) # type: ignore[attr-defined]
which = proto.WhichOneof("frame")
if which not in self.DESERIALIZABLE_FIELDS:
logger.error("Unable to deserialize a valid frame")

34
src/pipecat/serializers/telnyx.py
View File

@@ -8,10 +8,10 @@
import base64
import json
from typing import cast
import aiohttp
from loguru import logger
from pydantic import BaseModel
from pipecat.audio.dtmf.types import KeypadEntry
from pipecat.audio.utils import (
@@ -46,7 +46,7 @@ class TelnyxFrameSerializer(FrameSerializer):
credentials to be provided.
"""
class InputParams(BaseModel):
class InputParams(FrameSerializer.InputParams):
"""Configuration parameters for TelnyxFrameSerializer.
Parameters:
@@ -82,10 +82,26 @@ class TelnyxFrameSerializer(FrameSerializer):
api_key: Your Telnyx API key (required for auto hang-up).
params: Configuration parameters.
"""
params = params or TelnyxFrameSerializer.InputParams()
super().__init__(params)
self._params: TelnyxFrameSerializer.InputParams = params
# Validate hangup-related parameters if auto_hang_up is enabled
if self._params.auto_hang_up:
missing_credentials = []
if not call_control_id:
missing_credentials.append("call_control_id")
if not api_key:
missing_credentials.append("api_key")
if missing_credentials:
raise ValueError(
f"auto_hang_up is enabled but missing required parameters: {', '.join(missing_credentials)}"
)
self._stream_id = stream_id
self._call_control_id = call_control_id
self._api_key = api_key
self._params = params or TelnyxFrameSerializer.InputParams()
self._params.outbound_encoding = outbound_encoding
self._params.inbound_encoding = inbound_encoding
@@ -163,14 +179,10 @@ class TelnyxFrameSerializer(FrameSerializer):
async def _hang_up_call(self):
"""Hang up the Telnyx call using Telnyx's REST API."""
try:
call_control_id = self._call_control_id
api_key = self._api_key
if not call_control_id or not api_key:
logger.warning(
"Cannot hang up Telnyx call: call_control_id and api_key must be provided"
)
return
# __init__ guarantees these are non-None whenever auto_hang_up is True,
# which is the only path that reaches this method.
call_control_id = cast(str, self._call_control_id)
api_key = cast(str, self._api_key)
# Telnyx API endpoint for hanging up a call
endpoint = f"https://api.telnyx.com/v2/calls/{call_control_id}/actions/hangup"

13
src/pipecat/serializers/twilio.py
View File

@@ -8,6 +8,7 @@
import base64
import json
from typing import cast
from loguru import logger
@@ -75,7 +76,9 @@ class TwilioFrameSerializer(FrameSerializer):
edge: Twilio edge location (e.g., "sydney", "dublin"). Must be specified with region.
params: Configuration parameters.
"""
super().__init__(params or TwilioFrameSerializer.InputParams())
params = params or TwilioFrameSerializer.InputParams()
super().__init__(params)
self._params: TwilioFrameSerializer.InputParams = params
# Validate hangup-related parameters if auto_hang_up is enabled
if self._params.auto_hang_up:
@@ -178,9 +181,11 @@ class TwilioFrameSerializer(FrameSerializer):
try:
import aiohttp
account_sid = self._account_sid
auth_token = self._auth_token
call_sid = self._call_sid
# __init__ guarantees these are non-None whenever auto_hang_up is True,
# which is the only path that reaches this method.
account_sid = cast(str, self._account_sid)
auth_token = cast(str, self._auth_token)
call_sid = cast(str, self._call_sid)
region = self._region
edge = self._edge

4
src/pipecat/serializers/vonage.py
View File

@@ -54,7 +54,9 @@ class VonageFrameSerializer(FrameSerializer):
Args:
params: Configuration parameters.
"""
super().__init__(params or VonageFrameSerializer.InputParams())
params = params or VonageFrameSerializer.InputParams()
super().__init__(params)
self._params: VonageFrameSerializer.InputParams = params
self._vonage_sample_rate = self._params.vonage_sample_rate
self._sample_rate = 0 # Pipeline input rate