Clarify runner startup banner

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

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