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

Fix type errors in serializers and add to pyright checked set

Browse Source 
Moves src/pipecat/serializers into pyright's include list. Narrows
self._params to each subclass's InputParams in exotel, vonage, plivo,
twilio, genesys, and telnyx. In protobuf.py, renames the reassigned
frame local to avoid clobbering its Frame type and silences two dynamic
attribute accesses on the generated frames_pb2 module.

Also aligns telnyx and plivo hangup validation with twilio: if
auto_hang_up=True (the default) but required credentials are missing,
__init__ now raises ValueError instead of silently logging a warning
at call-end time. Previously a misconfigured serializer would construct
fine and fail to hang up the call later, leaving a phantom billable
session.
This commit is contained in:
Mark Backman
2026-04-21 18:12:54 -04:00
parent 7fc79511dd
commit df9642eb5a
8 changed files with 75 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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