From b825dd779e2d64e1ac2fa98744bf7b9be8c35568 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 19 May 2026 17:19:22 -0400 Subject: [PATCH] Clarify runner startup banner --- src/pipecat/runner/run.py | 137 +++++++++++++++++++++++--------------- tests/test_runner_run.py | 84 ++++++++++++++++++++++- 2 files changed, 163 insertions(+), 58 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 189bd3efd..5ff0f351e 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -137,6 +137,12 @@ TRANSPORT_ROUTE_DEPENDENCIES = { "webrtc": ("aiortc",), "websocket": ("fastapi", "websockets"), } +TRANSPORT_INSTALL_HINTS = { + "daily": "install pipecat-ai[daily]", + "webrtc": "install pipecat-ai[webrtc]", + "telephony": "install pipecat-ai[websocket]", + "websocket": "install pipecat-ai[websocket]", +} # Mirror Pipecat Cloud's 4-hour max session limit so dev rooms get cleaned up. PIPECAT_ROOM_EXP_HOURS = 4.0 @@ -203,6 +209,80 @@ def _transport_routes_enabled(transport: str) -> bool: return all(_is_module_available(module) for module in _transport_route_dependencies(transport)) +def _runner_url(args: argparse.Namespace) -> str: + """Return the browser URL for the runner prebuilt client.""" + return f"http://{args.host}:{args.port}" + + +def _transport_status_lists() -> tuple[list[str], list[str]]: + """Return enabled and disabled transport labels for the startup banner.""" + transports = ["daily", "webrtc", "telephony", "websocket"] + enabled = [] + disabled = [] + + for label in transports: + transport = TELEPHONY_TRANSPORTS[0] if label == "telephony" else label + if _transport_routes_enabled(transport): + enabled.append(label) + else: + disabled.append(f"{label} ({TRANSPORT_INSTALL_HINTS[label]})") + + return enabled, disabled + + +def _format_transport_status(labels: list[str]) -> str: + """Format a startup banner transport status list.""" + return ", ".join(labels) if labels else "none" + + +def _print_startup_message(args: argparse.Namespace): + """Print connection information for the development runner.""" + print() + if args.transport is None: + enabled, disabled = _transport_status_lists() + print("🚀 Bot ready!") + print(f" → Open: {_runner_url(args)}") + print(f" → Enabled transports: {_format_transport_status(enabled)}") + if disabled: + print(f" → Disabled transports: {_format_transport_status(disabled)}") + elif args.transport == "webrtc": + if args.esp32: + print("🚀 Bot ready! (ESP32 mode)") + elif args.whatsapp: + print("🚀 Bot ready! (WhatsApp)") + else: + print("🚀 Bot ready! (WebRTC)") + if _transport_routes_enabled("webrtc"): + print(f" → Open: {_runner_url(args)}") + else: + print(f" → WebRTC disabled ({TRANSPORT_INSTALL_HINTS['webrtc']})") + elif args.transport == "daily": + print("🚀 Bot ready! (Daily)") + if not _transport_routes_enabled("daily"): + print(f" → Daily disabled ({TRANSPORT_INSTALL_HINTS['daily']})") + else: + print(f" → Open: {_runner_url(args)}") + if args.dialin: + print( + f" → Daily dial-in webhook: " + f"http://{args.host}:{args.port}/daily-dialin-webhook" + ) + print(" → Configure this URL in your Daily phone number settings") + elif args.transport in TELEPHONY_TRANSPORTS: + print(f"🚀 Bot ready! ({args.transport.capitalize()})") + if not _transport_routes_enabled(args.transport): + print(f" → Telephony disabled ({TRANSPORT_INSTALL_HINTS['telephony']})") + else: + print(f" → Open: {_runner_url(args)}") + if args.proxy: + print(f" → XML webhook: http://{args.host}:{args.port}/") + print(f" → WebSocket: ws://{args.host}:{args.port}/ws") + elif args.transport == "vonage": + print() + print("🚀 Bot ready!") + print() + + def _get_bot_module(): """Get the bot module from the calling script.""" import importlib.util @@ -1226,64 +1306,11 @@ def main(parser: argparse.ArgumentParser | None = None): return # Print startup message - print() - if args.transport is None: - print("🚀 Bot ready!") - if _transport_routes_enabled("webrtc"): - print(f" → WebRTC: http://{args.host}:{args.port}/client") - else: - print(" → WebRTC: disabled (install pipecat-ai[webrtc])") - if _transport_routes_enabled("daily"): - print(f" → Daily: http://{args.host}:{args.port}/daily") - else: - print(" → Daily: disabled (install pipecat-ai[daily])") - if _transport_routes_enabled("twilio"): - print(f" → Telephony: ws://{args.host}:{args.port}/ws") - else: - print(" → Telephony: disabled (install pipecat-ai[websocket])") - if _transport_routes_enabled("websocket"): - print(f" → WebSocket: ws://{args.host}:{args.port}/ws-client") - else: - print(" → WebSocket: disabled (install pipecat-ai[websocket])") - elif args.transport == "webrtc": - if args.esp32: - print("🚀 Bot ready! (ESP32 mode)") - elif args.whatsapp: - print("🚀 Bot ready! (WhatsApp)") - else: - print("🚀 Bot ready! (WebRTC)") - if _transport_routes_enabled("webrtc"): - print(f" → Open http://{args.host}:{args.port}/client in your browser") - else: - print(" → WebRTC disabled (install pipecat-ai[webrtc])") - elif args.transport == "daily": - print("🚀 Bot ready! (Daily)") - if not _transport_routes_enabled("daily"): - print(" → Daily disabled (install pipecat-ai[daily])") - elif args.dialin: - print( - f" → Daily dial-in webhook: http://{args.host}:{args.port}/daily-dialin-webhook" - ) - print(f" → Configure this URL in your Daily phone number settings") - else: - print( - f" → Open http://{args.host}:{args.port}/daily in your browser to start a session" - ) - elif args.transport in TELEPHONY_TRANSPORTS: - print(f"🚀 Bot ready! ({args.transport.capitalize()})") - if not _transport_routes_enabled(args.transport): - print(" → Telephony disabled (install pipecat-ai[websocket])") - elif args.proxy: - print(f" → XML webhook: http://{args.host}:{args.port}/") - if _transport_routes_enabled(args.transport): - print(f" → WebSocket: ws://{args.host}:{args.port}/ws") - elif args.transport == "vonage": - print() - print(f"🚀 Bot ready!") + _print_startup_message(args) + if args.transport == "vonage": asyncio.run(_run_vonage()) print() return - print() RUNNER_DOWNLOADS_FOLDER = args.folder RUNNER_HOST = args.host diff --git a/tests/test_runner_run.py b/tests/test_runner_run.py index b4fc52117..6b0dc3e3d 100644 --- a/tests/test_runner_run.py +++ b/tests/test_runner_run.py @@ -5,9 +5,11 @@ # import argparse +import io import sys import types import unittest +from contextlib import redirect_stdout from unittest.mock import MagicMock, patch from fastapi import FastAPI @@ -15,6 +17,7 @@ from fastapi.testclient import TestClient from pydantic import BaseModel from pipecat.runner.run import ( + _print_startup_message, _setup_daily_routes, _setup_telephony_routes, _setup_unified_start_route, @@ -26,6 +29,12 @@ from pipecat.runner.run import ( class TestRunnerRun(unittest.TestCase): + def _capture_startup_message(self, args: argparse.Namespace) -> str: + buffer = io.StringIO() + with redirect_stdout(buffer): + _print_startup_message(args) + return buffer.getvalue() + def test_transport_route_dependencies_maps_transports_to_modules(self): self.assertEqual(_transport_route_dependencies("daily"), ("daily",)) self.assertEqual(_transport_route_dependencies("webrtc"), ("aiortc",)) @@ -70,9 +79,7 @@ class TestRunnerRun(unittest.TestCase): connection_module = types.ModuleType("pipecat.transports.smallwebrtc.connection") connection_module.SmallWebRTCConnection = MagicMock() - request_handler_module = types.ModuleType( - "pipecat.transports.smallwebrtc.request_handler" - ) + request_handler_module = types.ModuleType("pipecat.transports.smallwebrtc.request_handler") class IceCandidate(BaseModel): candidate: str @@ -239,6 +246,77 @@ class TestRunnerRun(unittest.TestCase): ), ) + def test_startup_message_all_transports_shows_open_url_and_transport_status(self): + args = argparse.Namespace(transport=None, host="localhost", port=7860) + + def routes_enabled(transport: str) -> bool: + return transport in {"twilio", "websocket"} + + with patch("pipecat.runner.run._transport_routes_enabled", side_effect=routes_enabled): + output = self._capture_startup_message(args) + + self.assertEqual( + output, + ( + "\n" + "🚀 Bot ready!\n" + " → Open: http://localhost:7860\n" + " → Enabled transports: telephony, websocket\n" + " → Disabled transports: daily (install pipecat-ai[daily]), " + "webrtc (install pipecat-ai[webrtc])\n" + "\n" + ), + ) + + def test_startup_message_all_transports_omits_disabled_status_when_all_enabled(self): + args = argparse.Namespace(transport=None, host="localhost", port=7860) + + with patch("pipecat.runner.run._transport_routes_enabled", return_value=True): + output = self._capture_startup_message(args) + + self.assertEqual( + output, + ( + "\n" + "🚀 Bot ready!\n" + " → Open: http://localhost:7860\n" + " → Enabled transports: daily, webrtc, telephony, websocket\n" + "\n" + ), + ) + + def test_startup_message_webrtc_uses_root_open_url(self): + args = argparse.Namespace( + transport="webrtc", host="localhost", port=7860, esp32=False, whatsapp=False + ) + + with patch("pipecat.runner.run._transport_routes_enabled", return_value=True): + output = self._capture_startup_message(args) + + self.assertIn(" → Open: http://localhost:7860\n", output) + self.assertNotIn("/client", output) + + def test_startup_message_daily_uses_root_open_url(self): + args = argparse.Namespace(transport="daily", host="localhost", port=7860, dialin=False) + + with patch("pipecat.runner.run._transport_routes_enabled", return_value=True): + output = self._capture_startup_message(args) + + self.assertIn(" → Open: http://localhost:7860\n", output) + self.assertNotIn("/daily in your browser", output) + + def test_startup_message_telephony_keeps_provider_endpoint_details(self): + args = argparse.Namespace( + transport="twilio", host="localhost", port=7860, proxy="example.ngrok.io" + ) + + with patch("pipecat.runner.run._transport_routes_enabled", return_value=True): + output = self._capture_startup_message(args) + + self.assertIn(" → Open: http://localhost:7860\n", output) + self.assertIn(" → XML webhook: http://localhost:7860/\n", output) + self.assertIn(" → WebSocket: ws://localhost:7860/ws\n", output) + if __name__ == "__main__": unittest.main()