#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""LiveKit transport implementation for Pipecat.
This module provides comprehensive LiveKit real-time communication integration
including audio streaming, data messaging, participant management, and room
event handling for conversational AI applications.
"""
import asyncio
import json
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any
from loguru import logger
from pydantic import BaseModel
from pipecat.audio.utils import create_stream_resampler
from pipecat.frames.frames import (
AudioRawFrame,
BotConnectedFrame,
CancelFrame,
ClientConnectedFrame,
EndFrame,
Frame,
ImageRawFrame,
InterruptionFrame,
OutputAudioRawFrame,
OutputDTMFFrame,
OutputDTMFUrgentFrame,
OutputTransportMessageFrame,
OutputTransportMessageUrgentFrame,
StartFrame,
UserAudioRawFrame,
UserImageRawFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.utils.asyncio.task_manager import BaseTaskManager
try:
from livekit import rtc
from livekit.rtc._proto import video_frame_pb2 as proto_video_frame
from tenacity import retry, stop_after_attempt, wait_exponential
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use LiveKit, you need to `pip install pipecat-ai[livekit]`.")
raise Exception(f"Missing module: {e}")
# DTMF mapping according to RFC 4733
DTMF_CODE_MAP = {
"0": 0,
"1": 1,
"2": 2,
"3": 3,
"4": 4,
"5": 5,
"6": 6,
"7": 7,
"8": 8,
"9": 9,
"*": 10,
"#": 11,
}
[docs]
@dataclass
class LiveKitOutputTransportMessageFrame(OutputTransportMessageFrame):
"""Frame for transport messages in LiveKit rooms.
Parameters:
participant_id: Optional ID of the participant this message is for/from.
"""
participant_id: str | None = None
[docs]
@dataclass
class LiveKitOutputTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame):
"""Frame for urgent transport messages in LiveKit rooms.
Parameters:
participant_id: Optional ID of the participant this message is for/from.
"""
participant_id: str | None = None
[docs]
class LiveKitParams(TransportParams):
"""Configuration parameters for LiveKit transport.
Inherits all parameters from TransportParams without additional configuration.
"""
pass
[docs]
class LiveKitCallbacks(BaseModel):
"""Callback handlers for LiveKit events.
Parameters:
on_connected: Called when connected to the LiveKit room.
on_disconnected: Called when disconnected from the LiveKit room.
on_participant_connected: Called when a participant joins the room.
on_participant_disconnected: Called when a participant leaves the room.
on_audio_track_subscribed: Called when an audio track is subscribed.
on_audio_track_unsubscribed: Called when an audio track is unsubscribed.
on_data_received: Called when data is received from a participant.
on_first_participant_joined: Called when the first participant joins.
"""
on_connected: Callable[[], Awaitable[None]]
on_disconnected: Callable[[], Awaitable[None]]
on_before_disconnect: Callable[[], Awaitable[None]]
on_participant_connected: Callable[[str], Awaitable[None]]
on_participant_disconnected: Callable[[str], Awaitable[None]]
on_audio_track_subscribed: Callable[[str], Awaitable[None]]
on_audio_track_unsubscribed: Callable[[str], Awaitable[None]]
on_video_track_subscribed: Callable[[str], Awaitable[None]]
on_video_track_unsubscribed: Callable[[str], Awaitable[None]]
on_data_received: Callable[[bytes, str], Awaitable[None]]
on_first_participant_joined: Callable[[str], Awaitable[None]]
[docs]
class LiveKitTransportClient:
"""Core client for interacting with LiveKit rooms.
Manages the connection to LiveKit rooms and handles all low-level API interactions
including room management, audio streaming, data messaging, and event handling.
"""
[docs]
def __init__(
self,
url: str,
token: str,
room_name: str,
params: LiveKitParams,
callbacks: LiveKitCallbacks,
transport_name: str,
):
"""Initialize the LiveKit transport client.
Args:
url: LiveKit server URL to connect to.
token: Authentication token for the room.
room_name: Name of the LiveKit room to join.
params: Configuration parameters for the transport.
callbacks: Event callback handlers.
transport_name: Name identifier for the transport.
"""
self._url = url
self._token = token
self._room_name = room_name
self._params = params
self._callbacks = callbacks
self._transport_name = transport_name
self._room: rtc.Room | None = None
self._participant_id: str = ""
self._connected = False
self._disconnect_counter = 0
self._audio_source: rtc.AudioSource | None = None
self._audio_track: rtc.LocalAudioTrack | None = None
self._audio_tracks = {}
self._audio_queue = asyncio.Queue()
self._video_tracks = {}
self._video_queue = asyncio.Queue()
self._other_participant_has_joined = False
self._task_manager: BaseTaskManager | None = None
self._async_lock = asyncio.Lock()
@property
def participant_id(self) -> str:
"""Get the participant ID for this client.
Returns:
The participant ID assigned by LiveKit.
"""
return self._participant_id
@property
def room(self) -> rtc.Room:
"""Get the LiveKit room instance.
Returns:
The LiveKit room object.
Raises:
Exception: If room object is not available.
"""
if not self._room:
raise Exception(f"{self}: missing room object (pipeline not started?)")
return self._room
[docs]
async def setup(self, setup: FrameProcessorSetup):
"""Setup the client with task manager and room initialization.
Args:
setup: The frame processor setup configuration.
"""
if self._task_manager:
return
self._task_manager = setup.task_manager
self._room = rtc.Room(loop=self._task_manager.get_event_loop())
# Set up room event handlers
self.room.on("participant_connected")(self._on_participant_connected_wrapper)
self.room.on("participant_disconnected")(self._on_participant_disconnected_wrapper)
self.room.on("track_subscribed")(self._on_track_subscribed_wrapper)
self.room.on("track_unsubscribed")(self._on_track_unsubscribed_wrapper)
self.room.on("data_received")(self._on_data_received_wrapper)
self.room.on("connected")(self._on_connected_wrapper)
self.room.on("disconnected")(self._on_disconnected_wrapper)
[docs]
async def cleanup(self):
"""Cleanup client resources."""
await self.disconnect()
[docs]
async def start(self, frame: StartFrame):
"""Start the client and initialize audio components.
Args:
frame: The start frame containing initialization parameters.
"""
self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
[docs]
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def connect(self):
"""Connect to the LiveKit room with retry logic."""
async with self._async_lock:
if self._connected:
# Increment disconnect counter if already connected.
self._disconnect_counter += 1
return
logger.info(f"Connecting to {self._room_name}")
try:
await self.room.connect(
self._url,
self._token,
options=rtc.RoomOptions(auto_subscribe=True),
)
self._connected = True
# Increment disconnect counter if we successfully connected.
self._disconnect_counter += 1
self._participant_id = self.room.local_participant.sid
logger.info(f"Connected to {self._room_name}")
# Set up audio source and track
self._audio_source = rtc.AudioSource(
self._out_sample_rate, self._params.audio_out_channels
)
self._audio_track = rtc.LocalAudioTrack.create_audio_track(
"pipecat-audio", self._audio_source
)
options = rtc.TrackPublishOptions()
options.source = rtc.TrackSource.SOURCE_MICROPHONE
await self.room.local_participant.publish_track(self._audio_track, options)
await self._callbacks.on_connected()
# Check if there are already participants in the room
participants = self.get_participants()
if participants and not self._other_participant_has_joined:
self._other_participant_has_joined = True
await self._callbacks.on_first_participant_joined(participants[0])
except Exception as e:
logger.error(f"Error connecting to {self._room_name}: {e}")
raise
[docs]
async def disconnect(self):
"""Disconnect from the LiveKit room."""
async with self._async_lock:
# Decrement leave counter when leaving.
self._disconnect_counter -= 1
if not self._connected or self._disconnect_counter > 0:
return
logger.info(f"Disconnecting from {self._room_name}")
await self._callbacks.on_before_disconnect()
await self.room.disconnect()
self._connected = False
logger.info(f"Disconnected from {self._room_name}")
await self._callbacks.on_disconnected()
[docs]
async def send_data(self, data: bytes, participant_id: str | None = None):
"""Send data to participants in the room.
Args:
data: The data bytes to send.
participant_id: Optional specific participant to send to.
"""
if not self._connected:
return
try:
if participant_id:
await self.room.local_participant.publish_data(
data, reliable=True, destination_identities=[participant_id]
)
else:
await self.room.local_participant.publish_data(data, reliable=True)
except Exception as e:
logger.error(f"Error sending data: {e}")
[docs]
async def send_dtmf(self, digit: str):
r"""Send DTMF tone to the room.
Args:
digit: The DTMF digit to send (0-9, \*, #).
"""
if not self._connected:
return
if digit not in DTMF_CODE_MAP:
logger.warning(f"Invalid DTMF digit: {digit}")
return
code = DTMF_CODE_MAP[digit]
try:
await self.room.local_participant.publish_dtmf(code=code, digit=digit)
except Exception as e:
logger.error(f"Error sending DTMF tone {digit}: {e}")
[docs]
async def publish_audio(self, audio_frame: rtc.AudioFrame) -> bool:
"""Publish an audio frame to the room.
Args:
audio_frame: The LiveKit audio frame to publish.
"""
if not self._connected or not self._audio_source:
return False
try:
await self._audio_source.capture_frame(audio_frame)
return True
except Exception as e:
# When using an audio mixer, the base output transport's
# with_mixer() generator continuously yields frames (mixed with
# background audio) even when no TTS audio is queued. During
# interruptions, the audio task is cancelled and recreated, but
# there is a brief window where the native LiveKit AudioSource
# rejects capture_frame() with an InvalidState error. This is a
# transient condition — the mixer will produce a new frame within
# milliseconds, so we silently drop these frames.
if "InvalidState" not in str(e):
logger.error(f"Error publishing audio: {e}")
return False
[docs]
def get_participants(self) -> list[str]:
"""Get list of participant IDs in the room.
Returns:
List of participant IDs.
"""
return [p.sid for p in self.room.remote_participants.values()]
[docs]
async def mute_participant(self, participant_id: str):
"""Mute a specific participant's audio tracks.
Args:
participant_id: ID of the participant to mute.
"""
participant = self.room.remote_participants.get(participant_id)
if participant:
for track in participant.tracks.values():
if track.kind == "audio":
await track.set_enabled(False)
[docs]
async def unmute_participant(self, participant_id: str):
"""Unmute a specific participant's audio tracks.
Args:
participant_id: ID of the participant to unmute.
"""
participant = self.room.remote_participants.get(participant_id)
if participant:
for track in participant.tracks.values():
if track.kind == "audio":
await track.set_enabled(True)
# Wrapper methods for event handlers
def _on_participant_connected_wrapper(self, participant: rtc.RemoteParticipant):
"""Wrapper for participant connected events."""
self._task_manager.create_task(
self._async_on_participant_connected(participant),
f"{self}::_async_on_participant_connected",
)
def _on_participant_disconnected_wrapper(self, participant: rtc.RemoteParticipant):
"""Wrapper for participant disconnected events."""
self._task_manager.create_task(
self._async_on_participant_disconnected(participant),
f"{self}::_async_on_participant_disconnected",
)
def _on_track_subscribed_wrapper(
self,
track: rtc.Track,
publication: rtc.RemoteTrackPublication,
participant: rtc.RemoteParticipant,
):
"""Wrapper for track subscribed events."""
self._task_manager.create_task(
self._async_on_track_subscribed(track, publication, participant),
f"{self}::_async_on_track_subscribed",
)
def _on_track_unsubscribed_wrapper(
self,
track: rtc.Track,
publication: rtc.RemoteTrackPublication,
participant: rtc.RemoteParticipant,
):
"""Wrapper for track unsubscribed events."""
self._task_manager.create_task(
self._async_on_track_unsubscribed(track, publication, participant),
f"{self}::_async_on_track_unsubscribed",
)
def _on_data_received_wrapper(self, data: rtc.DataPacket):
"""Wrapper for data received events."""
self._task_manager.create_task(
self._async_on_data_received(data),
f"{self}::_async_on_data_received",
)
def _on_connected_wrapper(self):
"""Wrapper for connected events."""
self._task_manager.create_task(self._async_on_connected(), f"{self}::_async_on_connected")
def _on_disconnected_wrapper(self):
"""Wrapper for disconnected events."""
self._task_manager.create_task(
self._async_on_disconnected(), f"{self}::_async_on_disconnected"
)
# Async methods for event handling
async def _async_on_participant_connected(self, participant: rtc.RemoteParticipant):
"""Handle participant connected events."""
logger.info(f"Participant connected: {participant.identity}")
await self._callbacks.on_participant_connected(participant.sid)
if not self._other_participant_has_joined:
self._other_participant_has_joined = True
await self._callbacks.on_first_participant_joined(participant.sid)
async def _async_on_participant_disconnected(self, participant: rtc.RemoteParticipant):
"""Handle participant disconnected events."""
logger.info(f"Participant disconnected: {participant.identity}")
await self._callbacks.on_participant_disconnected(participant.sid)
if len(self.get_participants()) == 0:
self._other_participant_has_joined = False
async def _async_on_track_subscribed(
self,
track: rtc.Track,
publication: rtc.RemoteTrackPublication,
participant: rtc.RemoteParticipant,
):
"""Handle track subscribed events."""
if track.kind == rtc.TrackKind.KIND_AUDIO:
logger.info(f"Audio track subscribed: {track.sid} from participant {participant.sid}")
self._audio_tracks[participant.sid] = track
audio_stream = rtc.AudioStream(track)
self._task_manager.create_task(
self._process_audio_stream(audio_stream, participant.sid),
f"{self}::_process_audio_stream",
)
await self._callbacks.on_audio_track_subscribed(participant.sid)
elif track.kind == rtc.TrackKind.KIND_VIDEO:
logger.info(f"Video track subscribed: {track.sid} from participant {participant.sid}")
self._video_tracks[participant.sid] = track
# Only process video stream if video input is enabled to prevent
# unbounded queue growth when there is no consumer for video frames.
if self._params.video_in_enabled:
video_stream = rtc.VideoStream(track)
self._task_manager.create_task(
self._process_video_stream(video_stream, participant.sid),
f"{self}::_process_video_stream",
)
await self._callbacks.on_video_track_subscribed(participant.sid)
async def _async_on_track_unsubscribed(
self,
track: rtc.Track,
publication: rtc.RemoteTrackPublication,
participant: rtc.RemoteParticipant,
):
"""Handle track unsubscribed events."""
logger.info(f"Track unsubscribed: {publication.sid} from {participant.identity}")
if track.kind == rtc.TrackKind.KIND_AUDIO:
await self._callbacks.on_audio_track_unsubscribed(participant.sid)
elif track.kind == rtc.TrackKind.KIND_VIDEO:
await self._callbacks.on_video_track_unsubscribed(participant.sid)
async def _async_on_data_received(self, data: rtc.DataPacket):
"""Handle data received events."""
await self._callbacks.on_data_received(data.data, data.participant.sid)
async def _async_on_connected(self):
"""Handle connected events."""
await self._callbacks.on_connected()
async def _async_on_disconnected(self, reason=None):
"""Handle disconnected events."""
self._connected = False
logger.info(f"Disconnected from {self._room_name}. Reason: {reason}")
await self._callbacks.on_disconnected()
async def _process_audio_stream(self, audio_stream: rtc.AudioStream, participant_id: str):
"""Process incoming audio stream from a participant."""
logger.info(f"Started processing audio stream for participant {participant_id}")
async for event in audio_stream:
if isinstance(event, rtc.AudioFrameEvent):
await self._audio_queue.put((event, participant_id))
else:
logger.warning(f"Received unexpected event type: {type(event)}")
[docs]
async def get_next_audio_frame(self):
"""Get the next audio frame from the queue."""
while True:
frame, participant_id = await self._audio_queue.get()
yield frame, participant_id
async def _process_video_stream(self, video_stream: rtc.VideoStream, participant_id: str):
"""Process incoming video stream from a participant."""
logger.info(f"Started processing video stream for participant {participant_id}")
async for event in video_stream:
if isinstance(event, rtc.VideoFrameEvent):
await self._video_queue.put((event, participant_id))
else:
logger.warning(f"Received unexpected event type: {type(event)}")
[docs]
async def get_next_video_frame(self):
"""Get the next video frame from the queue."""
while True:
frame, participant_id = await self._video_queue.get()
yield frame, participant_id
def __str__(self):
"""String representation of the LiveKit transport client."""
return f"{self._transport_name}::LiveKitTransportClient"
[docs]
class LiveKitOutputTransport(BaseOutputTransport):
"""Handles outgoing media streams and events to LiveKit rooms.
Manages sending audio frames and data messages to LiveKit room participants,
including audio format conversion for LiveKit compatibility.
"""
[docs]
def __init__(
self,
transport: BaseTransport,
client: LiveKitTransportClient,
params: LiveKitParams,
**kwargs,
):
"""Initialize the LiveKit output transport.
Args:
transport: The parent transport instance.
client: LiveKitTransportClient instance.
params: Configuration parameters.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(params, **kwargs)
self._transport = transport
self._client = client
# Whether we have seen a StartFrame already.
self._initialized = False
[docs]
async def start(self, frame: StartFrame):
"""Start the output transport and connect to LiveKit room.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
if self._initialized:
return
self._initialized = True
await self._client.start(frame)
await self._client.connect()
await self.set_transport_ready(frame)
logger.info("LiveKitOutputTransport started")
[docs]
async def stop(self, frame: EndFrame):
"""Stop the output transport and disconnect from LiveKit room.
Args:
frame: The end frame signaling transport shutdown.
"""
await super().stop(frame)
await self._client.disconnect()
logger.info("LiveKitOutputTransport stopped")
[docs]
async def cancel(self, frame: CancelFrame):
"""Cancel the output transport and disconnect from LiveKit room.
Args:
frame: The cancel frame signaling immediate cancellation.
"""
await super().cancel(frame)
await self._client.disconnect()
[docs]
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames, clearing the LiveKit AudioSource buffer on interruption.
When an InterruptionFrame arrives, any audio already submitted to the
LiveKit AudioSource (but not yet played out) is cleared immediately so
the bot stops speaking without delay.
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) and self._client._audio_source is not None:
self._client._audio_source.clear_queue()
[docs]
async def setup(self, setup: FrameProcessorSetup):
"""Setup the output transport with shared client setup.
Args:
setup: The frame processor setup configuration.
"""
await super().setup(setup)
await self._client.setup(setup)
[docs]
async def cleanup(self):
"""Cleanup output transport and shared resources."""
await super().cleanup()
await self._transport.cleanup()
[docs]
async def send_message(
self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame
):
"""Send a transport message to participants.
Args:
frame: The transport message frame to send.
"""
message = frame.message
if isinstance(message, dict):
# fix message encoding for dict-like messages, e.g. RTVI messages.
message = json.dumps(message, ensure_ascii=False)
if isinstance(
frame, (LiveKitOutputTransportMessageFrame, LiveKitOutputTransportMessageUrgentFrame)
):
await self._client.send_data(message.encode(), frame.participant_id)
else:
await self._client.send_data(message.encode())
[docs]
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the LiveKit room.
Args:
frame: The audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
livekit_audio = self._convert_pipecat_audio_to_livekit(frame.audio)
return await self._client.publish_audio(livekit_audio)
def _supports_native_dtmf(self) -> bool:
"""LiveKit supports native DTMF via telephone events.
Returns:
True, as LiveKit supports native DTMF transmission.
"""
return True
async def _write_dtmf_native(self, frame: OutputDTMFFrame | OutputDTMFUrgentFrame):
"""Use LiveKit's native publish_dtmf method for telephone events.
LiveKit's DTMF API sends a single tone per call, so when
``frame.buttons`` contains multiple entries only the first one is
sent.
Args:
frame: The DTMF frame to write.
"""
if not frame.buttons:
return
await self._client.send_dtmf(frame.buttons[0].value)
def _convert_pipecat_audio_to_livekit(self, pipecat_audio: bytes) -> rtc.AudioFrame:
"""Convert Pipecat audio data to LiveKit audio frame."""
bytes_per_sample = 2 # Assuming 16-bit audio
total_samples = len(pipecat_audio) // bytes_per_sample
samples_per_channel = total_samples // self._params.audio_out_channels
return rtc.AudioFrame(
data=pipecat_audio,
sample_rate=self.sample_rate,
num_channels=self._params.audio_out_channels,
samples_per_channel=samples_per_channel,
)
[docs]
class LiveKitTransport(BaseTransport):
"""Transport implementation for LiveKit real-time communication.
Provides comprehensive LiveKit integration including audio streaming, data
messaging, participant management, and room event handling for conversational
AI applications.
Event handlers available:
- on_connected: Called when the bot connects to the room.
- on_disconnected: Called when the bot disconnects from the room.
- on_before_disconnect: [sync] Called just before the bot disconnects.
- on_call_state_updated: Called when the call state changes. Args: (state: str)
- on_first_participant_joined: Called when the first participant joins.
Args: (participant_id: str)
- on_participant_connected: Called when a participant connects.
Args: (participant_id: str)
- on_participant_disconnected: Called when a participant disconnects.
Args: (participant_id: str)
- on_participant_left: Called when a participant leaves.
Args: (participant_id: str, reason: str)
- on_audio_track_subscribed: Called when an audio track is subscribed.
Args: (participant_id: str)
- on_audio_track_unsubscribed: Called when an audio track is unsubscribed.
Args: (participant_id: str)
- on_video_track_subscribed: Called when a video track is subscribed.
Args: (participant_id: str)
- on_video_track_unsubscribed: Called when a video track is unsubscribed.
Args: (participant_id: str)
- on_data_received: Called when data is received from a participant.
Args: (data: bytes, participant_id: str)
Example::
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant_id):
await task.queue_frame(TTSSpeakFrame("Hello!"))
@transport.event_handler("on_participant_disconnected")
async def on_participant_disconnected(transport, participant_id):
await task.queue_frame(EndFrame())
"""
[docs]
def __init__(
self,
url: str,
token: str,
room_name: str,
params: LiveKitParams | None = None,
input_name: str | None = None,
output_name: str | None = None,
):
"""Initialize the LiveKit transport.
Args:
url: LiveKit server URL to connect to.
token: Authentication token for the room.
room_name: Name of the LiveKit room to join.
params: Configuration parameters for the transport.
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)
callbacks = LiveKitCallbacks(
on_connected=self._on_connected,
on_disconnected=self._on_disconnected,
on_before_disconnect=self._on_before_disconnect,
on_participant_connected=self._on_participant_connected,
on_participant_disconnected=self._on_participant_disconnected,
on_audio_track_subscribed=self._on_audio_track_subscribed,
on_audio_track_unsubscribed=self._on_audio_track_unsubscribed,
on_video_track_subscribed=self._on_video_track_subscribed,
on_video_track_unsubscribed=self._on_video_track_unsubscribed,
on_data_received=self._on_data_received,
on_first_participant_joined=self._on_first_participant_joined,
)
self._params = params or LiveKitParams()
self._client = LiveKitTransportClient(
url, token, room_name, self._params, callbacks, self.name
)
self._input: LiveKitInputTransport | None = None
self._output: LiveKitOutputTransport | None = None
self._register_event_handler("on_connected")
self._register_event_handler("on_disconnected")
self._register_event_handler("on_participant_connected")
self._register_event_handler("on_participant_disconnected")
self._register_event_handler("on_audio_track_subscribed")
self._register_event_handler("on_audio_track_unsubscribed")
self._register_event_handler("on_video_track_subscribed")
self._register_event_handler("on_video_track_unsubscribed")
self._register_event_handler("on_data_received")
self._register_event_handler("on_first_participant_joined")
self._register_event_handler("on_participant_left")
self._register_event_handler("on_call_state_updated")
self._register_event_handler("on_before_disconnect", sync=True)
[docs]
def output(self) -> LiveKitOutputTransport:
"""Get the output transport for sending media and events.
Returns:
The LiveKit output transport instance.
"""
if not self._output:
self._output = LiveKitOutputTransport(
self, self._client, self._params, name=self._output_name
)
return self._output
@property
def participant_id(self) -> str:
"""Get the participant ID for this transport.
Returns:
The participant ID assigned by LiveKit.
"""
return self._client.participant_id
[docs]
async def send_audio(self, frame: OutputAudioRawFrame):
"""Send an audio frame to the LiveKit room.
Args:
frame: The audio frame to send.
"""
if self._output:
await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM)
[docs]
def get_participants(self) -> list[str]:
"""Get list of participant IDs in the room.
Returns:
List of participant IDs.
"""
return self._client.get_participants()
[docs]
async def mute_participant(self, participant_id: str):
"""Mute a specific participant's audio tracks.
Args:
participant_id: ID of the participant to mute.
"""
await self._client.mute_participant(participant_id)
[docs]
async def unmute_participant(self, participant_id: str):
"""Unmute a specific participant's audio tracks.
Args:
participant_id: ID of the participant to unmute.
"""
await self._client.unmute_participant(participant_id)
async def _on_connected(self):
"""Handle room connected events."""
await self._call_event_handler("on_connected")
if self._input:
await self._input.push_frame(BotConnectedFrame())
async def _on_disconnected(self):
"""Handle room disconnected events."""
await self._call_event_handler("on_disconnected")
async def _on_before_disconnect(self):
"""Handle before disconnection room events."""
await self._call_event_handler("on_before_disconnect")
async def _on_participant_connected(self, participant_id: str):
"""Handle participant connected events."""
await self._call_event_handler("on_participant_connected", participant_id)
if self._input:
await self._input.push_frame(ClientConnectedFrame())
async def _on_participant_disconnected(self, participant_id: str):
"""Handle participant disconnected events."""
await self._call_event_handler("on_participant_disconnected", participant_id)
await self._call_event_handler("on_participant_left", participant_id, "disconnected")
async def _on_audio_track_subscribed(self, participant_id: str):
"""Handle audio track subscribed events."""
await self._call_event_handler("on_audio_track_subscribed", participant_id)
participant = self._client.room.remote_participants.get(participant_id)
if participant:
for publication in participant.audio_tracks.values():
self._client._on_track_subscribed_wrapper(
publication.track, publication, participant
)
async def _on_audio_track_unsubscribed(self, participant_id: str):
"""Handle audio track unsubscribed events."""
await self._call_event_handler("on_audio_track_unsubscribed", participant_id)
async def _on_video_track_subscribed(self, participant_id: str):
"""Handle video track subscribed events."""
await self._call_event_handler("on_video_track_subscribed", participant_id)
participant = self._client.room.remote_participants.get(participant_id)
if participant:
for publication in participant.video_tracks.values():
self._client._on_track_subscribed_wrapper(
publication.track, publication, participant
)
async def _on_video_track_unsubscribed(self, participant_id: str):
"""Handle video track unsubscribed events."""
await self._call_event_handler("on_video_track_unsubscribed", participant_id)
async def _on_data_received(self, data: bytes, participant_id: str):
"""Handle data received events."""
if self._input:
await self._input.push_app_message(data.decode(), participant_id)
await self._call_event_handler("on_data_received", data, participant_id)
[docs]
async def send_message(self, message: str, participant_id: str | None = None):
"""Send a message to participants in the room.
Args:
message: The message string to send.
participant_id: Optional specific participant to send to.
"""
if self._output:
frame = LiveKitOutputTransportMessageFrame(
message=message, participant_id=participant_id
)
await self._output.send_message(frame)
[docs]
async def send_message_urgent(self, message: str, participant_id: str | None = None):
"""Send an urgent message to participants in the room.
Args:
message: The urgent message string to send.
participant_id: Optional specific participant to send to.
"""
if self._output:
frame = LiveKitOutputTransportMessageUrgentFrame(
message=message, participant_id=participant_id
)
await self._output.send_message(frame)
[docs]
async def on_room_event(self, event):
"""Handle room events.
Args:
event: The room event to handle.
"""
# Handle room events
pass
[docs]
async def on_participant_event(self, event):
"""Handle participant events.
Args:
event: The participant event to handle.
"""
# Handle participant events
pass
[docs]
async def on_track_event(self, event):
"""Handle track events.
Args:
event: The track event to handle.
"""
# Handle track events
pass
async def _on_call_state_updated(self, state: str):
"""Handle call state update events."""
await self._call_event_handler("on_call_state_updated", state)
async def _on_first_participant_joined(self, participant_id: str):
"""Handle first participant joined events."""
await self._call_event_handler("on_first_participant_joined", participant_id)