Merge branch 'main' into m-ods/assemblyai-universal-streaming

src/pipecat/__init__.py

@@ -4,10 +4,26 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import asyncio
+import sys
 from importlib.metadata import version
 
 from loguru import logger
 
 __version__ = version("pipecat-ai")
 
-logger.info(f"ᓚᘏᗢ Pipecat {__version__} ᓚᘏᗢ")
+event_loop = "asyncio"
+
+if sys.platform in ("linux", "darwin"):
+    try:
+        import asyncio
+
+        import uvloop
+
+        event_loop = f"uvloop {uvloop.__version__}"
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+    except ImportError:
+        logger.debug(f"Couldn't find `uvloop`")
+        pass
+
+logger.info(f"ᓚᘏᗢ Pipecat {__version__} ({event_loop}; Python {sys.version}) ᓚᘏᗢ")

src/pipecat/audio/vad/vad_analyzer.py

@@ -71,7 +71,7 @@ class VADAnalyzer(ABC):
         self.set_params(self._params)
 
     def set_params(self, params: VADParams):
-        logger.info(f"Setting VAD params to: {params}")
+        logger.debug(f"Setting VAD params to: {params}")
         self._params = params
         self._vad_frames = self.num_frames_required()
         self._vad_frames_num_bytes = self._vad_frames * self._num_channels * 2

src/pipecat/examples/daily_runner.py

@@ -0,0 +1,64 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+from typing import Optional
+
+import aiohttp
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    (url, token, _) = await configure_with_args(aiohttp_session)
+    return (url, token)
+
+
+async def configure_with_args(
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
+):
+    if not parser:
+        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+        )
+
+    if not key:
+        raise Exception(
+            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
+        )
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token, args)

src/pipecat/examples/run.py

@@ -0,0 +1,263 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+from contextlib import asynccontextmanager
+from typing import Any, Callable, Dict, Mapping, Optional
+
+import aiohttp
+import uvicorn
+from dotenv import load_dotenv
+from fastapi import BackgroundTasks, FastAPI, WebSocket
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import HTMLResponse, RedirectResponse
+from loguru import logger
+
+from pipecat.serializers.twilio import TwilioFrameSerializer
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import (
+    FastAPIWebsocketParams,
+    FastAPIWebsocketTransport,
+)
+from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
+from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+# Load environment variables
+load_dotenv(override=True)
+
+
+def get_transport_client_id(transport: BaseTransport, client: Any) -> str:
+    if isinstance(transport, SmallWebRTCTransport):
+        return client.pc_id
+    elif isinstance(transport, DailyTransport):
+        return client["id"]
+    logger.warning(f"Unable to get client id from unsupported transport {type(transport)}")
+    return ""
+
+
+async def maybe_capture_participant_camera(
+    transport: BaseTransport, client: Any, framerate: int = 0
+):
+    if isinstance(transport, DailyTransport):
+        await transport.capture_participant_video(
+            client["id"], framerate=framerate, video_source="camera"
+        )
+
+
+async def maybe_capture_participant_screen(
+    transport: BaseTransport, client: Any, framerate: int = 0
+):
+    if isinstance(transport, DailyTransport):
+        await transport.capture_participant_video(
+            client["id"], framerate=framerate, video_source="screenVideo"
+        )
+
+
+def run_example_daily(
+    run_example: Callable,
+    args: argparse.Namespace,
+    params: DailyParams,
+):
+    logger.info("Running example with DailyTransport...")
+
+    from pipecat.examples.daily_runner import configure
+
+    async def run():
+        async with aiohttp.ClientSession() as session:
+            (room_url, token) = await configure(session)
+
+            # Run example function with DailyTransport transport arguments.
+            transport = DailyTransport(room_url, token, "Pipecat", params=params)
+            await run_example(transport, args, True)
+
+    asyncio.run(run())
+
+
+def run_example_webrtc(
+    run_example: Callable,
+    args: argparse.Namespace,
+    params: TransportParams,
+):
+    logger.info("Running example with SmallWebRTCTransport...")
+
+    from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
+
+    app = FastAPI()
+
+    # Store connections by pc_id
+    pcs_map: Dict[str, SmallWebRTCConnection] = {}
+
+    ice_servers = [
+        IceServer(
+            urls="stun:stun.l.google.com:19302",
+        )
+    ]
+
+    # Mount the frontend at /
+    app.mount("/client", SmallWebRTCPrebuiltUI)
+
+    @app.get("/", include_in_schema=False)
+    async def root_redirect():
+        return RedirectResponse(url="/client/")
+
+    @app.post("/api/offer")
+    async def offer(request: dict, background_tasks: BackgroundTasks):
+        pc_id = request.get("pc_id")
+
+        if pc_id and pc_id in pcs_map:
+            pipecat_connection = pcs_map[pc_id]
+            logger.info(f"Reusing existing connection for pc_id: {pc_id}")
+            await pipecat_connection.renegotiate(
+                sdp=request["sdp"],
+                type=request["type"],
+                restart_pc=request.get("restart_pc", False),
+            )
+        else:
+            pipecat_connection = SmallWebRTCConnection(ice_servers)
+            await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"])
+
+            @pipecat_connection.event_handler("closed")
+            async def handle_disconnected(webrtc_connection: SmallWebRTCConnection):
+                logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}")
+                pcs_map.pop(webrtc_connection.pc_id, None)
+
+            # Run example function with SmallWebRTC transport arguments.
+            transport = SmallWebRTCTransport(params=params, webrtc_connection=pipecat_connection)
+            background_tasks.add_task(run_example, transport, args, False)
+
+        answer = pipecat_connection.get_answer()
+        # Updating the peer connection inside the map
+        pcs_map[answer["pc_id"]] = pipecat_connection
+
+        return answer
+
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        yield  # Run app
+        coros = [pc.close() for pc in pcs_map.values()]
+        await asyncio.gather(*coros)
+        pcs_map.clear()
+
+    uvicorn.run(app, host=args.host, port=args.port)
+
+
+def run_example_twilio(
+    run_example: Callable,
+    args: argparse.Namespace,
+    params: FastAPIWebsocketParams,
+):
+    logger.info("Running example with FastAPIWebsocketTransport (Twilio)...")
+
+    app = FastAPI()
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],  # Allow all origins for testing
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.post("/")
+    async def start_call():
+        logger.debug("POST TwiML")
+
+        xml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
+<Response>
+  <Connect>
+    <Stream url="wss://{args.proxy}/ws"></Stream>
+  </Connect>
+  <Pause length="40"/>
+</Response>
+        """
+        return HTMLResponse(content=xml_content, media_type="application/xml")
+
+    @app.websocket("/ws")
+    async def websocket_endpoint(websocket: WebSocket):
+        await websocket.accept()
+
+        logger.debug("WebSocket connection accepted")
+
+        # Reading Twilio data.
+        start_data = websocket.iter_text()
+        await start_data.__anext__()
+        call_data = json.loads(await start_data.__anext__())
+        print(call_data, flush=True)
+        stream_sid = call_data["start"]["streamSid"]
+        call_sid = call_data["start"]["callSid"]
+
+        # Create websocket transport and update params.
+        params.add_wav_header = False
+        params.serializer = TwilioFrameSerializer(
+            stream_sid=stream_sid,
+            call_sid=call_sid,
+            account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""),
+            auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""),
+        )
+        transport = FastAPIWebsocketTransport(websocket=websocket, params=params)
+        await run_example(transport, args, False)
+
+    uvicorn.run(app, host=args.host, port=args.port)
+
+
+def run_main(
+    run_example: Callable,
+    args: argparse.Namespace,
+    transport_params: Mapping[str, Callable] = {},
+):
+    if args.transport not in transport_params:
+        logger.error(f"Transport '{args.transport}' not supported by this example")
+        return
+
+    params = transport_params[args.transport]()
+    match args.transport:
+        case "daily":
+            run_example_daily(run_example, args, params)
+        case "webrtc":
+            run_example_webrtc(run_example, args, params)
+        case "twilio":
+            run_example_twilio(run_example, args, params)
+
+
+def main(
+    run_example: Callable,
+    *,
+    parser: Optional[argparse.ArgumentParser] = None,
+    transport_params: Mapping[str, Callable] = {},
+):
+    if not parser:
+        parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
+    parser.add_argument(
+        "--host", default="localhost", help="Host for HTTP server (default: localhost)"
+    )
+    parser.add_argument(
+        "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)"
+    )
+    parser.add_argument(
+        "--transport",
+        "-t",
+        type=str,
+        choices=["daily", "webrtc", "twilio"],
+        default="webrtc",
+        help="The transport this example should use",
+    )
+    parser.add_argument(
+        "--proxy", "-x", help="A public proxy host name (no protocol, e.g. proxy.example.com)"
+    )
+    parser.add_argument("--verbose", "-v", action="count", default=0)
+    args = parser.parse_args()
+
+    # Log level
+    logger.remove(0)
+    logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG")
+
+    # Import the bot file
+    run_main(run_example, args, transport_params)

src/pipecat/frames/frames.py

@@ -228,14 +228,15 @@ class TTSTextFrame(TextFrame):
 
 @dataclass
 class TranscriptionFrame(TextFrame):
-    """A text frame with transcription-specific data. Will be placed in the
-    transport's receive queue when a participant speaks.
+    """A text frame with transcription-specific data. The `result` field
+    contains the result from the STT service if available.
 
     """
 
     user_id: str
     timestamp: str
     language: Optional[Language] = None
+    result: Optional[Any] = None
 
     def __str__(self):
         return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
@@ -243,14 +244,16 @@ class TranscriptionFrame(TextFrame):
 
 @dataclass
 class InterimTranscriptionFrame(TextFrame):
-    """A text frame with interim transcription-specific data. Will be placed in
-    the transport's receive queue when a participant speaks.
+    """A text frame with interim transcription-specific data. The `result` field
+    contains the result from the STT service if available.
+
     """
 
     text: str
     user_id: str
     timestamp: str
     language: Optional[Language] = None
+    result: Optional[Any] = None
 
     def __str__(self):
         return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
@@ -413,22 +416,19 @@ class TransportMessageFrame(DataFrame):
 
 
 @dataclass
-class DTMFFrame(DataFrame):
+class DTMFFrame:
     """A DTMF button frame"""
 
     button: KeypadEntry
 
 
 @dataclass
-class InputDTMFFrame(DTMFFrame):
-    """A DTMF button input"""
+class OutputDTMFFrame(DTMFFrame, DataFrame):
+    """A DTMF keypress output that will be queued. If your transport supports
+    multiple dial-out destinations, use the `transport_destination` field to
+    specify where the DTMF keypress should be sent.
 
-    pass
-
-
-@dataclass
-class OutputDTMFFrame(DTMFFrame):
-    """A DTMF button output"""
+    """
 
     pass
 
@@ -777,6 +777,24 @@ class VisionImageRawFrame(InputImageRawFrame):
         return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})"
 
 
+@dataclass
+class InputDTMFFrame(DTMFFrame, SystemFrame):
+    """A DTMF keypress input."""
+
+    pass
+
+
+@dataclass
+class OutputDTMFUrgentFrame(DTMFFrame, SystemFrame):
+    """A DTMF keypress output that will be sent right away. If your transport
+    supports multiple dial-out destinations, use the `transport_destination`
+    field to specify where the DTMF keypress should be sent.
+
+    """
+
+    pass
+
+
 #
 # Control frames
 #

src/pipecat/pipeline/task.py

@@ -9,7 +9,7 @@ import time
 from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type
 
 from loguru import logger
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 from pipecat.clocks.base_clock import BaseClock
 from pipecat.clocks.system_clock import SystemClock
@@ -69,10 +69,10 @@ class PipelineParams(BaseModel):
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
     heartbeats_period_secs: float = HEARTBEAT_SECONDS
-    observers: List[BaseObserver] = []
+    observers: List[BaseObserver] = Field(default_factory=list)
     report_only_initial_ttfb: bool = False
     send_initial_empty_metrics: bool = True
-    start_metadata: Dict[str, Any] = {}
+    start_metadata: Dict[str, Any] = Field(default_factory=dict)
 
 
 class PipelineTaskSource(FrameProcessor):
@@ -236,6 +236,7 @@ class PipelineTask(BaseTask):
             )
             observers.append(self._turn_trace_observer)
         self._finished = False
+        self._cancelled = False
 
         # This queue receives frames coming from the pipeline upstream.
         self._up_queue = asyncio.Queue()
@@ -346,7 +347,6 @@ class PipelineTask(BaseTask):
 
     async def cancel(self):
         """Stops the running pipeline immediately."""
-        logger.debug(f"Canceling pipeline task {self}")
         await self._cancel()
 
     async def run(self):
@@ -406,12 +406,15 @@ class PipelineTask(BaseTask):
                 await self.queue_frame(frame)
 
     async def _cancel(self):
-        # Make sure everything is cleaned up downstream. This is sent
-        # out-of-band from the main streaming task which is what we want since
-        # we want to cancel right away.
-        await self._source.push_frame(CancelFrame())
-        # Only cancel the push task. Everything else will be cancelled in run().
-        await self._task_manager.cancel_task(self._process_push_task)
+        if not self._cancelled:
+            logger.debug(f"Canceling pipeline task {self}")
+            self._cancelled = True
+            # Make sure everything is cleaned up downstream. This is sent
+            # out-of-band from the main streaming task which is what we want since
+            # we want to cancel right away.
+            await self._source.push_frame(CancelFrame())
+            # Only cancel the push task. Everything else will be cancelled in run().
+            await self._task_manager.cancel_task(self._process_push_task)
 
     async def _create_tasks(self):
         self._process_up_task = self._task_manager.create_task(

src/pipecat/processors/aggregators/dtmf_aggregator.py

@@ -0,0 +1,143 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from typing import Optional
+
+from pipecat.frames.frames import (
+    BotInterruptionFrame,
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InputDTMFFrame,
+    KeypadEntry,
+    StartFrame,
+    TranscriptionFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.time import time_now_iso8601
+
+
+class DTMFAggregator(FrameProcessor):
+    """Aggregates DTMF frames into meaningful sequences for LLM processing.
+
+    The aggregator accumulates digits from InputDTMFFrame instances and flushes
+    when:
+    - Timeout occurs (configurable idle period)
+    - Termination digit is received (default: '#')
+    - EndFrame or CancelFrame is received
+
+    Emits TranscriptionFrame for compatibility with existing LLM context aggregators.
+
+    Args:
+        timeout: Idle timeout in seconds before flushing
+        termination_digit: Digit that triggers immediate flush
+        prefix: Prefix added to DTMF sequence in transcription
+    """
+
+    def __init__(
+        self,
+        timeout: float = 2.0,
+        termination_digit: KeypadEntry = KeypadEntry.POUND,
+        prefix: str = "DTMF: ",
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._aggregation = ""
+        self._idle_timeout = timeout
+        self._termination_digit = termination_digit
+        self._prefix = prefix
+
+        self._digit_event = asyncio.Event()
+        self._aggregation_task: Optional[asyncio.Task] = None
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartFrame):
+            self._create_aggregation_task()
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, (EndFrame, CancelFrame)):
+            if self._aggregation:
+                await self._flush_aggregation()
+            await self._stop_aggregation_task()
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, InputDTMFFrame):
+            # Push the DTMF frame downstream first
+            await self.push_frame(frame, direction)
+            # Then handle it in order for the TranscriptionFrame to be emitted
+            # after the InputDTMFFrame
+            await self._handle_dtmf_frame(frame)
+        else:
+            await self.push_frame(frame, direction)
+
+    async def _handle_dtmf_frame(self, frame: InputDTMFFrame):
+        """Handle DTMF input frame."""
+        is_first_digit = not self._aggregation
+
+        digit_value = frame.button.value
+        self._aggregation += digit_value
+
+        # For first digit, schedule interruption in separate task
+        if is_first_digit:
+            asyncio.create_task(self._send_interruption_task())
+
+        # Check for immediate flush conditions
+        if frame.button == self._termination_digit:
+            await self._flush_aggregation()
+        else:
+            # Signal digit received for timeout handling
+            self._digit_event.set()
+
+    async def _send_interruption_task(self):
+        """Send interruption frame safely in a separate task."""
+        try:
+            # Send the interruption frame
+            await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
+        except Exception as e:
+            # Log error but don't propagate
+            print(f"Error sending interruption: {e}")
+
+    def _create_aggregation_task(self) -> None:
+        """Creates the aggregation task if it hasn't been created yet."""
+        if not self._aggregation_task:
+            self._aggregation_task = self.create_task(self._aggregation_task_handler())
+
+    async def _stop_aggregation_task(self) -> None:
+        """Stops the aggregation task."""
+        if self._aggregation_task:
+            await self.cancel_task(self._aggregation_task)
+            self._aggregation_task = None
+
+    async def _aggregation_task_handler(self):
+        """Background task that handles timeout-based flushing."""
+        while True:
+            try:
+                await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout)
+                self._digit_event.clear()
+            except asyncio.TimeoutError:
+                if self._aggregation:
+                    await self._flush_aggregation()
+
+    async def _flush_aggregation(self):
+        """Flush the current aggregation as a TranscriptionFrame."""
+        if not self._aggregation:
+            return
+
+        sequence = self._aggregation
+        transcription_text = f"{self._prefix}{sequence}"
+
+        transcription_frame = TranscriptionFrame(
+            text=transcription_text, user_id="", timestamp=time_now_iso8601()
+        )
+        await self.push_frame(transcription_frame)
+
+        self._aggregation = ""
+
+    async def cleanup(self) -> None:
+        """Clean up resources."""
+        await super().cleanup()
+        await self._stop_aggregation_task()

src/pipecat/processors/frameworks/rtvi.py

@@ -110,7 +110,7 @@ class RTVIActionArgument(BaseModel):
 class RTVIAction(BaseModel):
     service: str
     action: str
-    arguments: List[RTVIActionArgument] = []
+    arguments: List[RTVIActionArgument] = Field(default_factory=list)
     result: Literal["bool", "number", "string", "array", "object"]
     handler: Callable[["RTVIProcessor", str, Dict[str, Any]], Awaitable[ActionResult]] = Field(
         exclude=True

src/pipecat/services/aws/llm.py

@@ -606,6 +606,21 @@ class AWSBedrockLLMService(LLMService):
         assistant = AWSBedrockAssistantContextAggregator(context, params=assistant_params)
         return AWSBedrockContextAggregatorPair(_user=user, _assistant=assistant)
 
+    def _create_no_op_tool(self):
+        """Create a no-operation tool for AWS Bedrock when tool content exists but no tools are defined.
+
+        This is required because AWS Bedrock doesn't allow empty tool configurations after tools were
+        previously set. Other LLM vendors allow NOT_GIVEN or empty tool configurations,
+        but AWS Bedrock requires at least one tool to be defined.
+        """
+        return {
+            "toolSpec": {
+                "name": "no_operation",
+                "description": "Internal placeholder function. Do not call this function.",
+                "inputSchema": {"json": {"type": "object", "properties": {}, "required": []}},
+            }
+        }
+
     @traced_llm
     async def _process_context(self, context: AWSBedrockLLMContext):
         # Usage tracking
@@ -616,6 +631,8 @@ class AWSBedrockLLMService(LLMService):
         cache_creation_input_tokens = 0
         use_completion_tokens_estimate = False
 
+        using_noop_tool = False
+
         try:
             await self.push_frame(LLMFullResponseStartFrame())
             await self.start_processing_metrics()
@@ -640,12 +657,28 @@ class AWSBedrockLLMService(LLMService):
             # Add system message
             request_params["system"] = context.system
 
-            # Add tools if present
-            if context.tools:
-                tool_config = {"tools": context.tools}
+            # Check if messages contain tool use or tool result content blocks
+            has_tool_content = False
+            for message in context.messages:
+                if isinstance(message.get("content"), list):
+                    for content_item in message["content"]:
+                        if "toolUse" in content_item or "toolResult" in content_item:
+                            has_tool_content = True
+                            break
+                if has_tool_content:
+                    break
 
-                # Add tool_choice if specified
-                if context.tool_choice:
+            # Handle tools: use current tools, or no-op if tool content exists but no current tools
+            tools = context.tools or []
+            if has_tool_content and not tools:
+                tools = [self._create_no_op_tool()]
+                using_noop_tool = True
+
+            if tools:
+                tool_config = {"tools": tools}
+
+                # Only add tool_choice if we have real tools (not just no-op)
+                if not using_noop_tool and context.tool_choice:
                     if context.tool_choice == "auto":
                         tool_config["toolChoice"] = {"auto": {}}
                     elif context.tool_choice == "none":
@@ -704,12 +737,17 @@ class AWSBedrockLLMService(LLMService):
                     if event["messageStop"]["stopReason"] == "tool_use" and tool_use_block:
                         try:
                             arguments = json.loads(json_accumulator) if json_accumulator else {}
-                            await self.call_function(
-                                context=context,
-                                tool_call_id=tool_use_block["id"],
-                                function_name=tool_use_block["name"],
-                                arguments=arguments,
-                            )
+
+                            # Only call function if it's not the no_operation tool
+                            if not using_noop_tool:
+                                await self.call_function(
+                                    context=context,
+                                    tool_call_id=tool_use_block["id"],
+                                    function_name=tool_use_block["name"],
+                                    arguments=arguments,
+                                )
+                            else:
+                                logger.debug("Ignoring no_operation tool call")
                         except json.JSONDecodeError:
                             logger.error(f"Failed to parse tool arguments: {json_accumulator}")
 

src/pipecat/services/aws/stt.py

@@ -305,6 +305,7 @@ class AWSTranscribeSTTService(STTService):
                                             "",
                                             time_now_iso8601(),
                                             self._settings["language"],
+                                            result=result,
                                         )
                                     )
                                     await self._handle_transcription(
@@ -320,6 +321,7 @@ class AWSTranscribeSTTService(STTService):
                                             "",
                                             time_now_iso8601(),
                                             self._settings["language"],
+                                            result=result,
                                         )
                                     )
                 elif headers.get(":message-type") == "exception":

src/pipecat/services/azure/stt.py

@@ -121,7 +121,13 @@ class AzureSTTService(STTService):
     def _on_handle_recognized(self, event):
         if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0:
             language = getattr(event.result, "language", None) or self._settings.get("language")
-            frame = TranscriptionFrame(event.result.text, "", time_now_iso8601(), language)
+            frame = TranscriptionFrame(
+                event.result.text,
+                "",
+                time_now_iso8601(),
+                language,
+                result=event,
+            )
             asyncio.run_coroutine_threadsafe(
                 self._handle_transcription(event.result.text, True, language), self.get_event_loop()
             )

src/pipecat/services/deepgram/stt.py

@@ -78,17 +78,20 @@ class DeepgramSTTService(STTService):
             vad_events=False,
         )
 
-        merged_options = default_options
+        merged_options = default_options.to_dict()
         if live_options:
-            merged_options = LiveOptions(**{**default_options.to_dict(), **live_options.to_dict()})
+            default_model = default_options.model
+            merged_options.update(live_options.to_dict())
+            # NOTE(aleix): Fixes an in deepgram-sdk where `model` is initialized
+            # to the string "None" instead of the value `None`.
+            if "model" in merged_options and merged_options["model"] == "None":
+                merged_options["model"] = default_model
 
-        # deepgram connection requires language to be a string
-        if isinstance(merged_options.language, Language) and hasattr(
-            merged_options.language, "value"
-        ):
-            merged_options.language = merged_options.language.value
+        if "language" in merged_options and isinstance(merged_options["language"], Language):
+            merged_options["language"] = merged_options["language"].value
 
-        self._settings = merged_options.to_dict()
+        self.set_model_name(merged_options["model"])
+        self._settings = merged_options
         self._addons = addons
 
         self._client = DeepgramClient(
@@ -209,14 +212,26 @@ class DeepgramSTTService(STTService):
             await self.stop_ttfb_metrics()
             if is_final:
                 await self.push_frame(
-                    TranscriptionFrame(transcript, "", time_now_iso8601(), language)
+                    TranscriptionFrame(
+                        transcript,
+                        "",
+                        time_now_iso8601(),
+                        language,
+                        result=result,
+                    )
                 )
                 await self._handle_transcription(transcript, is_final, language)
                 await self.stop_processing_metrics()
             else:
                 # For interim transcriptions, just push the frame without tracing
                 await self.push_frame(
-                    InterimTranscriptionFrame(transcript, "", time_now_iso8601(), language)
+                    InterimTranscriptionFrame(
+                        transcript,
+                        "",
+                        time_now_iso8601(),
+                        language,
+                        result=result,
+                    )
                 )
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):

src/pipecat/services/fal/stt.py

@@ -252,7 +252,11 @@ class FalSTTService(SegmentedSTTService):
                     await self._handle_transcription(text, True, self._settings["language"])
                     logger.debug(f"Transcription: [{text}]")
                     yield TranscriptionFrame(
-                        text, "", time_now_iso8601(), Language(self._settings["language"])
+                        text,
+                        "",
+                        time_now_iso8601(),
+                        Language(self._settings["language"]),
+                        result=response,
                     )
 
         except Exception as e:

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -937,7 +937,10 @@ class GeminiMultimodalLiveLLMService(LLMService):
             logger.debug(f"[Transcription:user] [{complete_sentence}]")
             await self.push_frame(
                 TranscriptionFrame(
-                    text=complete_sentence, user_id="", timestamp=time_now_iso8601()
+                    text=complete_sentence,
+                    user_id="",
+                    timestamp=time_now_iso8601(),
+                    result=evt,
                 ),
                 FrameDirection.UPSTREAM,
             )

src/pipecat/services/gladia/stt.py

@@ -408,7 +408,13 @@ class GladiaSTTService(STTService):
                     if confidence >= self._confidence:
                         if is_final:
                             await self.push_frame(
-                                TranscriptionFrame(transcript, "", time_now_iso8601(), language)
+                                TranscriptionFrame(
+                                    transcript,
+                                    "",
+                                    time_now_iso8601(),
+                                    language,
+                                    result=content,
+                                )
                             )
                             await self._handle_transcription(
                                 transcript=transcript,
@@ -418,7 +424,11 @@ class GladiaSTTService(STTService):
                         else:
                             await self.push_frame(
                                 InterimTranscriptionFrame(
-                                    transcript, "", time_now_iso8601(), language
+                                    transcript,
+                                    "",
+                                    time_now_iso8601(),
+                                    language,
+                                    result=content,
                                 )
                             )
                 elif content["type"] == "translation":

src/pipecat/services/google/llm.py

@@ -367,7 +367,7 @@ class GoogleLLMContext(OpenAILLMContext):
                     }
                 )
             elif part.function_call:
-                args = type(part.function_call).to_dict(part.function_call).get("args", {})
+                args = part.function_call.args if hasattr(part.function_call, "args") else {}
                 msg["tool_calls"] = [
                     {
                         "id": part.function_call.name,
@@ -382,7 +382,9 @@ class GoogleLLMContext(OpenAILLMContext):
             elif part.function_response:
                 msg["role"] = "tool"
                 resp = (
-                    type(part.function_response).to_dict(part.function_response).get("response", {})
+                    part.function_response.response
+                    if hasattr(part.function_response, "response")
+                    else {}
                 )
                 msg["tool_call_id"] = part.function_response.name
                 msg["content"] = json.dumps(resp)

src/pipecat/services/google/stt.py

@@ -816,7 +816,13 @@ class GoogleSTTService(STTService):
                     if result.is_final:
                         self._last_transcript_was_final = True
                         await self.push_frame(
-                            TranscriptionFrame(transcript, "", time_now_iso8601(), primary_language)
+                            TranscriptionFrame(
+                                transcript,
+                                "",
+                                time_now_iso8601(),
+                                primary_language,
+                                result=result,
+                            )
                         )
                         await self.stop_processing_metrics()
                         await self._handle_transcription(
@@ -829,7 +835,11 @@ class GoogleSTTService(STTService):
                         await self.stop_ttfb_metrics()
                         await self.push_frame(
                             InterimTranscriptionFrame(
-                                transcript, "", time_now_iso8601(), primary_language
+                                transcript,
+                                "",
+                                time_now_iso8601(),
+                                primary_language,
+                                result=result,
                             )
                         )
 

src/pipecat/services/openai_realtime_beta/events.py

@@ -115,7 +115,7 @@ class ResponseProperties(BaseModel):
     instructions: Optional[str] = None
     voice: Optional[str] = None
     output_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None
-    tools: Optional[List[Dict]] = []
+    tools: Optional[List[Dict]] = Field(default_factory=list)
     tool_choice: Optional[Literal["auto", "none", "required"]] = None
     temperature: Optional[float] = None
     max_response_output_tokens: Optional[Union[int, Literal["inf"]]] = None

src/pipecat/services/openai_realtime_beta/openai.py

@@ -464,7 +464,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
         if self._send_transcription_frames:
             await self.push_frame(
                 # no way to get a language code?
-                InterimTranscriptionFrame(evt.delta, "", time_now_iso8601())
+                InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt)
             )
 
     async def handle_evt_input_audio_transcription_completed(self, evt):
@@ -473,7 +473,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
         if self._send_transcription_frames:
             await self.push_frame(
                 # no way to get a language code?
-                TranscriptionFrame(evt.transcript, "", time_now_iso8601())
+                TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt)
             )
         pair = self._user_and_response_message_tuple
         if pair:

src/pipecat/services/riva/stt.py

@@ -256,7 +256,13 @@ class RivaSTTService(STTService):
                 if result.is_final:
                     await self.stop_processing_metrics()
                     await self.push_frame(
-                        TranscriptionFrame(transcript, "", time_now_iso8601(), self._language_code)
+                        TranscriptionFrame(
+                            transcript,
+                            "",
+                            time_now_iso8601(),
+                            self._language_code,
+                            result=result,
+                        )
                     )
                     await self._handle_transcription(
                         transcript=transcript,
@@ -266,7 +272,11 @@ class RivaSTTService(STTService):
                 else:
                     await self.push_frame(
                         InterimTranscriptionFrame(
-                            transcript, "", time_now_iso8601(), self._language_code
+                            transcript,
+                            "",
+                            time_now_iso8601(),
+                            self._language_code,
+                            result=result,
                         )
                     )
 

src/pipecat/services/tavus/video.py

@@ -24,13 +24,14 @@ from pipecat.frames.frames import (
     StartFrame,
     StartInterruptionFrame,
     TTSAudioRawFrame,
-    TTSStartedFrame,
-    TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
 from pipecat.services.ai_service import AIService
 from pipecat.transports.services.tavus import TavusCallbacks, TavusParams, TavusTransportClient
 
+# Using the same values that we do in the BaseOutputTransport
+BOT_VAD_STOP_SECS = 0.35
+
 
 class TavusVideoService(AIService):
     """
@@ -169,16 +170,8 @@ class TavusVideoService(AIService):
         if isinstance(frame, StartInterruptionFrame):
             await self._handle_interruptions()
             await self.push_frame(frame, direction)
-        elif isinstance(frame, TTSStartedFrame):
-            await self.start_processing_metrics()
-            await self.start_ttfb_metrics()
-            self._current_idx_str = str(frame.id)
         elif isinstance(frame, TTSAudioRawFrame):
-            await self._queue_audio(frame.audio, frame.sample_rate, done=False)
-        elif isinstance(frame, TTSStoppedFrame):
-            await self._queue_audio(b"\x00\x00", self._client.in_sample_rate, done=True)
-            await self.stop_ttfb_metrics()
-            await self.stop_processing_metrics()
+            await self._queue.put(frame)
         else:
             await self.push_frame(frame, direction)
 
@@ -191,9 +184,6 @@ class TavusVideoService(AIService):
         await self._client.stop()
         self._other_participant_has_joined = False
 
-    async def _queue_audio(self, audio: bytes, in_rate: int, done: bool):
-        await self._queue.put((audio, in_rate, done))
-
     async def _create_send_task(self):
         if not self._send_task:
             self._queue = asyncio.Queue()
@@ -204,15 +194,6 @@ class TavusVideoService(AIService):
             await self.cancel_task(self._send_task)
             self._send_task = None
 
-    # TODO (Filipi): this should be all that is needed use this Microphone Echo mode
-    # https://docs.tavus.io/sections/conversational-video-interface/layers-and-modes-overview#microphone-echo
-    # This would allow us to send an audio stream for the replica to repeat
-    # Checking with Tavus what is the right way to create the Persona to make it work
-    # async def _send_task_handler(self):
-    #    while True:
-    #        (audio, in_rate, done) = await self._queue.get()
-    #        await self._client.write_raw_audio_frames(audio)
-
     async def _send_task_handler(self):
         # Daily app-messages have a 4kb limit and also a rate limit of 20
         # messages per second. Below, we only consider the rate limit because 1
@@ -221,38 +202,52 @@ class TavusVideoService(AIService):
         # limit (even including base64 encoding). For a sample rate of 16000,
         # that would be 32000 / 20 = 1600.
         sample_rate = self._client.out_sample_rate
+        # 50 ms of audio
         MAX_CHUNK_SIZE = int((sample_rate * 2) / 20)
 
         audio_buffer = bytearray()
+        current_idx_str = None
+        silence = b"\x00" * MAX_CHUNK_SIZE
         samples_sent = 0
-        start_time = time.time()
+        start_time = None
 
         while True:
-            (audio, in_rate, done) = await self._queue.get()
+            try:
+                frame = await asyncio.wait_for(self._queue.get(), timeout=BOT_VAD_STOP_SECS)
+                if isinstance(frame, TTSAudioRawFrame):
+                    # starting the new inference
+                    if current_idx_str is None:
+                        current_idx_str = str(frame.id)
+                        samples_sent = 0
+                        start_time = time.time()
 
-            if done:
+                    audio = await self._resampler.resample(
+                        frame.audio, frame.sample_rate, sample_rate
+                    )
+                    audio_buffer.extend(audio)
+                    while len(audio_buffer) >= MAX_CHUNK_SIZE:
+                        chunk = audio_buffer[:MAX_CHUNK_SIZE]
+                        audio_buffer = audio_buffer[MAX_CHUNK_SIZE:]
+
+                        # Compute wait time for synchronization
+                        wait = start_time + (samples_sent / sample_rate) - time.time()
+                        if wait > 0:
+                            logger.trace(f"TavusVideoService _send_task_handler wait: {wait}")
+                            await asyncio.sleep(wait)
+
+                        await self._client.encode_audio_and_send(
+                            bytes(chunk), False, current_idx_str
+                        )
+
+                        # Update timestamp based on number of samples sent
+                        samples_sent += len(chunk) // 2  # 2 bytes per sample (16-bit)
+            except asyncio.TimeoutError:
+                # Bot has stopped speaking
                 # Send any remaining audio.
                 if len(audio_buffer) > 0:
                     await self._client.encode_audio_and_send(
-                        bytes(audio_buffer), done, self._current_idx_str
+                        bytes(audio_buffer), False, current_idx_str
                     )
-                await self._client.encode_audio_and_send(audio, done, self._current_idx_str)
+                await self._client.encode_audio_and_send(silence, True, current_idx_str)
                 audio_buffer.clear()
-            else:
-                audio = await self._resampler.resample(audio, in_rate, sample_rate)
-                audio_buffer.extend(audio)
-                while len(audio_buffer) >= MAX_CHUNK_SIZE:
-                    chunk = audio_buffer[:MAX_CHUNK_SIZE]
-                    audio_buffer = audio_buffer[MAX_CHUNK_SIZE:]
-
-                    # Compute wait time for synchronization
-                    wait = start_time + (samples_sent / sample_rate) - time.time()
-                    if wait > 0:
-                        await asyncio.sleep(wait)
-
-                    await self._client.encode_audio_and_send(
-                        bytes(chunk), done, self._current_idx_str
-                    )
-
-                    # Update timestamp based on number of samples sent
-                    samples_sent += len(chunk) // 2  # 2 bytes per sample (16-bit)
+                current_idx_str = None

src/pipecat/transports/base_output.py

@@ -25,6 +25,8 @@ from pipecat.frames.frames import (
     Frame,
     MixerControlFrame,
     OutputAudioRawFrame,
+    OutputDTMFFrame,
+    OutputDTMFUrgentFrame,
     OutputImageRawFrame,
     SpriteFrame,
     StartFrame,
@@ -132,12 +134,13 @@ class BaseOutputTransport(FrameProcessor):
     async def register_audio_destination(self, destination: str):
         pass
 
-    async def write_raw_video_frame(
-        self, frame: OutputImageRawFrame, destination: Optional[str] = None
-    ):
+    async def write_video_frame(self, frame: OutputImageRawFrame):
         pass
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+        pass
+
+    async def write_dtmf(self, frame: OutputDTMFFrame | OutputDTMFUrgentFrame):
         pass
 
     async def send_audio(self, frame: OutputAudioRawFrame):
@@ -171,6 +174,8 @@ class BaseOutputTransport(FrameProcessor):
             await self._handle_frame(frame)
         elif isinstance(frame, TransportMessageUrgentFrame):
             await self.send_message(frame)
+        elif isinstance(frame, OutputDTMFUrgentFrame):
+            await self.write_dtmf(frame)
         elif isinstance(frame, SystemFrame):
             await self.push_frame(frame, direction)
         # Control frames.
@@ -425,6 +430,8 @@ class BaseOutputTransport(FrameProcessor):
                 await self._set_video_images(frame.images)
             elif isinstance(frame, TransportMessageFrame):
                 await self._transport.send_message(frame)
+            elif isinstance(frame, OutputDTMFFrame):
+                await self._transport.write_dtmf(frame)
 
         def _next_frame(self) -> AsyncGenerator[Frame, None]:
             async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[Frame, None]:
@@ -498,7 +505,7 @@ class BaseOutputTransport(FrameProcessor):
 
                 # Send audio.
                 if isinstance(frame, OutputAudioRawFrame):
-                    await self._transport.write_raw_audio_frames(frame.audio, self._destination)
+                    await self._transport.write_audio_frame(frame)
 
         #
         # Video handling
@@ -581,8 +588,7 @@ class BaseOutputTransport(FrameProcessor):
             frame = await self._transport.get_event_loop().run_in_executor(
                 self._executor, resize_frame, frame
             )
-
-            await self._transport.write_raw_video_frame(frame, self._destination)
+            await self._transport.write_video_frame(frame)
 
         #
         # Clock handling

src/pipecat/transports/base_transport.py

@@ -7,7 +7,7 @@
 from abc import abstractmethod
 from typing import List, Mapping, Optional
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
 from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer
@@ -34,7 +34,7 @@ class TransportParams(BaseModel):
     audio_out_bitrate: int = 96000
     audio_out_10ms_chunks: int = 4
     audio_out_mixer: Optional[BaseAudioMixer | Mapping[Optional[str], BaseAudioMixer]] = None
-    audio_out_destinations: List[str] = []
+    audio_out_destinations: List[str] = Field(default_factory=list)
     audio_in_enabled: bool = False
     audio_in_sample_rate: Optional[int] = None
     audio_in_channels: int = 1
@@ -49,7 +49,7 @@ class TransportParams(BaseModel):
     video_out_bitrate: int = 800000
     video_out_framerate: int = 30
     video_out_color_format: str = "RGB"
-    video_out_destinations: List[str] = []
+    video_out_destinations: List[str] = Field(default_factory=list)
     vad_enabled: bool = False
     vad_audio_passthrough: bool = False
     vad_analyzer: Optional[VADAnalyzer] = None

src/pipecat/transports/local/audio.py

@@ -10,7 +10,7 @@ from typing import Optional
 
 from loguru import logger
 
-from pipecat.frames.frames import InputAudioRawFrame, StartFrame
+from pipecat.frames.frames import InputAudioRawFrame, OutputAudioRawFrame, StartFrame
 from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
@@ -122,10 +122,10 @@ class LocalAudioOutputTransport(BaseOutputTransport):
             self._out_stream.close()
             self._out_stream = None
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         if self._out_stream:
             await self.get_event_loop().run_in_executor(
-                self._executor, self._out_stream.write, frames
+                self._executor, self._out_stream.write, frame.audio
             )
 
 

src/pipecat/transports/local/tk.py

@@ -12,7 +12,12 @@ from typing import Optional
 import numpy as np
 from loguru import logger
 
-from pipecat.frames.frames import InputAudioRawFrame, OutputImageRawFrame, StartFrame
+from pipecat.frames.frames import (
+    InputAudioRawFrame,
+    OutputAudioRawFrame,
+    OutputImageRawFrame,
+    StartFrame,
+)
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -135,15 +140,13 @@ class TkOutputTransport(BaseOutputTransport):
             self._out_stream.close()
             self._out_stream = None
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         if self._out_stream:
             await self.get_event_loop().run_in_executor(
-                self._executor, self._out_stream.write, frames
+                self._executor, self._out_stream.write, frame.audio
             )
 
-    async def write_raw_video_frame(
-        self, frame: OutputImageRawFrame, destination: Optional[str] = None
-    ):
+    async def write_video_frame(self, frame: OutputImageRawFrame):
         self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
 
     def _write_frame_to_tk(self, frame: OutputImageRawFrame):

src/pipecat/transports/network/fastapi_websocket.py

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
     TransportMessageFrame,
     TransportMessageUrgentFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
 from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
@@ -45,7 +45,7 @@ except ModuleNotFoundError as e:
 
 class FastAPIWebsocketParams(TransportParams):
     add_wav_header: bool = False
-    serializer: FrameSerializer
+    serializer: Optional[FrameSerializer] = None
     session_timeout: Optional[int] = None
 
 
@@ -125,7 +125,8 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
     async def start(self, frame: StartFrame):
         await super().start(frame)
         await self._client.setup(frame)
-        await self._params.serializer.setup(frame)
+        if self._params.serializer:
+            await self._params.serializer.setup(frame)
         if not self._monitor_websocket_task and self._params.session_timeout:
             self._monitor_websocket_task = self.create_task(self._monitor_websocket())
         await self._client.trigger_client_connected()
@@ -158,6 +159,9 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
     async def _receive_messages(self):
         try:
             async for message in self._client.receive():
+                if not self._params.serializer:
+                    continue
+
                 frame = await self._params.serializer.deserialize(message)
 
                 if not frame:
@@ -192,7 +196,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         self._client = client
         self._params = params
 
-        # write_raw_audio_frames() is called quickly, as soon as we get audio
+        # write_audio_frame() is called quickly, as soon as we get audio
         # (e.g. from the TTS), and since this is just a network connection we
         # would be sending it to quickly. Instead, we want to block to emulate
         # an audio device, this is what the send interval is. It will be
@@ -203,7 +207,8 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
     async def start(self, frame: StartFrame):
         await super().start(frame)
         await self._client.setup(frame)
-        await self._params.serializer.setup(frame)
+        if self._params.serializer:
+            await self._params.serializer.setup(frame)
         self._send_interval = (self.audio_chunk_size / self.sample_rate) / 2
         await self.set_transport_ready(frame)
 
@@ -231,7 +236,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
     async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         await self._write_frame(frame)
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         if self._client.is_closing:
             return
 
@@ -241,7 +246,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
             return
 
         frame = OutputAudioRawFrame(
-            audio=frames,
+            audio=frame.audio,
             sample_rate=self.sample_rate,
             num_channels=self._params.audio_out_channels,
         )
@@ -266,6 +271,9 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         await self._write_audio_sleep()
 
     async def _write_frame(self, frame: Frame):
+        if not self._params.serializer:
+            return
+
         try:
             payload = await self._params.serializer.serialize(frame)
             if payload:
@@ -302,7 +310,9 @@ class FastAPIWebsocketTransport(BaseTransport):
             on_session_timeout=self._on_session_timeout,
         )
 
-        is_binary = self._params.serializer.type == FrameSerializerType.BINARY
+        is_binary = False
+        if self._params.serializer:
+            is_binary = self._params.serializer.type == FrameSerializerType.BINARY
         self._client = FastAPIWebsocketClient(websocket, is_binary, self._callbacks)
 
         self._input = FastAPIWebsocketInputTransport(

src/pipecat/transports/network/small_webrtc.py

@@ -50,7 +50,6 @@ class SmallWebRTCCallbacks(BaseModel):
     on_app_message: Callable[[Any], Awaitable[None]]
     on_client_connected: Callable[[SmallWebRTCConnection], Awaitable[None]]
     on_client_disconnected: Callable[[SmallWebRTCConnection], Awaitable[None]]
-    on_client_closed: Callable[[SmallWebRTCConnection], Awaitable[None]]
 
 
 class RawAudioTrack(AudioStreamTrack):
@@ -169,7 +168,7 @@ class SmallWebRTCClient:
         @self._webrtc_connection.event_handler("disconnected")
         async def on_disconnected(connection: SmallWebRTCConnection):
             logger.debug("Peer connection lost.")
-            await self._handle_client_disconnected()
+            await self._handle_peer_disconnected()
 
         @self._webrtc_connection.event_handler("closed")
         async def on_closed(connection: SmallWebRTCConnection):
@@ -284,13 +283,11 @@ class SmallWebRTCClient:
                 )
                 yield audio_frame
 
-    async def write_raw_audio_frames(self, data: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         if self._can_send() and self._audio_output_track:
-            await self._audio_output_track.add_audio_bytes(data)
+            await self._audio_output_track.add_audio_bytes(frame.audio)
 
-    async def write_raw_video_frame(
-        self, frame: OutputImageRawFrame, destination: Optional[str] = None
-    ):
+    async def write_video_frame(self, frame: OutputImageRawFrame):
         if self._can_send() and self._video_output_track:
             self._video_output_track.add_video_frame(frame)
 
@@ -313,7 +310,7 @@ class SmallWebRTCClient:
             logger.info(f"Disconnecting to Small WebRTC")
             self._closing = True
             await self._webrtc_connection.disconnect()
-            await self._handle_client_disconnected()
+            await self._handle_peer_disconnected()
 
     async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         if self._can_send():
@@ -338,19 +335,18 @@ class SmallWebRTCClient:
 
         await self._callbacks.on_client_connected(self._webrtc_connection)
 
-    async def _handle_client_disconnected(self):
+    async def _handle_peer_disconnected(self):
         self._audio_input_track = None
         self._video_input_track = None
         self._audio_output_track = None
         self._video_output_track = None
-        await self._callbacks.on_client_disconnected(self._webrtc_connection)
 
     async def _handle_client_closed(self):
         self._audio_input_track = None
         self._video_input_track = None
         self._audio_output_track = None
         self._video_output_track = None
-        await self._callbacks.on_client_closed(self._webrtc_connection)
+        await self._callbacks.on_client_disconnected(self._webrtc_connection)
 
     async def _handle_app_message(self, message: Any):
         await self._callbacks.on_app_message(message)
@@ -501,13 +497,11 @@ class SmallWebRTCOutputTransport(BaseOutputTransport):
     async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         await self._client.send_message(frame)
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
-        await self._client.write_raw_audio_frames(frames)
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+        await self._client.write_audio_frame(frame)
 
-    async def write_raw_video_frame(
-        self, frame: OutputImageRawFrame, destination: Optional[str] = None
-    ):
-        await self._client.write_raw_video_frame(frame)
+    async def write_video_frame(self, frame: OutputImageRawFrame):
+        await self._client.write_video_frame(frame)
 
 
 class SmallWebRTCTransport(BaseTransport):
@@ -525,7 +519,6 @@ class SmallWebRTCTransport(BaseTransport):
             on_app_message=self._on_app_message,
             on_client_connected=self._on_client_connected,
             on_client_disconnected=self._on_client_disconnected,
-            on_client_closed=self._on_client_closed,
         )
 
         self._client = SmallWebRTCClient(webrtc_connection, self._callbacks)
@@ -538,7 +531,6 @@ class SmallWebRTCTransport(BaseTransport):
         self._register_event_handler("on_app_message")
         self._register_event_handler("on_client_connected")
         self._register_event_handler("on_client_disconnected")
-        self._register_event_handler("on_client_closed")
 
     def input(self) -> SmallWebRTCInputTransport:
         if not self._input:
@@ -572,6 +564,3 @@ class SmallWebRTCTransport(BaseTransport):
 
     async def _on_client_disconnected(self, webrtc_connection):
         await self._call_event_handler("on_client_disconnected", webrtc_connection)
-
-    async def _on_client_closed(self, webrtc_connection):
-        await self._call_event_handler("on_client_closed", webrtc_connection)

src/pipecat/transports/network/websocket_client.py

@@ -24,6 +24,7 @@ from pipecat.frames.frames import (
     TransportMessageFrame,
     TransportMessageUrgentFrame,
 )
+from pipecat.processors.frame_processor import FrameProcessorSetup
 from pipecat.serializers.base_serializer import FrameSerializer
 from pipecat.serializers.protobuf import ProtobufFrameSerializer
 from pipecat.transports.base_input import BaseInputTransport
@@ -34,7 +35,7 @@ from pipecat.utils.asyncio import BaseTaskManager
 
 class WebsocketClientParams(TransportParams):
     add_wav_header: bool = True
-    serializer: FrameSerializer = ProtobufFrameSerializer()
+    serializer: Optional[FrameSerializer] = None
 
 
 class WebsocketClientCallbacks(BaseModel):
@@ -68,10 +69,10 @@ class WebsocketClientSession:
             )
         return self._task_manager
 
-    async def setup(self, frame: StartFrame):
+    async def setup(self, task_manager: BaseTaskManager):
         self._leave_counter += 1
         if not self._task_manager:
-            self._task_manager = frame.task_manager
+            self._task_manager = task_manager
 
     async def connect(self):
         if self._websocket:
@@ -131,10 +132,14 @@ class WebsocketClientInputTransport(BaseInputTransport):
         self._session = session
         self._params = params
 
+    async def setup(self, setup: FrameProcessorSetup):
+        await super().setup(setup)
+        await self._session.setup(setup.task_manager)
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
-        await self._params.serializer.setup(frame)
-        await self._session.setup(frame)
+        if self._params.serializer:
+            await self._params.serializer.setup(frame)
         await self._session.connect()
         await self.set_transport_ready(frame)
 
@@ -151,6 +156,8 @@ class WebsocketClientInputTransport(BaseInputTransport):
         await self._transport.cleanup()
 
     async def on_message(self, websocket, message):
+        if not self._params.serializer:
+            return
         frame = await self._params.serializer.deserialize(message)
         if not frame:
             return
@@ -173,7 +180,7 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
         self._session = session
         self._params = params
 
-        # write_raw_audio_frames() is called quickly, as soon as we get audio
+        # write_audio_frame() is called quickly, as soon as we get audio
         # (e.g. from the TTS), and since this is just a network connection we
         # would be sending it to quickly. Instead, we want to block to emulate
         # an audio device, this is what the send interval is. It will be
@@ -181,11 +188,15 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
         self._send_interval = 0
         self._next_send_time = 0
 
+    async def setup(self, setup: FrameProcessorSetup):
+        await super().setup(setup)
+        await self._session.setup(setup.task_manager)
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
         self._send_interval = (self.audio_chunk_size / self.sample_rate) / 2
-        await self._params.serializer.setup(frame)
-        await self._session.setup(frame)
+        if self._params.serializer:
+            await self._params.serializer.setup(frame)
         await self._session.connect()
         await self.set_transport_ready(frame)
 
@@ -204,9 +215,9 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
     async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         await self._write_frame(frame)
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         frame = OutputAudioRawFrame(
-            audio=frames,
+            audio=frame.audio,
             sample_rate=self.sample_rate,
             num_channels=self._params.audio_out_channels,
         )
@@ -231,6 +242,8 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
         await self._write_audio_sleep()
 
     async def _write_frame(self, frame: Frame):
+        if not self._params.serializer:
+            return
         payload = await self._params.serializer.serialize(frame)
         if payload:
             await self._session.send(payload)
@@ -255,6 +268,7 @@ class WebsocketClientTransport(BaseTransport):
         super().__init__()
 
         self._params = params or WebsocketClientParams()
+        self._params.serializer = self._params.serializer or ProtobufFrameSerializer()
 
         callbacks = WebsocketClientCallbacks(
             on_connected=self._on_connected,

src/pipecat/transports/network/websocket_server.py

@@ -40,7 +40,7 @@ except ModuleNotFoundError as e:
 
 class WebsocketServerParams(TransportParams):
     add_wav_header: bool = False
-    serializer: FrameSerializer
+    serializer: Optional[FrameSerializer] = None
     session_timeout: Optional[int] = None
 
 
@@ -80,7 +80,8 @@ class WebsocketServerInputTransport(BaseInputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
-        await self._params.serializer.setup(frame)
+        if self._params.serializer:
+            await self._params.serializer.setup(frame)
         if not self._server_task:
             self._server_task = self.create_task(self._server_task_handler())
         await self.set_transport_ready(frame)
@@ -134,6 +135,9 @@ class WebsocketServerInputTransport(BaseInputTransport):
         # Handle incoming messages
         try:
             async for message in websocket:
+                if not self._params.serializer:
+                    continue
+
                 frame = await self._params.serializer.deserialize(message)
 
                 if not frame:
@@ -178,7 +182,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
 
         self._websocket: Optional[websockets.WebSocketServerProtocol] = None
 
-        # write_raw_audio_frames() is called quickly, as soon as we get audio
+        # write_audio_frame() is called quickly, as soon as we get audio
         # (e.g. from the TTS), and since this is just a network connection we
         # would be sending it to quickly. Instead, we want to block to emulate
         # an audio device, this is what the send interval is. It will be
@@ -194,7 +198,8 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
-        await self._params.serializer.setup(frame)
+        if self._params.serializer:
+            await self._params.serializer.setup(frame)
         self._send_interval = (self.audio_chunk_size / self.sample_rate) / 2
         await self.set_transport_ready(frame)
 
@@ -220,14 +225,14 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
     async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         await self._write_frame(frame)
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         if not self._websocket:
             # Simulate audio playback with a sleep.
             await self._write_audio_sleep()
             return
 
         frame = OutputAudioRawFrame(
-            audio=frames,
+            audio=frame.audio,
             sample_rate=self.sample_rate,
             num_channels=self._params.audio_out_channels,
         )
@@ -252,6 +257,9 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
         await self._write_audio_sleep()
 
     async def _write_frame(self, frame: Frame):
+        if not self._params.serializer:
+            return
+
         try:
             payload = await self._params.serializer.serialize(frame)
             if payload and self._websocket:

src/pipecat/transports/services/daily.py

@@ -22,6 +22,8 @@ from pipecat.frames.frames import (
     Frame,
     InterimTranscriptionFrame,
     OutputAudioRawFrame,
+    OutputDTMFFrame,
+    OutputDTMFUrgentFrame,
     OutputImageRawFrame,
     SpriteFrame,
     StartFrame,
@@ -370,9 +372,10 @@ class DailyTransportClient(EventHandler):
         self._custom_audio_tracks[destination] = await self.add_custom_audio_track(destination)
         self._client.update_publishing({"customAudio": {destination: True}})
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         future = self._get_event_loop().create_future()
 
+        destination = frame.transport_destination
         audio_source: Optional[CustomAudioSource] = None
         if not destination and self._microphone_track:
             audio_source = self._microphone_track.source
@@ -381,17 +384,15 @@ class DailyTransportClient(EventHandler):
             audio_source = track.source
 
         if audio_source:
-            audio_source.write_frames(frames, completion=completion_callback(future))
+            audio_source.write_frames(frame.audio, completion=completion_callback(future))
         else:
             logger.warning(f"{self} unable to write audio frames to destination [{destination}]")
             future.set_result(None)
 
         await future
 
-    async def write_raw_video_frame(
-        self, frame: OutputImageRawFrame, destination: Optional[str] = None
-    ):
-        if not destination and self._camera:
+    async def write_video_frame(self, frame: OutputImageRawFrame):
+        if not frame.transport_destination and self._camera:
             self._camera.write_frame(frame.image)
 
     async def setup(self, setup: FrameProcessorSetup):
@@ -476,7 +477,7 @@ class DailyTransportClient(EventHandler):
                 logger.info(f"Joined {self._room_url}")
 
                 if self._params.transcription_enabled:
-                    await self._start_transcription()
+                    await self.start_transcription(self._params.transcription_settings)
 
                 await self._callbacks.on_joined(data)
 
@@ -491,23 +492,6 @@ class DailyTransportClient(EventHandler):
             self._joining = False
             await self._callbacks.on_error(error_msg)
 
-    async def _start_transcription(self):
-        if not self._token:
-            logger.warning("Transcription can't be started without a room token")
-            return
-
-        logger.info(f"Enabling transcription with settings {self._params.transcription_settings}")
-
-        future = self._get_event_loop().create_future()
-        self._client.start_transcription(
-            settings=self._params.transcription_settings.model_dump(exclude_none=True),
-            completion=completion_callback(future),
-        )
-        error = await future
-        if error:
-            logger.error(f"Unable to start transcription: {error}")
-            return
-
     async def _join(self):
         future = self._get_event_loop().create_future()
 
@@ -577,7 +561,7 @@ class DailyTransportClient(EventHandler):
         logger.info(f"Leaving {self._room_url}")
 
         if self._params.transcription_enabled:
-            await self._stop_transcription()
+            await self.stop_transcription()
 
         # Remove any custom tracks, if any.
         for track_name, _ in self._custom_audio_tracks.items():
@@ -597,15 +581,6 @@ class DailyTransportClient(EventHandler):
             logger.error(error_msg)
             await self._callbacks.on_error(error_msg)
 
-    async def _stop_transcription(self):
-        if not self._token:
-            return
-        future = self._get_event_loop().create_future()
-        self._client.stop_transcription(completion=completion_callback(future))
-        error = await future
-        if error:
-            logger.error(f"Unable to stop transcription: {error}")
-
     async def _leave(self):
         future = self._get_event_loop().create_future()
         self._client.leave(completion=completion_callback(future))
@@ -623,14 +598,22 @@ class DailyTransportClient(EventHandler):
         return self._client.participant_counts()
 
     async def start_dialout(self, settings):
+        logger.debug(f"Starting dialout: settings={settings}")
+
         future = self._get_event_loop().create_future()
         self._client.start_dialout(settings, completion=completion_callback(future))
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to start dialout: {error}")
 
     async def stop_dialout(self, participant_id):
+        logger.debug(f"Stopping dialout: participant_id={participant_id}")
+
         future = self._get_event_loop().create_future()
         self._client.stop_dialout(participant_id, completion=completion_callback(future))
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to stop dialout: {error}")
 
     async def send_dtmf(self, settings):
         future = self._get_event_loop().create_future()
@@ -648,16 +631,54 @@ class DailyTransportClient(EventHandler):
         await future
 
     async def start_recording(self, streaming_settings, stream_id, force_new):
+        logger.debug(
+            f"Starting recording: stream_id={stream_id} force_new={force_new} settings={streaming_settings}"
+        )
+
         future = self._get_event_loop().create_future()
         self._client.start_recording(
             streaming_settings, stream_id, force_new, completion=completion_callback(future)
         )
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to start recording: {error}")
 
     async def stop_recording(self, stream_id):
+        logger.debug(f"Stopping recording: stream_id={stream_id}")
+
         future = self._get_event_loop().create_future()
         self._client.stop_recording(stream_id, completion=completion_callback(future))
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to stop recording: {error}")
+
+    async def start_transcription(self, settings):
+        if not self._token:
+            logger.warning("Transcription can't be started without a room token")
+            return
+
+        logger.debug(f"Starting transcription: settings={settings}")
+
+        future = self._get_event_loop().create_future()
+        self._client.start_transcription(
+            settings=self._params.transcription_settings.model_dump(exclude_none=True),
+            completion=completion_callback(future),
+        )
+        error = await future
+        if error:
+            logger.error(f"Unable to start transcription: {error}")
+
+    async def stop_transcription(self):
+        if not self._token:
+            return
+
+        logger.debug(f"Stopping transcription")
+
+        future = self._get_event_loop().create_future()
+        self._client.stop_transcription(completion=completion_callback(future))
+        error = await future
+        if error:
+            logger.error(f"Unable to stop transcription: {error}")
 
     async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
         if not self._joined:
@@ -695,7 +716,9 @@ class DailyTransportClient(EventHandler):
 
         self._audio_renderers.setdefault(participant_id, {})[audio_source] = callback
 
-        logger.info(f"Starting to capture [{audio_source}] audio from participant {participant_id}")
+        logger.debug(
+            f"Starting to capture [{audio_source}] audio from participant {participant_id}"
+        )
 
         self._client.set_audio_renderer(
             participant_id,
@@ -723,6 +746,10 @@ class DailyTransportClient(EventHandler):
 
         self._video_renderers.setdefault(participant_id, {})[video_source] = callback
 
+        logger.debug(
+            f"Starting to capture [{video_source}] video from participant {participant_id}"
+        )
+
         self._client.set_video_renderer(
             participant_id,
             self._video_frame_received,
@@ -1106,7 +1133,7 @@ class DailyInputTransport(BaseInputTransport):
             next_time = prev_time + 1 / framerate
             render_frame = (next_time - curr_time) < 0.1
 
-        elif self._video_renderers[participant_id][video_source]["render_next_frame"]:
+        if self._video_renderers[participant_id][video_source]["render_next_frame"]:
             request_frame = self._video_renderers[participant_id][video_source][
                 "render_next_frame"
             ].pop(0)
@@ -1194,13 +1221,19 @@ class DailyOutputTransport(BaseOutputTransport):
     async def register_audio_destination(self, destination: str):
         await self._client.register_audio_destination(destination)
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
-        await self._client.write_raw_audio_frames(frames, destination)
+    async def write_dtmf(self, frame: OutputDTMFFrame | OutputDTMFUrgentFrame):
+        await self._client.send_dtmf(
+            {
+                "sessionId": frame.transport_destination,
+                "tones": frame.button.value,
+            }
+        )
 
-    async def write_raw_video_frame(
-        self, frame: OutputImageRawFrame, destination: Optional[str] = None
-    ):
-        await self._client.write_raw_video_frame(frame, destination)
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+        await self._client.write_audio_frame(frame)
+
+    async def write_video_frame(self, frame: OutputImageRawFrame):
+        await self._client.write_video_frame(frame)
 
 
 class DailyTransport(BaseTransport):
@@ -1346,6 +1379,14 @@ class DailyTransport(BaseTransport):
         await self._client.stop_dialout(participant_id)
 
     async def send_dtmf(self, settings):
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "`DailyTransport.send_dtmf()` is deprecated, push an `OutputDTMFFrame` or an `OutputDTMFUrgentFrame` instead.",
+                DeprecationWarning,
+            )
         await self._client.send_dtmf(settings)
 
     async def sip_call_transfer(self, settings):
@@ -1360,6 +1401,12 @@ class DailyTransport(BaseTransport):
     async def stop_recording(self, stream_id=None):
         await self._client.stop_recording(stream_id)
 
+    async def start_transcription(self, settings=None):
+        await self._client.start_transcription(settings)
+
+    async def stop_transcription(self):
+        await self._client.stop_transcription()
+
     async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
         """Sends a chat message to Daily's Prebuilt main room.
 
@@ -1551,10 +1598,16 @@ class DailyTransport(BaseTransport):
         except KeyError:
             language = None
         if is_final:
-            frame = TranscriptionFrame(text, participant_id, timestamp, language)
+            frame = TranscriptionFrame(text, participant_id, timestamp, language, result=message)
             logger.debug(f"Transcription (from: {participant_id}): [{text}]")
         else:
-            frame = InterimTranscriptionFrame(text, participant_id, timestamp, language)
+            frame = InterimTranscriptionFrame(
+                text,
+                participant_id,
+                timestamp,
+                language,
+                result=message,
+            )
 
         if self._input:
             await self._input.push_transcription_frame(frame)

src/pipecat/transports/services/livekit.py

@@ -477,8 +477,8 @@ class LiveKitOutputTransport(BaseOutputTransport):
         else:
             await self._client.send_data(frame.message.encode())
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
-        livekit_audio = self._convert_pipecat_audio_to_livekit(frames)
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+        livekit_audio = self._convert_pipecat_audio_to_livekit(frame.audio)
         await self._client.publish_audio(livekit_audio)
 
     def _convert_pipecat_audio_to_livekit(self, pipecat_audio: bytes) -> rtc.AudioFrame:

src/pipecat/transports/services/tavus.py

@@ -11,17 +11,18 @@ from pydantic import BaseModel
 
 from pipecat.audio.utils import create_default_resampler
 from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     Frame,
     InputAudioRawFrame,
+    OutputAudioRawFrame,
     OutputImageRawFrame,
     StartFrame,
     StartInterruptionFrame,
     TransportMessageFrame,
     TransportMessageUrgentFrame,
-    TTSStartedFrame,
-    TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 from pipecat.transports.base_input import BaseInputTransport
@@ -290,12 +291,18 @@ class TavusTransportClient:
         await self.send_message(transport_frame)
 
     async def update_subscriptions(self, participant_settings=None, profile_settings=None):
+        if not self._client:
+            return
+
         await self._client.update_subscriptions(
             participant_settings=participant_settings, profile_settings=profile_settings
         )
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
-        await self._client.write_raw_audio_frames(frames, destination)
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+        if not self._client:
+            return
+
+        await self._client.write_audio_frame(frame)
 
 
 class TavusInputTransport(BaseInputTransport):
@@ -365,7 +372,8 @@ class TavusOutputTransport(BaseOutputTransport):
         self._client = client
         self._params = params
         self._samples_sent = 0
-        self._start_time = time.time()
+        self._start_time = None
+        self._current_idx_str: Optional[str] = None
 
     async def setup(self, setup: FrameProcessorSetup):
         await super().setup(setup)
@@ -377,8 +385,6 @@ class TavusOutputTransport(BaseOutputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
-        self._samples_sent = 0
-        self._start_time = time.time()
         await self._client.start(frame)
         await self.set_transport_ready(frame)
 
@@ -394,34 +400,46 @@ class TavusOutputTransport(BaseOutputTransport):
         logger.info(f"TavusOutputTransport sending message {frame}")
         await self._client.send_message(frame)
 
+    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        # The BotStartedSpeakingFrame and BotStoppedSpeakingFrame are created inside BaseOutputTransport
+        # so TavusOutputTransport never receives these frames.
+        # This is a workaround, so we can more reliably be aware when the bot has started or stopped speaking
+        if direction == FrameDirection.DOWNSTREAM:
+            if isinstance(frame, BotStartedSpeakingFrame):
+                if self._current_idx_str is not None:
+                    logger.warning("TavusOutputTransport self._current_idx_str is already defined!")
+                self._current_idx_str = str(frame.id)
+                self._start_time = time.time()
+                self._samples_sent = 0
+            elif isinstance(frame, BotStoppedSpeakingFrame):
+                silence = b"\x00" * self.audio_chunk_size
+                await self._client.encode_audio_and_send(silence, True, self._current_idx_str)
+                self._current_idx_str = None
+        await super().push_frame(frame, direction)
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
         if isinstance(frame, StartInterruptionFrame):
             await self._handle_interruptions()
-        elif isinstance(frame, TTSStartedFrame):
-            self._current_idx_str = str(frame.id)
-        elif isinstance(frame, TTSStoppedFrame):
-            logger.debug(f"TAVUS: {self}: stopped speaking")
-            await self._client.encode_audio_and_send(b"\x00\x00", True, self._current_idx_str)
 
     async def _handle_interruptions(self):
         await self._client.send_interrupt_message()
 
-    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame):
         # Compute wait time for synchronization
-        wait = self._start_time + (self._samples_sent / self._sample_rate) - time.time()
+        wait = self._start_time + (self._samples_sent / self.sample_rate) - time.time()
         if wait > 0:
+            logger.trace(f"TavusOutputTransport write_audio_frame wait: {wait}")
             await asyncio.sleep(wait)
 
-        await self._client.encode_audio_and_send(frames, False, self._current_idx_str)
+        if self._current_idx_str is None:
+            logger.warning("TavusOutputTransport self._current_idx_str not defined yet!")
+            return
+
+        await self._client.encode_audio_and_send(frame.audio, False, self._current_idx_str)
 
         # Update timestamp based on number of samples sent
-        self._samples_sent += len(frames) // 2  # 2 bytes per sample (16-bit)
-
-    async def write_raw_video_frame(
-        self, frame: OutputImageRawFrame, destination: Optional[str] = None
-    ):
-        pass
+        self._samples_sent += len(frame.audio) // 2  # 2 bytes per sample (16-bit)
 
 
 class TavusTransport(BaseTransport):

src/pipecat/utils/base_object.py

@@ -49,14 +49,16 @@ class BaseObject(ABC):
         return decorator
 
     def add_event_handler(self, event_name: str, handler):
-        if event_name not in self._event_handlers:
-            raise Exception(f"Event handler {event_name} not registered")
-        self._event_handlers[event_name].append(handler)
+        if event_name in self._event_handlers:
+            self._event_handlers[event_name].append(handler)
+        else:
+            logger.warning(f"Event handler {event_name} not registered")
 
     def _register_event_handler(self, event_name: str):
-        if event_name in self._event_handlers:
-            raise Exception(f"Event handler {event_name} already registered")
-        self._event_handlers[event_name] = []
+        if event_name not in self._event_handlers:
+            self._event_handlers[event_name] = []
+        else:
+            logger.warning(f"Event handler {event_name} not registered")
 
     async def _call_event_handler(self, event_name: str, *args, **kwargs):
         # If we haven't registered an event handler, we don't need to do

src/pipecat/utils/tracing/service_attributes.py

@@ -171,6 +171,7 @@ def add_llm_span_attributes(
     model: str,
     stream: bool = True,
     messages: Optional[str] = None,
+    output: Optional[str] = None,
     tools: Optional[str] = None,
     tool_count: Optional[int] = None,
     tool_choice: Optional[str] = None,
@@ -188,6 +189,7 @@ def add_llm_span_attributes(
         model: Model name/identifier
         stream: Whether streaming is enabled
         messages: JSON-serialized messages
+        output: Aggregated output text from the LLM
         tools: JSON-serialized tools configuration
         tool_count: Number of tools available
         tool_choice: Tool selection configuration
@@ -208,6 +210,9 @@ def add_llm_span_attributes(
     if messages:
         span.set_attribute("input", messages)
 
+    if output:
+        span.set_attribute("output", output)
+
     if tools:
         span.set_attribute("tools", tools)
 

src/pipecat/utils/tracing/service_decorators.py

@@ -282,6 +282,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
     - Tool configurations
     - Token usage metrics
     - Performance metrics like TTFB
+    - Aggregated output text
 
     Args:
         func: The LLM method to trace.
@@ -313,6 +314,26 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
                     span_name, context=parent_context
                 ) as current_span:
                     try:
+                        # Store original method and output aggregator
+                        original_push_frame = self.push_frame
+                        output_text = ""  # Simple string accumulation
+
+                        async def traced_push_frame(frame, direction=None):
+                            nonlocal output_text
+                            # Capture text from LLMTextFrame during streaming
+                            if (
+                                hasattr(frame, "__class__")
+                                and frame.__class__.__name__ == "LLMTextFrame"
+                                and hasattr(frame, "text")
+                            ):
+                                output_text += frame.text
+
+                            # Call original
+                            if direction is not None:
+                                return await original_push_frame(frame, direction)
+                            else:
+                                return await original_push_frame(frame)
+
                         # For token usage monitoring
                         original_start_llm_usage_metrics = None
                         if hasattr(self, "start_llm_usage_metrics"):
@@ -331,6 +352,9 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
                             self.start_llm_usage_metrics = wrapped_start_llm_usage_metrics
 
                         try:
+                            # Replace push_frame to capture output
+                            self.push_frame = traced_push_frame
+
                             # Detect if we're using Google's service
                             is_google_service = "google" in service_class_name.lower()
 
@@ -411,13 +435,24 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
 
                             # Add all gathered attributes to the span
                             add_llm_span_attributes(span=current_span, **attribute_kwargs)
-                        except Exception as e:
-                            logging.warning(f"Error adding initial LLM attributes: {e}")
 
-                        # Call the original function
-                        return await f(self, context, *args, **kwargs)
+                        except Exception as e:
+                            logging.warning(f"Error setting up LLM tracing: {e}")
+                            # Don't raise - let the function execute anyway
+
+                        # Run function with modified push_frame to capture the output
+                        result = await f(self, context, *args, **kwargs)
+
+                        # Add aggregated output after function completes, if available
+                        if output_text:
+                            current_span.set_attribute("output", output_text)
+
+                        return result
+
                     finally:
-                        # Restore the original methods if we overrode them
+                        # Always restore the original methods
+                        self.push_frame = original_push_frame
+
                         if (
                             "original_start_llm_usage_metrics" in locals()
                             and original_start_llm_usage_metrics