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

Use shared telephony route availability checks

Browse Source
This commit is contained in:
Mark Backman
2026-05-20 08:57:48 -04:00
parent 861dd23873
commit d11a4ba0cd
2 changed files with 7 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/pipecat/runner/run.py
View File

@@ -135,6 +135,7 @@ TELEPHONY_TRANSPORTS = ["twilio", "telnyx", "plivo", "exotel"]
TRANSPORT_ROUTE_DEPENDENCIES = {
"daily": ("daily",),
"webrtc": ("aiortc",),
"telephony": ("fastapi", "websockets"),
"websocket": ("fastapi", "websockets"),
}
TRANSPORT_INSTALL_HINTS = {
@@ -193,7 +194,7 @@ def _transport_route_dependencies(transport: str) -> tuple[str, ...]:
Module names required to enable the transport route.
"""
if transport in TELEPHONY_TRANSPORTS:
return TRANSPORT_ROUTE_DEPENDENCIES["websocket"]
return TRANSPORT_ROUTE_DEPENDENCIES["telephony"]
return TRANSPORT_ROUTE_DEPENDENCIES.get(transport, ())
@@ -221,8 +222,7 @@ def _transport_status_lists() -> tuple[list[str], list[str]]:
disabled = []
for label in transports:
transport = TELEPHONY_TRANSPORTS[0] if label == "telephony" else label
if _transport_routes_enabled(transport):
if _transport_routes_enabled(label):
enabled.append(label)
else:
disabled.append(f"{label} ({TRANSPORT_INSTALL_HINTS[label]})")
@@ -1051,7 +1051,7 @@ def _setup_telephony_routes(app: FastAPI, args: argparse.Namespace):
specific telephony transport is chosen via ``-t`` because the XML template
is provider-specific and requires a proxy hostname (``--proxy``).
"""
if not _transport_routes_enabled("twilio"):
if not _transport_routes_enabled("telephony"):
return
if args.transport in TELEPHONY_TRANSPORTS:

4
tests/test_runner_run.py
View File

@@ -39,6 +39,7 @@ class TestRunnerRun(unittest.TestCase):
self.assertEqual(_transport_route_dependencies("daily"), ("daily",))
self.assertEqual(_transport_route_dependencies("webrtc"), ("aiortc",))
self.assertEqual(_transport_route_dependencies("websocket"), ("fastapi", "websockets"))
self.assertEqual(_transport_route_dependencies("telephony"), ("fastapi", "websockets"))
self.assertEqual(_transport_route_dependencies("twilio"), ("fastapi", "websockets"))
self.assertEqual(_transport_route_dependencies("telnyx"), ("fastapi", "websockets"))
self.assertEqual(_transport_route_dependencies("plivo"), ("fastapi", "websockets"))
@@ -53,6 +54,7 @@ class TestRunnerRun(unittest.TestCase):
self.assertFalse(_transport_routes_enabled("daily"))
self.assertFalse(_transport_routes_enabled("webrtc"))
self.assertTrue(_transport_routes_enabled("websocket"))
self.assertTrue(_transport_routes_enabled("telephony"))
self.assertTrue(_transport_routes_enabled("twilio"))
self.assertTrue(_transport_routes_enabled("vonage"))
@@ -250,7 +252,7 @@ class TestRunnerRun(unittest.TestCase):
args = argparse.Namespace(transport=None, host="localhost", port=7860)
def routes_enabled(transport: str) -> bool:
return transport in {"twilio", "websocket"}
return transport in {"telephony", "websocket"}
with patch("pipecat.runner.run._transport_routes_enabled", side_effect=routes_enabled):
output = self._capture_startup_message(args)