#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Tavus transport implementation for Pipecat.
This module provides integration with the Tavus platform for creating conversational
AI applications with avatars. It manages conversation sessions and provides real-time
audio/video streaming capabilities through the Tavus API.
"""
import asyncio
import base64
import os
import time
from collections.abc import Awaitable, Callable, Mapping
from functools import partial
from typing import Any
import aiohttp
from daily.daily import AudioData
from loguru import logger
from pydantic import BaseModel
from pipecat.audio.utils import create_stream_resampler
from pipecat.frames.frames import (
BotConnectedFrame,
BotStoppedSpeakingFrame,
CancelFrame,
ClientConnectedFrame,
EndFrame,
Frame,
InputAudioRawFrame,
InterruptionFrame,
OutputAudioRawFrame,
OutputTransportMessageFrame,
OutputTransportMessageUrgentFrame,
StartFrame,
TTSStoppedFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BOT_VAD_STOP_FALLBACK_SECS, BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import (
DailyCallbacks,
DailyParams,
DailyTransportClient,
)
[docs]
class TavusApi:
"""Helper class for interacting with the Tavus API (v2).
Provides methods for creating and managing conversations with Tavus avatars,
including conversation lifecycle management and persona information retrieval.
"""
BASE_URL = "https://tavusapi.com/v2"
MOCK_CONVERSATION_ID = "dev-conversation"
MOCK_PERSONA_NAME = "TestTavusTransport"
[docs]
def __init__(self, api_key: str, session: aiohttp.ClientSession):
"""Initialize the TavusApi client.
Args:
api_key: Tavus API key for authentication.
session: An aiohttp session for making HTTP requests.
"""
self._api_key = api_key
self._session = session
self._headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
# Only for development
self._dev_room_url = os.getenv("TAVUS_SAMPLE_ROOM_URL")
[docs]
async def create_conversation(self, replica_id: str, persona_id: str) -> dict:
"""Create a new conversation with the specified replica and persona.
Args:
replica_id: ID of the replica to use in the conversation.
persona_id: ID of the persona to use in the conversation.
Returns:
Dictionary containing conversation_id and conversation_url.
"""
if self._dev_room_url:
return {
"conversation_id": self.MOCK_CONVERSATION_ID,
"conversation_url": self._dev_room_url,
}
logger.debug(f"Creating Tavus conversation: replica={replica_id}, persona={persona_id}")
url = f"{self.BASE_URL}/conversations"
payload = {
"replica_id": replica_id,
"persona_id": persona_id,
}
async with self._session.post(url, headers=self._headers, json=payload) as r:
r.raise_for_status()
response = await r.json()
logger.debug(f"Created Tavus conversation: {response}")
return response
[docs]
async def end_conversation(self, conversation_id: str):
"""End an existing conversation.
Args:
conversation_id: ID of the conversation to end.
"""
if conversation_id is None or conversation_id == self.MOCK_CONVERSATION_ID:
return
url = f"{self.BASE_URL}/conversations/{conversation_id}/end"
async with self._session.post(url, headers=self._headers) as r:
r.raise_for_status()
logger.debug(f"Ended Tavus conversation {conversation_id}")
[docs]
async def get_persona_name(self, persona_id: str) -> str:
"""Get the name of a persona by ID.
Args:
persona_id: ID of the persona to retrieve.
Returns:
The name of the persona.
"""
if self._dev_room_url is not None:
return self.MOCK_PERSONA_NAME
url = f"{self.BASE_URL}/personas/{persona_id}"
async with self._session.get(url, headers=self._headers) as r:
r.raise_for_status()
response = await r.json()
logger.debug(f"Fetched Tavus persona: {response}")
return response["persona_name"]
[docs]
class TavusCallbacks(BaseModel):
"""Callback handlers for Tavus events.
Parameters:
on_joined: Called when the bot joins the Daily room.
on_participant_joined: Called when a participant joins the conversation.
on_participant_left: Called when a participant leaves the conversation.
"""
on_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]]
[docs]
class TavusParams(DailyParams):
"""Configuration parameters for the Tavus transport.
Parameters:
audio_in_enabled: Whether to enable audio input from participants.
audio_out_enabled: Whether to enable audio output to participants.
microphone_out_enabled: Whether to enable microphone output track.
audio_out_faster_than_realtime: Whether to send bot audio app messages as fast as
possible instead of paced to real playback time. Speeds up avatar rendering, but
breaks any downstream processor (e.g. ``AudioBufferProcessor``) that relies on bot
audio arriving at the pipeline at roughly real time.
"""
audio_in_enabled: bool = True
audio_out_enabled: bool = True
microphone_out_enabled: bool = False
audio_out_faster_than_realtime: bool = False
[docs]
class TavusTransportClient:
"""Transport client that integrates Pipecat with the Tavus platform.
A transport client that integrates a Pipecat Bot with the Tavus platform by managing
conversation sessions using the Tavus API.
This client uses `TavusApi` to interact with the Tavus backend services. When a conversation
is started via `TavusApi`, Tavus provides a `roomURL` that can be used to connect the Pipecat Bot
into the same virtual room where the TavusBot is operating.
"""
[docs]
def __init__(
self,
*,
bot_name: str,
params: TavusParams = TavusParams(),
callbacks: TavusCallbacks,
api_key: str,
replica_id: str,
persona_id: str = "pipecat0",
session: aiohttp.ClientSession,
) -> None:
"""Initialize the Tavus transport client.
Args:
bot_name: The name of the Pipecat bot instance.
params: Optional parameters for Tavus operation.
callbacks: Callback handlers for Tavus-related events.
api_key: API key for authenticating with Tavus API.
replica_id: ID of the replica to use in the Tavus conversation.
persona_id: ID of the Tavus persona. Defaults to "pipecat0", which
signals Tavus to use the TTS voice of the Pipecat bot instead
of a Tavus persona voice, and to expect audio over the
`conversation.echo` app message API.
session: The aiohttp session for making async HTTP requests.
"""
self._bot_name = bot_name
self._api = TavusApi(api_key, session)
self._replica_id = replica_id
self._persona_id = persona_id
self._conversation_id: str | None = None
self._client: DailyTransportClient | None = None
self._callbacks = callbacks
self._params = params
self._task_manager = None
self._resampler = create_stream_resampler()
self._audio_queue: asyncio.Queue | None = None
self._send_task = None
# Utterance tracking for the realtime-paced send path (see
# send_realtime_audio_frame/end_realtime_utterance/reset_realtime_utterance).
self._realtime_inference_id: str | None = None
async def _initialize(self) -> str:
"""Initialize the conversation and return the room URL."""
response = await self._api.create_conversation(self._replica_id, self._persona_id)
self._conversation_id = response["conversation_id"]
return response["conversation_url"]
[docs]
async def setup(self, setup: FrameProcessorSetup):
"""Setup the client and initialize the conversation.
Args:
setup: The frame processor setup configuration.
"""
self._task_manager = setup.task_manager
if self._conversation_id is not None:
logger.debug(f"Conversation ID already defined: {self._conversation_id}")
return
try:
room_url = await self._initialize()
daily_callbacks = DailyCallbacks(
on_active_speaker_changed=partial(
self._on_handle_callback, "on_active_speaker_changed"
),
on_joined=self._on_joined,
on_left=self._on_left,
on_before_leave=partial(self._on_handle_callback, "on_before_leave"),
on_error=partial(self._on_handle_callback, "on_error"),
on_app_message=partial(self._on_handle_callback, "on_app_message"),
on_call_state_updated=partial(self._on_handle_callback, "on_call_state_updated"),
on_client_connected=partial(self._on_handle_callback, "on_client_connected"),
on_client_disconnected=partial(self._on_handle_callback, "on_client_disconnected"),
on_dialin_connected=partial(self._on_handle_callback, "on_dialin_connected"),
on_dialin_ready=partial(self._on_handle_callback, "on_dialin_ready"),
on_dialin_stopped=partial(self._on_handle_callback, "on_dialin_stopped"),
on_dialin_error=partial(self._on_handle_callback, "on_dialin_error"),
on_dialin_warning=partial(self._on_handle_callback, "on_dialin_warning"),
on_dialout_answered=partial(self._on_handle_callback, "on_dialout_answered"),
on_dialout_connected=partial(self._on_handle_callback, "on_dialout_connected"),
on_dialout_stopped=partial(self._on_handle_callback, "on_dialout_stopped"),
on_dialout_error=partial(self._on_handle_callback, "on_dialout_error"),
on_dialout_warning=partial(self._on_handle_callback, "on_dialout_warning"),
on_dtmf_event=partial(self._on_handle_callback, "on_dtmf_event"),
on_participant_joined=self._callbacks.on_participant_joined,
on_participant_left=self._callbacks.on_participant_left,
on_participant_updated=partial(self._on_handle_callback, "on_participant_updated"),
on_transcription_message=partial(
self._on_handle_callback, "on_transcription_message"
),
on_recording_started=partial(self._on_handle_callback, "on_recording_started"),
on_recording_stopped=partial(self._on_handle_callback, "on_recording_stopped"),
on_recording_error=partial(self._on_handle_callback, "on_recording_error"),
on_transcription_stopped=partial(
self._on_handle_callback, "on_transcription_stopped"
),
on_transcription_error=partial(self._on_handle_callback, "on_transcription_error"),
)
self._client = DailyTransportClient(
room_url, None, "Pipecat", self._params, daily_callbacks, self._bot_name
)
await self._client.setup(setup)
except Exception as e:
logger.error(f"Failed to setup TavusTransportClient: {e}")
await self._api.end_conversation(self._conversation_id)
self._conversation_id = None
[docs]
async def cleanup(self):
"""Cleanup client resources."""
await self._end_conversation()
try:
await self._client.cleanup()
except Exception as e:
logger.error(f"Exception during cleanup: {e}")
@property
def conversation_id(self) -> str | None:
"""Get the current conversation ID."""
return self._conversation_id
async def _on_joined(self, data):
"""Handle joined event."""
logger.debug("TavusTransportClient joined!")
await self._callbacks.on_joined(data)
async def _on_left(self):
"""Handle left event."""
logger.debug("TavusTransportClient left!")
async def _on_handle_callback(self, event_name, *args, **kwargs):
"""Handle generic callback events."""
logger.trace(f"[Callback] {event_name} called with args={args}, kwargs={kwargs}")
[docs]
async def get_persona_name(self) -> str:
"""Get the persona name from the API.
Returns:
The name of the current persona.
"""
return await self._api.get_persona_name(self._persona_id)
[docs]
async def start(self, frame: StartFrame):
"""Start the client and join the room.
Args:
frame: The start frame containing initialization parameters.
"""
logger.debug("TavusTransportClient start invoked!")
await self._client.start(frame)
await self._client.join()
[docs]
async def stop(self):
"""Stop the client and end the conversation."""
await self._client.leave()
await self._end_conversation()
async def _end_conversation(self):
"""End the Tavus conversation if one is active.
Idempotent so it can run from both ``stop()`` and ``cleanup()`` without
ending the conversation twice.
"""
if self._conversation_id is None:
return
conversation_id = self._conversation_id
self._conversation_id = None
await self._api.end_conversation(conversation_id)
[docs]
async def capture_participant_video(
self,
participant_id: str,
callback: Callable,
framerate: int = 30,
video_source: str = "camera",
color_format: str = "RGB",
):
"""Capture video from a participant.
Args:
participant_id: ID of the participant to capture video from.
callback: Callback function to handle video frames.
framerate: Desired framerate for video capture.
video_source: Video source to capture from.
color_format: Color format for video frames.
"""
await self._client.capture_participant_video(
participant_id, callback, framerate, video_source, color_format
)
[docs]
async def capture_participant_audio(
self,
participant_id: str,
callback: Callable,
audio_source: str = "microphone",
sample_rate: int = 16000,
callback_interval_ms: int = 20,
):
"""Capture audio from a participant.
Args:
participant_id: ID of the participant to capture audio from.
callback: Callback function to handle audio data.
audio_source: Audio source to capture from.
sample_rate: Desired sample rate for audio capture.
callback_interval_ms: Interval between audio callbacks in milliseconds.
"""
await self._client.capture_participant_audio(
participant_id, callback, audio_source, sample_rate, callback_interval_ms
)
[docs]
async def send_message(
self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame
):
"""Send a message to participants.
Args:
frame: The message frame to send.
"""
if self._client is None:
return
await self._client.send_message(frame)
@property
def out_sample_rate(self) -> int:
"""Get the output sample rate.
Returns:
The output sample rate in Hz.
"""
return self._client.out_sample_rate
@property
def in_sample_rate(self) -> int:
"""Get the input sample rate.
Returns:
The input sample rate in Hz.
"""
return self._client.in_sample_rate
[docs]
async def send_interrupt_message(self) -> None:
"""Send an interrupt message to the conversation."""
transport_frame = OutputTransportMessageUrgentFrame(
message={
"message_type": "conversation",
"event_type": "conversation.interrupt",
"conversation_id": self._conversation_id,
}
)
await self.send_message(transport_frame)
[docs]
async def encode_audio_and_send(
self, audio: bytes, done: bool, inference_id: str | None
) -> None:
"""Base64-encode audio bytes and send as a conversation.echo app message.
Args:
audio: Raw PCM bytes at the client output sample rate, 16-bit mono.
done: True when this is the final chunk for the current inference.
inference_id: Identifier tying all chunks of one utterance together.
"""
audio_base64 = base64.b64encode(audio).decode("utf-8")
transport_frame = OutputTransportMessageUrgentFrame(
message={
"message_type": "conversation",
"event_type": "conversation.echo",
"conversation_id": self._conversation_id,
"properties": {
"modality": "audio",
"inference_id": inference_id,
"audio": audio_base64,
"done": done,
"sample_rate": self.out_sample_rate,
},
}
)
await self.send_message(transport_frame)
[docs]
async def start_send_task(self) -> None:
"""Start the audio accumulation and send task."""
if not self._send_task:
self._audio_queue = asyncio.Queue()
self._send_task = self._task_manager.create_task(
self._send_task_handler(), "TavusTransportClient::send_task"
)
[docs]
async def cancel_send_task(self) -> None:
"""Cancel the send task and discard any buffered audio."""
if self._send_task:
await self._task_manager.cancel_task(self._send_task)
self._send_task = None
self._audio_queue = None
[docs]
async def queue_tts_frame(self, frame: OutputAudioRawFrame | TTSStoppedFrame) -> bool:
"""Add an audio frame or end-of-utterance signal to the send queue.
Args:
frame: An audio frame, or TTSStoppedFrame signalling end of utterance.
Returns:
True if the frame was queued, False if the queue is not active.
"""
if self._audio_queue is None:
return False
await self._audio_queue.put(frame)
return True
[docs]
async def send_realtime_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Send a single audio frame immediately, paced to real playback time.
Used instead of the queue/send-task path when audio isn't sent faster than
realtime: the caller (TavusOutputTransport) already paces one frame at a time.
Args:
frame: The audio frame to send.
Returns:
True, since the message is sent synchronously.
"""
if self._realtime_inference_id is None:
self._realtime_inference_id = str(frame.id)
await self.encode_audio_and_send(frame.audio, False, self._realtime_inference_id)
return True
[docs]
async def end_realtime_utterance(self) -> None:
"""Close the current realtime utterance, if any, with a done marker."""
if self._realtime_inference_id is None:
return
await self._send_utterance_done(self._realtime_inference_id)
self._realtime_inference_id = None
[docs]
def reset_realtime_utterance(self) -> None:
"""Discard the current realtime utterance without sending a done marker."""
self._realtime_inference_id = None
async def _send_utterance_done(self, inference_id: str) -> None:
"""Send a 40ms silence frame with done=True to close an utterance."""
done_silence = bytes(int(self.out_sample_rate * 2 / 25))
await self.encode_audio_and_send(done_silence, True, inference_id)
async def _flush_and_end_utterance(self, audio_buffer: bytearray, inference_id: str) -> None:
"""Flush remaining buffered audio then send the done marker."""
if audio_buffer:
await self.encode_audio_and_send(bytes(audio_buffer), False, inference_id)
audio_buffer.clear()
await self._send_utterance_done(inference_id)
async def _send_task_handler(self) -> None:
"""Accumulate audio into chunks and send via conversation.echo.
Derives inference_id from the first frame of each utterance. Accumulates
resampled audio until 100ms is reached, then sends with done=False.
Primary end-of-utterance signal: TTSStoppedFrame in the queue (queued by
TavusOutputTransport on BotStoppedSpeakingFrame, or by TavusVideoService on
TTSStoppedFrame). Fallback: BOT_VAD_STOP_FALLBACK_SECS timeout.
"""
sample_rate = self.out_sample_rate
audio_chunk_bytes = int(sample_rate * 2 * 0.1) # 100ms, 16-bit mono
audio_buffer = bytearray()
inference_id: str | None = None
while True:
try:
frame = await asyncio.wait_for(
self._audio_queue.get(), timeout=BOT_VAD_STOP_FALLBACK_SECS
)
if isinstance(frame, TTSStoppedFrame):
# Primary end-of-utterance signal — flush and mark done.
if inference_id:
await self._flush_and_end_utterance(audio_buffer, inference_id)
inference_id = None
else:
if inference_id is None:
inference_id = str(frame.id)
audio = frame.audio
if frame.sample_rate != sample_rate:
audio = await self._resampler.resample(
audio, frame.sample_rate, sample_rate
)
audio_buffer.extend(audio)
while len(audio_buffer) >= audio_chunk_bytes:
chunk = bytes(audio_buffer[:audio_chunk_bytes])
del audio_buffer[:audio_chunk_bytes]
await self.encode_audio_and_send(chunk, False, inference_id)
self._audio_queue.task_done()
except TimeoutError:
# Fallback: no frames received — flush if mid-utterance.
if not inference_id:
continue
await self._flush_and_end_utterance(audio_buffer, inference_id)
inference_id = None
[docs]
async def update_subscriptions(self, participant_settings=None, profile_settings=None):
"""Update subscription settings for participants.
Args:
participant_settings: Per-participant subscription settings.
profile_settings: Global subscription profile settings.
"""
if not self._client:
return
await self._client.update_subscriptions(
participant_settings=participant_settings, profile_settings=profile_settings
)
[docs]
async def register_audio_destination(self, destination: str, auto_silence: bool | None = True):
"""Register an audio destination for output.
Args:
destination: The destination identifier to register.
auto_silence: If True, the audio source inserts silence when no audio is available.
If False, the source waits for audio data. Defaults to True.
"""
if not self._client:
return
await self._client.register_audio_destination(destination, auto_silence=auto_silence)
[docs]
class TavusOutputTransport(BaseOutputTransport):
"""Output transport for sending audio and events to Tavus conversations.
Handles outgoing audio streams to participants and manages the custom
audio track expected by the Tavus platform.
"""
[docs]
def __init__(
self,
client: TavusTransportClient,
params: TavusParams,
**kwargs,
):
"""Initialize the Tavus output transport.
Args:
client: The Tavus transport client instance.
params: Transport configuration parameters.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(params, **kwargs)
self._client = client
self._params = params
# Whether we have seen a StartFrame already.
self._initialized = False
# Pacing state used when audio isn't sent faster than realtime (see
# write_audio_frame/_write_audio_sleep).
self._send_interval: float = 0
self._next_send_time: float = 0
[docs]
async def setup(self, setup: FrameProcessorSetup):
"""Setup the output transport.
Args:
setup: The frame processor setup configuration.
"""
await super().setup(setup)
await self._client.setup(setup)
[docs]
async def cleanup(self):
"""Cleanup output transport resources."""
await super().cleanup()
await self._client.cleanup()
[docs]
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
"""Signal end of utterance when bot stops speaking."""
if direction == FrameDirection.DOWNSTREAM and isinstance(frame, BotStoppedSpeakingFrame):
# Handle BotStoppedSpeakingFrame because, by the time it is received, the base output transport
# has already sent all audio frames via write_audio_frame(). At this point it is safe to mark
# the utterance as done.
if self._params.audio_out_faster_than_realtime:
await self._client.queue_tts_frame(TTSStoppedFrame())
else:
await self._client.end_realtime_utterance()
await super().push_frame(frame, direction)
[docs]
async def start(self, frame: StartFrame):
"""Start the output transport.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
if self._initialized:
return
self._initialized = True
await self._client.start(frame)
if self._params.audio_out_faster_than_realtime:
await self._client.start_send_task()
else:
self._send_interval = (self.audio_chunk_size / self._client.out_sample_rate) / 2
self._next_send_time = 0
await self.set_transport_ready(frame)
[docs]
async def stop(self, frame: EndFrame):
"""Stop the output transport.
Args:
frame: The end frame signaling transport shutdown.
"""
if self._params.audio_out_faster_than_realtime:
await self._client.cancel_send_task()
await super().stop(frame)
await self._client.stop()
[docs]
async def cancel(self, frame: CancelFrame):
"""Cancel the output transport.
Args:
frame: The cancel frame signaling immediate cancellation.
"""
if self._params.audio_out_faster_than_realtime:
await self._client.cancel_send_task()
await super().cancel(frame)
await self._client.stop()
[docs]
async def send_message(
self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame
):
"""Send a message to participants.
Args:
frame: The message frame to send.
"""
await self._client.send_message(frame)
[docs]
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames and handle interruptions.
Args:
frame: The frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
if isinstance(frame, InterruptionFrame):
await self._handle_interruptions()
async def _handle_interruptions(self):
"""Handle interruption events by discarding buffered audio and sending interrupt message."""
if self._params.audio_out_faster_than_realtime:
await self._client.cancel_send_task()
await self._client.send_interrupt_message()
await self._client.start_send_task()
else:
self._client.reset_realtime_utterance()
self._next_send_time = 0
await self._client.send_interrupt_message()
[docs]
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Send an audio frame via app message.
Args:
frame: The audio frame to write.
Returns:
True if the frame was sent (or queued) successfully.
"""
if self._params.audio_out_faster_than_realtime:
return await self._client.queue_tts_frame(frame)
sent = await self._client.send_realtime_audio_frame(frame)
await self._write_audio_sleep()
return sent
async def _write_audio_sleep(self):
"""Simulate real playback timing so audio is sent at roughly realtime pace."""
current_time = time.monotonic()
sleep_duration = max(0, self._next_send_time - current_time)
await asyncio.sleep(sleep_duration)
if sleep_duration == 0:
self._next_send_time = time.monotonic() + self._send_interval
else:
self._next_send_time += self._send_interval
[docs]
class TavusTransport(BaseTransport):
"""Transport implementation for Tavus video calls.
When used, the Pipecat bot joins the same virtual room as the Tavus Avatar and the user.
This is achieved by using `TavusTransportClient`, which initiates the conversation via
`TavusApi` and obtains a room URL that all participants connect to.
Event handlers available:
- on_connected(transport, data): Bot connected to the room
- on_client_connected(transport, participant): Participant connected to the session
- on_client_disconnected(transport, participant): Participant disconnected from the session
Example::
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, participant):
...
"""
[docs]
def __init__(
self,
bot_name: str,
session: aiohttp.ClientSession,
api_key: str,
replica_id: str,
persona_id: str = "pipecat0",
params: TavusParams = TavusParams(),
input_name: str | None = None,
output_name: str | None = None,
):
"""Initialize the Tavus transport.
Args:
bot_name: The name of the Pipecat bot.
session: aiohttp session used for async HTTP requests.
api_key: Tavus API key for authentication.
replica_id: ID of the replica model used for voice generation.
persona_id: ID of the Tavus persona. Defaults to "pipecat0", which
signals Tavus to use the TTS voice of the Pipecat bot instead
of a Tavus persona voice, and to expect audio over the
`conversation.echo` app message API.
params: Optional Tavus-specific configuration parameters.
input_name: Optional name for the input transport.
output_name: Optional name for the output transport.
"""
super().__init__(input_name=input_name, output_name=output_name)
self._params = params
callbacks = TavusCallbacks(
on_joined=self._on_joined,
on_participant_joined=self._on_participant_joined,
on_participant_left=self._on_participant_left,
)
self._client = TavusTransportClient(
bot_name="Pipecat",
callbacks=callbacks,
api_key=api_key,
replica_id=replica_id,
persona_id=persona_id,
session=session,
params=params,
)
self._input: TavusInputTransport | None = None
self._output: TavusOutputTransport | None = None
self._tavus_participant_id = None
# Register supported handlers. The user will only be able to register
# these handlers.
self._register_event_handler("on_connected")
self._register_event_handler("on_client_connected")
self._register_event_handler("on_client_disconnected")
async def _on_joined(self, data):
"""Handle bot joined room event."""
await self._call_event_handler("on_connected", data)
if self._input:
await self._input.push_frame(BotConnectedFrame())
async def _on_participant_left(self, participant, reason):
"""Handle participant left events."""
persona_name = await self._client.get_persona_name()
if participant.get("info", {}).get("userName", "") != persona_name:
await self._on_client_disconnected(participant)
async def _on_participant_joined(self, participant):
"""Handle participant joined events."""
# get persona, look up persona_name, set this as the bot name to ignore
persona_name = await self._client.get_persona_name()
# Ignore the Tavus replica's microphone
if participant.get("info", {}).get("userName", "") == persona_name:
self._tavus_participant_id = participant["id"]
else:
await self._on_client_connected(participant)
if self._tavus_participant_id:
logger.debug(f"Ignoring {self._tavus_participant_id}'s microphone")
await self.update_subscriptions(
participant_settings={
self._tavus_participant_id: {
"media": {"microphone": "unsubscribed"},
}
}
)
if self._input:
await self._input.start_capturing_audio(participant)
[docs]
async def update_subscriptions(self, participant_settings=None, profile_settings=None):
"""Update subscription settings for participants.
Args:
participant_settings: Per-participant subscription settings.
profile_settings: Global subscription profile settings.
"""
await self._client.update_subscriptions(
participant_settings=participant_settings,
profile_settings=profile_settings,
)
[docs]
def output(self) -> FrameProcessor:
"""Get the output transport for sending media and events.
Returns:
The Tavus output transport instance.
"""
if not self._output:
self._output = TavusOutputTransport(client=self._client, params=self._params)
return self._output
async def _on_client_connected(self, participant: Any):
"""Handle client connected events."""
await self._call_event_handler("on_client_connected", participant)
if self._input:
await self._input.push_frame(ClientConnectedFrame())
async def _on_client_disconnected(self, participant: Any):
"""Handle client disconnected events."""
await self._call_event_handler("on_client_disconnected", participant)