Use RunnerArguments in examples

examples/runner-examples/01-all-transport-bot.py

@@ -28,17 +28,19 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
 from pipecat.runner.types import (
     DailyRunnerArguments,
+    RunnerArguments,
     SmallWebRTCRunnerArguments,
     WebSocketRunnerArguments,
 )
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport
 
 load_dotenv(override=True)
 
 
-async def run_bot(transport):
+async def run_bot(transport: BaseTransport):
     """Main bot logic that works with any transport."""
     logger.info(f"Starting bot")
 
@@ -100,9 +102,7 @@ async def run_bot(transport):
     await runner.run(task)
 
 
-async def bot(
-    runner_args: DailyRunnerArguments | SmallWebRTCRunnerArguments | WebSocketRunnerArguments,
-):
+async def bot(runner_args: RunnerArguments):
     """Main bot entry point compatible with Pipecat Cloud."""
 
     if isinstance(runner_args, DailyRunnerArguments):

examples/runner-examples/01-all-transport-factory-bot.py

@@ -26,16 +26,12 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
-from pipecat.runner.types import (
-    DailyRunnerArguments,
-    SmallWebRTCRunnerArguments,
-    WebSocketRunnerArguments,
-)
+from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
 from pipecat.transports.services.daily import DailyParams
 
@@ -74,7 +70,7 @@ transport_params = {
 }
 
 
-async def run_bot(transport):
+async def run_bot(transport: BaseTransport):
     """Main bot logic that works with any transport."""
     logger.info("Starting bot")
 
@@ -136,9 +132,7 @@ async def run_bot(transport):
     await runner.run(task)
 
 
-async def bot(
-    runner_args: DailyRunnerArguments | SmallWebRTCRunnerArguments | WebSocketRunnerArguments,
-):
+async def bot(runner_args: RunnerArguments):
     """Main bot entry point compatible with Pipecat Cloud."""
     transport = await create_transport(runner_args, transport_params)
     await run_bot(transport)

examples/runner-examples/02-two-transport-bot.py

@@ -19,18 +19,16 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
-from pipecat.runner.types import (
-    DailyRunnerArguments,
-    SmallWebRTCRunnerArguments,
-)
+from pipecat.runner.types import DailyRunnerArguments, RunnerArguments, SmallWebRTCRunnerArguments
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport
 
 load_dotenv(override=True)
 
 
-async def run_bot(transport):
+async def run_bot(transport: BaseTransport):
     """Main bot logic that works with any transport."""
     logger.info(f"Starting bot")
 
@@ -92,7 +90,7 @@ async def run_bot(transport):
     await runner.run(task)
 
 
-async def bot(runner_args: DailyRunnerArguments | SmallWebRTCRunnerArguments):
+async def bot(runner_args: RunnerArguments):
     """Main bot entry point compatible with Pipecat Cloud."""
 
     if isinstance(runner_args, DailyRunnerArguments):

examples/runner-examples/03-single-transport-bot.py

@@ -19,17 +19,17 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
-from pipecat.runner.types import SmallWebRTCRunnerArguments
+from pipecat.runner.types import RunnerArguments
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
 
 load_dotenv(override=True)
 
 
-async def run_bot(transport):
+async def run_bot(transport: BaseTransport):
     """Main bot logic that works with any transport."""
     logger.info(f"Starting bot")
 
@@ -91,7 +91,7 @@ async def run_bot(transport):
     await runner.run(task)
 
 
-async def bot(runner_args: SmallWebRTCRunnerArguments):
+async def bot(runner_args: RunnerArguments):
     """Main bot entry point compatible with Pipecat Cloud."""
 
     transport = SmallWebRTCTransport(

src/pipecat/runner/run.py

@@ -21,7 +21,7 @@ are established.
 
 Single transport example::
 
-    async def bot(runner_args: DailyRunnerArguments):
+    async def bot(runner_args: RunnerArguments):
         transport = DailyTransport(
             runner_args.room_url,
             runner_args.token,
@@ -37,7 +37,7 @@ Single transport example::
 
 Multiple transport example::
 
-    async def bot(runner_args):
+    async def bot(runner_args: RunnerArguments):
         # Type-safe transport detection
         if isinstance(runner_args, DailyRunnerArguments):
             transport = setup_daily_transport(runner_args)  # Your application code