#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Runner session argument types for the development runner.
These types are used by the development runner to pass transport-specific
information to bot functions.
"""
import argparse
from dataclasses import dataclass, field
from typing import Any
from fastapi import WebSocket
from pydantic import BaseModel, ConfigDict, Field
[docs]
class DialinSettings(BaseModel):
"""Dial-in settings from the Daily webhook.
This model matches the structure sent by Pipecat Cloud and Daily.co webhooks
for incoming PSTN/SIP calls.
Parameters:
call_id: Unique identifier for the call (UUID representing sessionId in SIP Network)
call_domain: Daily domain for the call (UUID representing Daily Domain on SIP Network)
To: The dialed phone number (optional)
From: The caller's phone number (optional)
sip_headers: Optional SIP headers from the call
"""
call_id: str
call_domain: str
To: str | None = None
From: str | None = None
sip_headers: dict[str, str] | None = None
[docs]
class DailyDialinRequest(BaseModel):
"""Request data for Daily PSTN dial-in requests.
This is the structure passed in runner_args.body for dial-in calls.
It matches the payload structure from Pipecat Cloud's dial-in webhook handler.
Parameters:
dialin_settings: Dial-in configuration including call_id, call_domain, To, From
daily_api_key: Daily API key for pinlessCallUpdate (required for dial-in)
daily_api_url: Daily API URL (staging or production)
"""
dialin_settings: DialinSettings
daily_api_key: str
daily_api_url: str
[docs]
class CallData(BaseModel):
"""Parsed telephony handshake data from the provider's first WebSocket messages.
Populated by :func:`pipecat.runner.utils.parse_telephony_websocket` and exposed on
``WebSocketRunnerArguments.call_data`` by ``create_transport``. Gives typed
attribute access — ``call_data.to_number``, ``call_data.call_id`` — while staying
dict-compatible (``call_data["call_id"]``, ``call_data.get("body", {})``) so bots
written against the old dict keep working.
Fields are populated per provider; absent ones stay ``None``. Provider-specific keys
not modeled here remain accessible (``extra="allow"``).
This base holds the fields common to all providers. Provider-specific fields live
on subclasses (:class:`TelnyxCallData`, :class:`ExotelCallData`), which
``parse_telephony_websocket`` / ``create_transport`` construct per provider.
Parameters:
stream_id: Provider media-stream identifier.
call_id: Provider call identifier, normalized across providers (Twilio
``callSid``, Plivo ``callId``, Exotel ``call_sid``, Telnyx
``call_control_id``).
from_number: Caller's number. Wire key ``from``.
to_number: Dialed number. Wire key ``to``.
body: Custom parameters sent by the provider (e.g. Twilio TwiML stream
parameters).
"""
model_config = ConfigDict(populate_by_name=True, extra="allow")
stream_id: str | None = None
call_id: str | None = None
from_number: str | None = Field(default=None, alias="from")
to_number: str | None = Field(default=None, alias="to")
body: dict = Field(default_factory=dict)
def _wire_dict(self) -> dict:
"""The original provider dict shape (wire/alias keys, including extras)."""
return self.model_dump(by_alias=True)
def __getitem__(self, key: str):
"""Dict-style access, e.g. ``call_data["call_id"]``."""
return self._wire_dict()[key]
def __contains__(self, key: str) -> bool:
"""``"call_id" in call_data`` — True only when the provider set the value."""
wire = self._wire_dict()
return key in wire and wire[key] is not None
[docs]
def get(self, key: str, default: Any = None) -> Any:
"""Dict-style ``.get`` returning ``default`` when the key is missing or unset."""
value = self._wire_dict().get(key, default)
return default if value is None else value
[docs]
class TelnyxCallData(CallData):
"""Telnyx-specific parsed telephony handshake data.
Parameters:
outbound_encoding: Telnyx outbound media encoding.
"""
outbound_encoding: str | None = None
[docs]
class ExotelCallData(CallData):
"""Exotel-specific parsed telephony handshake data.
Parameters:
account_sid: Exotel account sid.
custom_parameters: Exotel custom parameters.
"""
account_sid: str | None = None
custom_parameters: str | dict | None = None
[docs]
@dataclass
class RunnerArguments:
"""Base class for runner session arguments.
Parameters:
handle_sigint: Whether the bot should install a SIGINT handler.
handle_sigterm: Whether the bot should install a SIGTERM handler.
pipeline_idle_timeout_secs: Seconds the pipeline may stay idle before
shutting down.
body: Optional request body data passed from the runner entry point.
call_data: Parsed telephony handshake as a :class:`CallData` model — typed
attribute access (``call_data.to_number``) that's also dict-compatible
(``call_data["call_id"]``). Populated by ``create_transport`` (or a direct
``parse_telephony_websocket`` call) for telephony connections; ``None``
otherwise. Lives on the base so any bot can read ``runner_args.call_data``
uniformly, mirroring ``body``.
session_id: Identifier for this bot session.
cli_args: Parsed CLI arguments from the runner, when launched via the
development runner.
"""
# Use kw_only so subclasses don't need to worry about ordering.
handle_sigint: bool = field(init=False, kw_only=True)
handle_sigterm: bool = field(init=False, kw_only=True)
pipeline_idle_timeout_secs: int = field(init=False, kw_only=True)
body: Any | None = field(default_factory=dict, kw_only=True)
call_data: CallData | None = field(default=None, kw_only=True)
session_id: str | None = field(default=None, kw_only=True)
cli_args: argparse.Namespace | None = field(default=None, init=False, kw_only=True)
def __post_init__(self):
self.handle_sigint = False
self.handle_sigterm = False
self.pipeline_idle_timeout_secs = 300
[docs]
@dataclass
class DailyRunnerArguments(RunnerArguments):
"""Daily transport session arguments for the runner.
Parameters:
room_url: Daily room URL to join
token: Authentication token for the room
body: Additional request data
"""
room_url: str
token: str | None = None
[docs]
@dataclass
class VonageRunnerArguments(RunnerArguments):
"""Vonage transport session arguments for the runner.
Parameters:
application_id: Vonage application ID
vonage_session_id: Vonage session ID
token: Vonage Session Token
"""
application_id: str
vonage_session_id: str
token: str
[docs]
@dataclass
class WebSocketRunnerArguments(RunnerArguments):
"""WebSocket transport session arguments for the runner.
The parsed telephony handshake is available on the inherited ``call_data`` field
(a :class:`CallData` model), populated by ``create_transport``.
Parameters:
websocket: WebSocket connection for audio streaming
transport_type: Transport type identifier. Set to ``"websocket"`` for plain
WebSocket connections; ``None`` triggers auto-detection from the first
telephony provider message. After auto-detection, ``create_transport``
overwrites this in place with the detected provider (e.g. ``"twilio"``).
body: Additional request data
"""
websocket: WebSocket
transport_type: str | None = None
[docs]
@dataclass
class SmallWebRTCRunnerArguments(RunnerArguments):
"""Small WebRTC transport session arguments for the runner.
Parameters:
webrtc_connection: Pre-configured WebRTC peer connection
"""
webrtc_connection: Any
[docs]
@dataclass
class LiveKitRunnerArguments(RunnerArguments):
"""LiveKit transport session arguments for the runner.
Parameters:
room_name: LiveKit room name to join
token: Authentication token for the room
body: Additional request data
"""
room_name: str
url: str
token: str