Merge pull request #2891 from pipecat-ai/aleix/daily-pipecat-runner-args

runner: allow starting a bot from Daily's /start endpoint

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- It is now possible to start a bot from the `/start` endpoint when using the
+  runner Daily's transport. This follows the Pipecat Cloud format with
+  `createDailyRoom` and `body` fields in the POST request body.
+
 - Added an ellipsis character (`…`) to the end of sentence detection in the
   string utils.
 
@@ -88,6 +92,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `RunnerArguments` now include the `body` field, so there's no need to add it
+  to subclasses. Also, all `RunnerArguments` fields are now keyword-only.
+
 - `CartesiaSTTService` now inherits from `WebsocketSTTService`.
 
 - Package upgrades:

src/pipecat/runner/daily.py

@@ -76,6 +76,7 @@ class DailyRoomConfig(BaseModel):
 async def configure(
     aiohttp_session: aiohttp.ClientSession,
     *,
+    api_key: Optional[str] = None,
     room_exp_duration: Optional[float] = 2.0,
     token_exp_duration: Optional[float] = 2.0,
     sip_caller_phone: Optional[str] = None,
@@ -92,6 +93,7 @@ async def configure(
 
     Args:
         aiohttp_session: HTTP session for making API requests.
+        api_key: Daily API key.
         room_exp_duration: Room expiration time in hours.
         token_exp_duration: Token expiration time in hours.
         sip_caller_phone: Phone number or identifier for SIP display name.
@@ -129,7 +131,7 @@ async def configure(
         config = await configure(session, room_properties=custom_props)
     """
     # Check for required API key
-    api_key = os.getenv("DAILY_API_KEY")
+    api_key = api_key or os.getenv("DAILY_API_KEY")
     if not api_key:
         raise Exception(
             "DAILY_API_KEY environment variable is required. "

src/pipecat/runner/run.py

@@ -82,6 +82,7 @@ from loguru import logger
 
 from pipecat.runner.types import (
     DailyRunnerArguments,
+    RunnerArguments,
     SmallWebRTCRunnerArguments,
     WebSocketRunnerArguments,
 )
@@ -529,9 +530,9 @@ def _setup_daily_routes(app: FastAPI):
     """Set up Daily-specific routes."""
 
     @app.get("/")
-    async def start_agent():
+    async def create_room_and_start_agent():
         """Launch a Daily bot and redirect to room."""
-        print("Starting bot with Daily transport")
+        print("Starting bot with Daily transport and redirecting to Daily room")
 
         import aiohttp
 
@@ -546,11 +547,11 @@ def _setup_daily_routes(app: FastAPI):
             asyncio.create_task(bot_module.bot(runner_args))
             return RedirectResponse(room_url)
 
-    async def _handle_rtvi_request(request: Request):
+    @app.post("/start")
-        """Common handler for both /start and /connect endpoints.
+    async def start_agent(request: Request):
+        """Handler for /start endpoints.
 
         Expects POST body like::
-
             {
                 "createDailyRoom": true,
                 "dailyRoomProperties": { "start_video_off": true },
@@ -567,45 +568,32 @@ def _setup_daily_routes(app: FastAPI):
             logger.error(f"Failed to parse request body: {e}")
             request_data = {}
 
-        # Extract the body data that should be passed to the bot
+        create_daily_room = request_data.get("createDailyRoom", False)
-        # This mimics Pipecat Cloud's behavior
+        body = request_data.get("body", {})
-        bot_body = request_data.get("body", {})
 
-        # Log the extracted body data for debugging
+        bot_module = _get_bot_module()
-        if bot_body:
+
-            logger.info(f"Extracted body data for bot: {bot_body}")
+        result = None
+        if create_daily_room:
+            import aiohttp
+
+            from pipecat.runner.daily import configure
+
+            async with aiohttp.ClientSession() as session:
+                room_url, token = await configure(session)
+                runner_args = DailyRunnerArguments(room_url=room_url, token=token, body=body)
+                result = {
+                    "dailyRoom": room_url,
+                    "dailyToken": token,
+                    "sessionId": str(uuid.uuid4()),
+                }
         else:
-            logger.debug("No body data provided in request")
+            runner_args = RunnerArguments(body=body)
 
-        from pipecat.runner.daily import configure
+        # Start the bot in the background
+        asyncio.create_task(bot_module.bot(runner_args))
 
-        async with aiohttp.ClientSession() as session:
+        return result
-            room_url, token = await configure(session)
-
-            # Start the bot in the background with extracted body data
-            bot_module = _get_bot_module()
-            runner_args = DailyRunnerArguments(room_url=room_url, token=token, body=bot_body)
-            asyncio.create_task(bot_module.bot(runner_args))
-            # Match PCC /start endpoint response format:
-            return {"dailyRoom": room_url, "dailyToken": token}
-
-    @app.post("/start")
-    async def rtvi_start(request: Request):
-        """Launch a Daily bot and return connection info for RTVI clients."""
-        return await _handle_rtvi_request(request)
-
-    @app.post("/connect")
-    async def rtvi_connect(request: Request):
-        """Launch a Daily bot and return connection info for RTVI clients.
-
-        .. deprecated:: 0.0.78
-            Use /start instead. This endpoint will be removed in a future version.
-        """
-        logger.warning(
-            "DEPRECATED: /connect endpoint is deprecated. Please use /start instead. "
-            "This endpoint will be removed in a future version."
-        )
-        return await _handle_rtvi_request(request)
 
 
 def _setup_telephony_routes(app: FastAPI, *, transport_type: str, proxy: str):

src/pipecat/runner/types.py

@@ -20,9 +20,11 @@ from fastapi import WebSocket
 class RunnerArguments:
     """Base class for runner session arguments."""
 
-    handle_sigint: bool = field(init=False)
+    # Use kw_only so subclasses don't need to worry about ordering.
-    handle_sigterm: bool = field(init=False)
+    handle_sigint: bool = field(init=False, kw_only=True)
-    pipeline_idle_timeout_secs: int = field(init=False)
+    handle_sigterm: bool = field(init=False, kw_only=True)
+    pipeline_idle_timeout_secs: int = field(init=False, kw_only=True)
+    body: Optional[Any] = field(default_factory=dict, kw_only=True)
 
     def __post_init__(self):
         self.handle_sigint = False
@@ -42,7 +44,6 @@ class DailyRunnerArguments(RunnerArguments):
 
     room_url: str
     token: Optional[str] = None
-    body: Optional[Any] = field(default_factory=dict)
 
 
 @dataclass
@@ -55,7 +56,6 @@ class WebSocketRunnerArguments(RunnerArguments):
     """
 
     websocket: WebSocket
-    body: Optional[Any] = field(default_factory=dict)
 
 
 @dataclass