Use create_transport to simplify bot()

CHANGELOG.md

@@ -5,6 +5,21 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+
+- Generated bots now configure their transport through a single
+  `create_transport(runner_args, transport_params)` call instead of per-transport
+  `match`/`case` boilerplate, building on Pipecat's feature-complete
+  `create_transport`. Only Daily PSTN dial-out and Twilio+SIP keep a bespoke flow.
+  Telephony bots also ship **active** caller personalization (typed `CallData` /
+  `CallInfo` attribute access) matching the Pipecat examples.
+
+- Simplified the generated run instructions to just `uv run bot.py` for every
+  transport (the dev runner serves all transports and the caller selects one), with
+  a short ngrok/webhook note for telephony.
+
 ## [1.3.0] - 2026-05-29
 
 ### Added

pyproject.toml

@@ -53,7 +53,6 @@ dev = [
     "build~=1.4.0",
     # Pipecat with all service extras for auto-generating imports and configs
     "pipecat-ai[aic,anthropic,assemblyai,asyncai,aws,aws-nova-sonic,azure,camb,cartesia,cerebras,daily,deepseek,deepgram,elevenlabs,fal,fireworks,fish,gladia,google,gradium,groq,heygen,hume,inception,inworld,kokoro,lmnt,mistral,moondream,neuphonic,novita,nvidia,openai,openrouter,perplexity,piper,qwen,resembleai,rime,sagemaker,sambanova,sarvam,silero,simli,smallest,soniox,speechmatics,tavus,together,ultravox,webrtc,websocket,xai]==1.3.0",
-    "tomli>=2.0.0; python_version<'3.11'",  # TOML parsing for tests
     "setuptools~=78.1.1",
     "setuptools_scm~=8.3.1",
 ]

scripts/imports/import_generator.py

@@ -369,6 +369,9 @@ def _get_external_module_path(class_name: str) -> str | None:
 # a deprecated module. This mapping forces the correct module.
 _MODULE_OVERRIDES: dict[str, str] = {
     "LLMUserAggregatorParams": "pipecat.processors.aggregators.llm_response_universal",
+    # Explicit so regeneration doesn't depend on the searched source tree containing it
+    # — see the "create_transport" feature. The collapsed bot() imports create_transport.
+    "create_transport": "pipecat.runner.utils",
 }
 
 

src/pipecat_cli/config_validator.py

@@ -268,7 +268,7 @@ def validate_and_build_config(
         elif t == "twilio_daily_sip_dialout":
             resolved_twilio_daily_sip_mode = "dial-out"
 
-    return ProjectConfig(
+    config = ProjectConfig(
         project_name=name,
         bot_type=bot_type,
         transports=resolved_transports,
@@ -291,6 +291,7 @@ def validate_and_build_config(
         enable_krisp=enable_krisp,
         enable_observability=observability,
     )
+    return config
 
 
 def load_config_from_file(path: Path) -> dict:

src/pipecat_cli/generators/project.py

@@ -367,8 +367,7 @@ def _generate_readme(self, project_path: Path) -> None:
         # Get human-readable labels for all services
         all_transports = ServiceRegistry.WEBRTC_TRANSPORTS + ServiceRegistry.TELEPHONY_TRANSPORTS
 
-        # Get run commands and categorize transports
-        run_commands = self._get_run_commands()
+        # Categorize transports for the run instructions
         telephony_transports = {"twilio", "telnyx", "plivo", "exotel"}
         webrtc_transports = {"smallwebrtc", "daily"}
         has_telephony = any(t in telephony_transports for t in self.config.transports)
@@ -408,7 +407,6 @@ def _generate_readme(self, project_path: Path) -> None:
             "generate_client": self.config.generate_client,
             "client_framework": self.config.client_framework,
             "client_server": self.config.client_server,
-            "run_commands": run_commands,
             "has_telephony": has_telephony,
             "has_webrtc": has_webrtc,
             "daily_pstn_mode": self.config.daily_pstn_mode,
@@ -483,9 +481,6 @@ def print_next_steps(self, project_path: Path) -> None:
                 )
             return
 
-        # Determine run command based on transport
-        run_commands = self._get_run_commands()
-
         # Client setup
         if self.config.generate_client:
             console.print("\n  [bold]Client setup:[/bold]")
@@ -502,47 +497,20 @@ def print_next_steps(self, project_path: Path) -> None:
         console.print("  • Create .env file: [bold cyan]cp .env.example .env[/bold cyan]")
         console.print("  • [bold]Edit .env and add your API keys[/bold]")
 
-        # Categorize transports
+        # Every standard transport runs the same way: `uv run bot.py`. The runner
+        # serves all transports and the caller selects which one — a web/mobile client
+        # picks its transport when it connects, and a telephony provider connects to
+        # /ws. Telephony additionally needs a public tunnel so the provider can reach
+        # the bot.
         telephony_transports = {"twilio", "telnyx", "plivo", "exotel"}
-        webrtc_transports = {"smallwebrtc", "daily"}
-
         has_telephony = any(t in telephony_transports for t in self.config.transports)
-        has_webrtc = any(t in webrtc_transports for t in self.config.transports)
 
-        # Get categorized commands
-        webrtc_cmds = [cmd for cmd in run_commands if cmd["label"] in ["SmallWebRTC", "Daily"]]
-        telephony_cmds = [
-            cmd for cmd in run_commands if cmd["label"] in ["Twilio", "Telnyx", "Plivo", "Exotel"]
-        ]
-
-        if has_telephony and has_webrtc:
-            # Mixed: show both local and production workflows
-            console.print("  • Run your bot:\n")
-            console.print("     [bold]For local development:[/bold]")
-            for cmd in webrtc_cmds:
-                console.print(f"       • {cmd['label']}: [bold cyan]{cmd['command']}[/bold cyan]")
-            console.print("\n     [bold]For telephony deployment:[/bold]")
-            console.print("       • Run ngrok: [bold cyan]ngrok http 7860[/bold cyan]")
-            console.print("       • Run bot:")
-            for cmd in telephony_cmds:
-                console.print(f"         • {cmd['label']}: [bold cyan]{cmd['command']}[/bold cyan]")
-        elif has_telephony:
-            # Telephony only
-            console.print("  • Run ngrok tunnel: [bold cyan]ngrok http 7860[/bold cyan]")
-            console.print("  • Run your bot:")
-            for cmd in run_commands:
-                if cmd["label"]:
-                    console.print(f"     • {cmd['label']}: [bold cyan]{cmd['command']}[/bold cyan]")
-                else:
-                    console.print(f"     [bold cyan]{cmd['command']}[/bold cyan]")
-        else:
-            # WebRTC only
-            console.print("  • Run your bot:")
-            for cmd in run_commands:
-                if cmd["label"]:
-                    console.print(f"     • {cmd['label']}: [bold cyan]{cmd['command']}[/bold cyan]")
-                else:
-                    console.print(f"     [bold cyan]{cmd['command']}[/bold cyan]")
+        console.print("  • Run your bot: [bold cyan]uv run bot.py[/bold cyan]")
+        if has_telephony:
+            console.print(
+                "  • Expose it for telephony: [bold cyan]ngrok http 7860[/bold cyan], then point "
+                "your provider's webhook at [bold cyan]wss://<your-ngrok-host>/ws[/bold cyan]"
+            )
 
         # Add cloud deployment info if applicable
         if self.config.deploy_to_cloud:
@@ -552,29 +520,6 @@ def print_next_steps(self, project_path: Path) -> None:
         else:
             console.print("\n[dim]See README.md for detailed setup instructions.[/dim]\n")
 
-    def _get_run_commands(self) -> list[dict[str, str]]:
-        """Get transport-specific run commands with labels."""
-        commands = []
-
-        for transport in self.config.transports:
-            if transport == "smallwebrtc":
-                commands.append({"label": "SmallWebRTC", "command": "uv run bot.py"})
-            elif transport == "daily":
-                commands.append({"label": "Daily", "command": "uv run bot.py --transport daily"})
-            elif transport in {"twilio", "telnyx", "plivo", "exotel"}:
-                commands.append(
-                    {
-                        "label": transport.title(),
-                        "command": f"uv run bot.py --transport {transport} --proxy your_url.ngrok.io",
-                    }
-                )
-
-        # If no specific commands, default to basic run
-        if not commands:
-            commands.append({"label": "", "command": "uv run bot.py"})
-
-        return commands
-
     def _generate_client(self, client_path: Path) -> None:
         """Generate client application files."""
         # Determine which template to use

src/pipecat_cli/registry/_imports.py

@@ -13,28 +13,17 @@
 # Import statements mapping for services and transports
 IMPORTS = {
     # Transports - WebRTC
-    "daily": [
-        "from pipecat.runner.types import DailyRunnerArguments",
-        "from pipecat.transports.daily.transport import DailyTransport, DailyParams",
-    ],
-    "smallwebrtc": [
-        "from pipecat.runner.types import SmallWebRTCRunnerArguments",
-        "from pipecat.transports.base_transport import TransportParams",
-        "from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection",
-        "from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport",
-    ],
+    "daily": ["from pipecat.transports.daily.transport import DailyParams"],
+    "smallwebrtc": ["from pipecat.transports.base_transport import TransportParams"],
     "websocket": [
-        "from pipecat.runner.types import WebSocketRunnerArguments",
         "from pipecat.serializers.protobuf import ProtobufFrameSerializer",
-        "from pipecat.transports.websocket.fastapi import FastAPIWebsocketTransport, FastAPIWebsocketParams",
+        "from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams",
     ],
     # Transports - Telephony
     "twilio": [
-        "from pipecat.runner.types import WebSocketRunnerArguments",
-        "from pipecat.serializers.twilio import TwilioFrameSerializer",
-        "from pipecat.transports.websocket.fastapi import FastAPIWebsocketTransport, FastAPIWebsocketParams",
+        "from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams",
         "import aiohttp",
-        "from pipecat.runner.utils import parse_telephony_websocket",
+        "from pydantic import BaseModel",
     ],
     "twilio_daily_sip_dialin": [
         "from pipecat.transports.daily.transport import DailyParams, DailyTransport",
@@ -44,35 +33,20 @@
     "twilio_daily_sip_dialout": [
         "from pipecat.transports.daily.transport import DailyParams, DailyTransport",
         "from server_utils import AgentRequest, DialoutSettings",
-        "from typing import Any, Optional",
+        "from typing import Any",
     ],
     "daily_pstn_dialin": [
-        "from pipecat.transports.daily.transport import DailyParams, DailyDialinSettings, DailyTransport",
+        "from pipecat.transports.daily.transport import DailyParams",
         "from pipecat.runner.types import DailyDialinRequest",
     ],
     "daily_pstn_dialout": [
         "from pipecat.transports.daily.transport import DailyParams, DailyTransport",
         "from server_utils import AgentRequest, DialoutSettings",
-        "from typing import Any, Optional",
-    ],
-    "telnyx": [
-        "from pipecat.runner.types import WebSocketRunnerArguments",
-        "from pipecat.serializers.telnyx import TelnyxFrameSerializer",
-        "from pipecat.transports.websocket.fastapi import FastAPIWebsocketTransport, FastAPIWebsocketParams",
-        "from pipecat.runner.utils import parse_telephony_websocket",
-    ],
-    "plivo": [
-        "from pipecat.runner.types import WebSocketRunnerArguments",
-        "from pipecat.serializers.plivo import PlivoFrameSerializer",
-        "from pipecat.transports.websocket.fastapi import FastAPIWebsocketTransport, FastAPIWebsocketParams",
-        "from pipecat.runner.utils import parse_telephony_websocket",
-    ],
-    "exotel": [
-        "from pipecat.runner.types import WebSocketRunnerArguments",
-        "from pipecat.serializers.exotel import ExotelFrameSerializer",
-        "from pipecat.transports.websocket.fastapi import FastAPIWebsocketTransport, FastAPIWebsocketParams",
-        "from pipecat.runner.utils import parse_telephony_websocket",
+        "from typing import Any",
     ],
+    "telnyx": ["from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams"],
+    "plivo": ["from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams"],
+    "exotel": ["from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams"],
     # STT Services
     "assemblyai_stt": ["from pipecat.services.assemblyai.stt import AssemblyAISTTService"],
     "aws_transcribe_stt": ["from pipecat.services.aws.stt import AWSTranscribeSTTService"],
@@ -229,17 +203,18 @@
     ],
     "runner": [
         "from dotenv import load_dotenv",
-        "from pipecat.frames.frames import LLMRunFrame",
         "from pipecat.runner.types import RunnerArguments",
         "from pipecat.transports.base_transport import BaseTransport",
     ],
+    "llm_run_frame": ["from pipecat.frames.frames import LLMRunFrame"],
     "observability": [
         "from pipecat_tail.observer import TailObserver",
         "from pipecat_whisker import WhiskerObserver",
     ],
     "external_turn_strategies": [
         "from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies"
     ],
+    "create_transport": ["from pipecat.runner.utils import create_transport"],
 }
 
 # Base imports always included in generated bot files

src/pipecat_cli/registry/service_loader.py

@@ -269,6 +269,25 @@ def get_imports_for_services(
         if features.get("observability"):
             imports.update(ServiceRegistry.FEATURE_IMPORTS["observability"])
 
+        # Most bots build transports via create_transport, so import it whenever the
+        # bot uses that collapsed path. Only dial-out and SIP keep a bespoke flow that
+        # constructs the transport by hand; Daily PSTN dial-in is collapsed and goes
+        # through create_transport like the rest.
+        _bespoke_transport = {
+            "daily_pstn_dialout",
+            "twilio_daily_sip_dialin",
+            "twilio_daily_sip_dialout",
+        }
+        transport_values = set(transport_list) if "transports" in services else set()
+        if transport_values and not (transport_values & _bespoke_transport):
+            imports.update(ServiceRegistry.FEATURE_IMPORTS["create_transport"])
+
+        # LLMRunFrame kicks off the conversation on connect. Dial-out bots wait for the
+        # callee instead, so they neither queue nor import it.
+        _dialout_transports = {"daily_pstn_dialout", "twilio_daily_sip_dialout"}
+        if not (transport_values & _dialout_transports):
+            imports.update(ServiceRegistry.FEATURE_IMPORTS["llm_run_frame"])
+
         # Some STT services perform their own end-of-turn detection
         stt_value = services.get("stt", "")
         if ServiceLoader.uses_external_turn_detection(stt_value):