diff --git a/changelog/5297.added.md b/changelog/5297.added.md new file mode 100644 index 0000000000..5b94e0d45e --- /dev/null +++ b/changelog/5297.added.md @@ -0,0 +1 @@ +- Added LiveKit as a transport option in the development runner: `python bot.py -t livekit`, and `POST /start` support (`"transport": "livekit"`). Requires `LIVEKIT_URL`, `LIVEKIT_API_KEY`, and `LIVEKIT_API_SECRET` to be configured on the server. diff --git a/changelog/5297.fixed.2.md b/changelog/5297.fixed.2.md new file mode 100644 index 0000000000..1e0da85995 --- /dev/null +++ b/changelog/5297.fixed.2.md @@ -0,0 +1 @@ +- Fixed `LiveKitTransport` identifying participants by LiveKit's `sid` (a per-connection session id) everywhere it surfaces a `participant_id` — event handlers, `LiveKitInputTransportMessageFrame`, `get_participants()` — while `get_participant_metadata()`, `mute_participant()`, and `unmute_participant()` look the id up in `room.remote_participants`, which LiveKit keys by `identity` instead. The id `get_participants()`/events handed out could never be fed into those three lookup methods, so `get_participant_metadata()` silently returned `{}` and `mute_participant()`/`unmute_participant()` silently did nothing. `participant_id` is now consistently the participant's LiveKit identity throughout. Those three methods also referenced `is_speaking` and `tracks`, attributes the current `livekit` SDK no longer has (`track_publications` replaces `tracks`); `get_participant_metadata()` no longer includes `is_speaking`, and muting now unsubscribes from the participant's audio track via `track_publications`. diff --git a/changelog/5297.fixed.md b/changelog/5297.fixed.md new file mode 100644 index 0000000000..a3b7eff5c4 --- /dev/null +++ b/changelog/5297.fixed.md @@ -0,0 +1 @@ +- Fixed `LiveKitTransport` never delivering client messages (including RTVI's `client-ready` handshake) to the pipeline. Incoming data-channel messages were wrapped in an output-message frame and pushed downstream only, so `RTVIProcessor` never saw them — instead, the output transport picked the misrouted frame back up and echoed it straight back out to the room. Messages are now parsed and broadcast as `InputTransportMessageFrame` in both directions, matching Daily and SmallWebRTC, so RTVI-based bots using LiveKit now complete the client-ready/bot-ready handshake and receive client messages correctly. Non-JSON or non-object data on the channel is ignored rather than raising, and still fires `on_data_received` for backwards compatibility. diff --git a/src/pipecat/runner/livekit.py b/src/pipecat/runner/livekit.py index 1be30f3d08..fc24007860 100644 --- a/src/pipecat/runner/livekit.py +++ b/src/pipecat/runner/livekit.py @@ -84,6 +84,49 @@ def generate_token_with_agent( return token.to_jwt() +def livekit_credentials() -> tuple[str, str, str]: + """Return ``(url, api_key, api_secret)`` from the environment. + + Raises: + Exception: If ``LIVEKIT_URL``, ``LIVEKIT_API_KEY``, or ``LIVEKIT_API_SECRET`` + is not set in the environment. + """ + url = os.getenv("LIVEKIT_URL") + api_key = os.getenv("LIVEKIT_API_KEY") + api_secret = os.getenv("LIVEKIT_API_SECRET") + if not url or not api_key or not api_secret: + raise Exception( + "LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must be set in environment variables." + ) + return url, api_key, api_secret + + +def generate_session_tokens( + room_name: str, session_id: str, api_key: str, api_secret: str +) -> tuple[str, str]: + """Mint distinct agent/user tokens for a session. + + Identities are suffixed with the session id so concurrent sessions sharing + a room (e.g. a fixed ``LIVEKIT_ROOM_NAME``) don't collide — LiveKit evicts + the earlier connection when a new one joins with the same identity. + + Args: + room_name: Name of the LiveKit room. + session_id: Unique id for this session; used to suffix participant identities. + api_key: LiveKit API key. + api_secret: LiveKit API secret. + + Returns: + ``(agent_token, user_token)``, for the bot and the caller respectively. + """ + suffix = session_id[:8] + agent_token = generate_token_with_agent( + room_name, f"Pipecat Agent-{suffix}", api_key, api_secret + ) + user_token = generate_token(room_name, f"User-{suffix}", api_key, api_secret) + return agent_token, user_token + + async def configure(): """Configure LiveKit room URL and token from arguments or environment. diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index d0130a17d6..5288365b1c 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -56,6 +56,7 @@ async def bot(runner_args: RunnerArguments): Supported transports: - Daily - Creates rooms and tokens, runs bot as participant +- LiveKit - Mints room tokens, runs bot as participant - MOQ - Media over QUIC, connects to a MOQ relay for pub/sub streaming - Telephony - Handles webhook and WebSocket connections for Twilio, Telnyx, Plivo, Exotel - WebRTC - Provides local WebRTC interface with prebuilt UI @@ -63,8 +64,8 @@ async def bot(runner_args: RunnerArguments): The ``/start`` endpoint accepts:: { - "transport": "webrtc", // "webrtc" | "daily" | "twilio" | "telnyx" | - // "plivo" | "exotel" — default: "webrtc" + "transport": "webrtc", // "webrtc" | "daily" | "livekit" | "twilio" | + // "telnyx" | "plivo" | "exotel" — default: "webrtc" // WebRTC-specific "enableDefaultIceServers": false, @@ -84,6 +85,7 @@ async def bot(runner_args: RunnerArguments): - Daily (direct, testing only): ``python bot.py -d`` - ESP32: ``python bot.py -t webrtc --esp32 --host 192.168.1.100`` - Exotel: ``python bot.py -t exotel`` (no proxy needed, but ngrok connection to HTTP 7860 is required) +- LiveKit only: ``python bot.py -t livekit`` - MOQ (bot is the server, local dev): ``python bot.py -t moq`` (serve mode and ``--moq-tls-generate localhost`` are the defaults) - MOQ (bot and browser both dial a relay): ``python bot.py -t moq --moq-connect @@ -127,6 +129,7 @@ async def bot(runner_args: RunnerArguments): from pipecat.runner.types import ( DailyRunnerArguments, EvalRunnerArguments, + LiveKitRunnerArguments, MOQRunnerArguments, RunnerArguments, SmallWebRTCRunnerArguments, @@ -158,6 +161,7 @@ async def bot(runner_args: RunnerArguments): TELEPHONY_TRANSPORTS = ["twilio", "telnyx", "plivo", "exotel"] TRANSPORT_ROUTE_DEPENDENCIES = { "daily": ("daily",), + "livekit": ("livekit.api",), "webrtc": ("aiortc",), "telephony": ("fastapi", "websockets"), "websocket": ("fastapi", "websockets"), @@ -165,6 +169,7 @@ async def bot(runner_args: RunnerArguments): } TRANSPORT_INSTALL_HINTS = { "daily": "install pipecat-ai[daily]", + "livekit": "install pipecat-ai[livekit]", "webrtc": "install pipecat-ai[webrtc]", "telephony": "install pipecat-ai[websocket]", "websocket": "install pipecat-ai[websocket]", @@ -260,7 +265,7 @@ def _runner_url(args: argparse.Namespace) -> str: def _transport_status_lists() -> tuple[list[str], list[str]]: """Return enabled and disabled transport labels for the startup banner.""" - transports = ["daily", "webrtc", "telephony", "websocket", "moq"] + transports = ["daily", "livekit", "webrtc", "telephony", "websocket", "moq"] enabled = [] disabled = [] @@ -436,6 +441,12 @@ def _print_startup_message(args: argparse.Namespace): f"http://{args.host}:{args.port}/daily-dialin-webhook" ) print(" → Configure this URL in your Daily phone number settings") + elif args.transport == "livekit": + print("🚀 Bot ready! (LiveKit)") + if not _transport_routes_enabled("livekit"): + print(f" → LiveKit disabled ({TRANSPORT_INSTALL_HINTS['livekit']})") + else: + print(f" → Open: {_runner_url(args)}") elif args.transport in TELEPHONY_TRANSPORTS: print(f"🚀 Bot ready! ({args.transport.capitalize()})") if not _transport_routes_enabled(args.transport): @@ -636,7 +647,7 @@ def _setup_unified_start_route( When ``-t`` was passed on the command line, requests for any other transport are rejected with HTTP 400. """ - ALL_TRANSPORTS = ["webrtc", "daily", *TELEPHONY_TRANSPORTS, "websocket", "moq"] + ALL_TRANSPORTS = ["webrtc", "daily", "livekit", *TELEPHONY_TRANSPORTS, "websocket", "moq"] @app.get("/status") async def status(): @@ -655,6 +666,7 @@ class StartBotResult(TypedDict, total=False): iceConfig: IceConfig | None dailyRoom: str | None dailyToken: str | None + url: str | None wsUrl: str | None token: str | None # MoQ-specific. Carries everything the browser needs to construct @@ -668,8 +680,8 @@ async def start_agent(request: Request): Accepts:: { - "transport": "webrtc", // "webrtc" | "daily" | "twilio" | "telnyx" | - // "plivo" | "exotel" — default: "webrtc" + "transport": "webrtc", // "webrtc" | "daily" | "livekit" | "twilio" | + // "telnyx" | "plivo" | "exotel" — default: "webrtc" // WebRTC-specific "enableDefaultIceServers": false, @@ -792,6 +804,35 @@ async def start_agent(request: Request): _start_bot_session(bot_module.bot(runner_args)) return result + elif transport == "livekit": + from pipecat.runner.livekit import generate_session_tokens, livekit_credentials + + livekit_url, api_key, api_secret = livekit_credentials() + + body = request_data.get("body", {}) + room_name = os.getenv("LIVEKIT_ROOM_NAME") or f"pipecat-{uuid.uuid4().hex[:8]}" + session_id = str(uuid.uuid4()) + agent_token, user_token = generate_session_tokens( + room_name, session_id, api_key, api_secret + ) + + bot_module = _get_bot_module() + runner_args = LiveKitRunnerArguments( + room_name=room_name, + url=livekit_url, + token=agent_token, + body=body, + session_id=session_id, + ) + runner_args.cli_args = args + _start_bot_session(bot_module.bot(runner_args)) + + return StartBotResult( + url=livekit_url, + token=user_token, + sessionId=session_id, + ) + elif transport in TELEPHONY_TRANSPORTS: # Telephony: the bot starts when the provider connects to /ws. # Return the WebSocket URL so the caller knows where to point their provider. @@ -1548,7 +1589,7 @@ def main(parser: argparse.ArgumentParser | None = None): - --host: Server host address (default: localhost) - --port: Server port (default: 7860) - -t/--transport: Restrict to a single transport and set as default for /start - (daily, webrtc, websocket, twilio, telnyx, plivo, exotel). Omit to support + (daily, livekit, webrtc, websocket, twilio, telnyx, plivo, exotel). Omit to support all transports. - -x/--proxy: Public proxy hostname for telephony webhooks - -d/--direct: Connect directly to Daily room (automatically sets transport to daily) @@ -1576,7 +1617,16 @@ def main(parser: argparse.ArgumentParser | None = None): "-t", "--transport", type=str, - choices=["daily", "eval", "moq", "vonage", "webrtc", "websocket", *TELEPHONY_TRANSPORTS], + choices=[ + "daily", + "eval", + "livekit", + "moq", + "vonage", + "webrtc", + "websocket", + *TELEPHONY_TRANSPORTS, + ], default=None, help=( "Restrict the server to a single transport and set it as the default for /start. " diff --git a/src/pipecat/runner/types.py b/src/pipecat/runner/types.py index 1054d690ed..b0319d588c 100644 --- a/src/pipecat/runner/types.py +++ b/src/pipecat/runner/types.py @@ -238,6 +238,7 @@ class LiveKitRunnerArguments(RunnerArguments): Parameters: room_name: LiveKit room name to join + url: LiveKit server URL to connect to token: Authentication token for the room body: Additional request data """ diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index 7bbf24a570..edf339a987 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -31,6 +31,7 @@ Frame, ImageRawFrame, InputDTMFFrame, + InputTransportMessageFrame, InterruptionFrame, OutputAudioRawFrame, OutputDTMFFrame, @@ -73,6 +74,17 @@ } +@dataclass +class LiveKitInputTransportMessageFrame(InputTransportMessageFrame): + """Frame for incoming transport messages from LiveKit rooms. + + Parameters: + participant_id: ID of the participant this message is from. + """ + + participant_id: str | None = None + + @dataclass class LiveKitOutputTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in LiveKit rooms. @@ -264,8 +276,8 @@ async def connect(self): # 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}") + self._participant_id = self.room.local_participant.identity + logger.info(f"Connected to {self._room_name} as {self._participant_id}") # Set up audio source and track self._audio_source = rtc.AudioSource( @@ -377,15 +389,15 @@ def get_participants(self) -> list[str]: """Get list of participant IDs in the room. Returns: - List of participant IDs. + List of participant LiveKit identities. """ - return [p.sid for p in self.room.remote_participants.values()] + return [p.identity for p in self.room.remote_participants.values()] async def get_participant_metadata(self, participant_id: str) -> dict: """Get metadata for a specific participant. Args: - participant_id: ID of the participant to get metadata for. + participant_id: LiveKit identity of the participant to get metadata for. Returns: Dictionary containing participant metadata. @@ -393,11 +405,9 @@ async def get_participant_metadata(self, participant_id: str) -> dict: participant = self.room.remote_participants.get(participant_id) if participant: return { - "id": participant.sid, + "id": participant.identity, "name": participant.name, "metadata": participant.metadata, - # TODO: not a LiveKit participant attribute; this raises if reached. - "is_speaking": participant.is_speaking, # pyright: ignore[reportAttributeAccessIssue] } return {} @@ -410,30 +420,33 @@ async def set_participant_metadata(self, metadata: str): await self.room.local_participant.set_metadata(metadata) async def mute_participant(self, participant_id: str): - """Mute a specific participant's audio tracks. + """Stop receiving a specific participant's audio. + + LiveKit doesn't let one participant force-mute another's microphone; + this unsubscribes the bot from their audio track instead. Args: - participant_id: ID of the participant to mute. + participant_id: LiveKit identity of the participant to stop + receiving audio from. """ participant = self.room.remote_participants.get(participant_id) if participant: - # TODO: not a LiveKit participant attribute; this raises if reached. - for track in participant.tracks.values(): # pyright: ignore[reportAttributeAccessIssue] - if track.kind == "audio": - await track.set_enabled(False) + for publication in participant.track_publications.values(): + if publication.kind == rtc.TrackKind.KIND_AUDIO: + publication.set_subscribed(False) async def unmute_participant(self, participant_id: str): - """Unmute a specific participant's audio tracks. + """Resume receiving a specific participant's audio. Args: - participant_id: ID of the participant to unmute. + participant_id: LiveKit identity of the participant to resume + receiving audio from. """ participant = self.room.remote_participants.get(participant_id) if participant: - # TODO: not a LiveKit participant attribute; this raises if reached. - for track in participant.tracks.values(): # pyright: ignore[reportAttributeAccessIssue] - if track.kind == "audio": - await track.set_enabled(True) + for publication in participant.track_publications.values(): + if publication.kind == rtc.TrackKind.KIND_AUDIO: + publication.set_subscribed(True) # Wrapper methods for event handlers def _on_participant_connected_wrapper(self, participant: rtc.RemoteParticipant): @@ -518,15 +531,15 @@ def _on_sip_dtmf_received_wrapper(self, dtmf: rtc.SipDTMF): 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) + await self._callbacks.on_participant_connected(participant.identity) if not self._other_participant_has_joined: self._other_participant_has_joined = True - await self._callbacks.on_first_participant_joined(participant.sid) + await self._callbacks.on_first_participant_joined(participant.identity) 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) + await self._callbacks.on_participant_disconnected(participant.identity) if len(self.get_participants()) == 0: self._other_participant_has_joined = False @@ -540,36 +553,40 @@ async def _async_on_track_subscribed( assert self._task_manager is not None if track.kind == rtc.TrackKind.KIND_AUDIO: - logger.info(f"Audio track subscribed: {track.sid} from participant {participant.sid}") + logger.info( + f"Audio track subscribed: {track.sid} from participant {participant.identity}" + ) # If the participant is re-publishing (e.g. mute/unmute cycle), # close + cancel the previous stream/task before replacing the # registry entry, so two producers never feed ``_audio_queue`` # for the same participant. - await self._close_audio_stream(participant.sid) - self._audio_tracks[participant.sid] = track + await self._close_audio_stream(participant.identity) + self._audio_tracks[participant.identity] = track audio_stream = rtc.AudioStream(track) task = self._task_manager.create_task( - self._process_audio_stream(audio_stream, participant.sid), + self._process_audio_stream(audio_stream, participant.identity), f"{self}::_process_audio_stream", ) - self._audio_streams[participant.sid] = (audio_stream, task) - await self._callbacks.on_audio_track_subscribed(participant.sid) + self._audio_streams[participant.identity] = (audio_stream, task) + await self._callbacks.on_audio_track_subscribed(participant.identity) elif track.kind == rtc.TrackKind.KIND_VIDEO: - logger.info(f"Video track subscribed: {track.sid} from participant {participant.sid}") + logger.info( + f"Video track subscribed: {track.sid} from participant {participant.identity}" + ) # Symmetric: clean up any prior video stream/task for the same # participant before replacing. - await self._close_video_stream(participant.sid) - self._video_tracks[participant.sid] = track + await self._close_video_stream(participant.identity) + self._video_tracks[participant.identity] = 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) task = self._task_manager.create_task( - self._process_video_stream(video_stream, participant.sid), + self._process_video_stream(video_stream, participant.identity), f"{self}::_process_video_stream", ) - self._video_streams[participant.sid] = (video_stream, task) - await self._callbacks.on_video_track_subscribed(participant.sid) + self._video_streams[participant.identity] = (video_stream, task) + await self._callbacks.on_video_track_subscribed(participant.identity) async def _async_on_track_unsubscribed( self, @@ -580,11 +597,11 @@ async def _async_on_track_unsubscribed( """Handle track unsubscribed events.""" logger.info(f"Track unsubscribed: {publication.sid} from {participant.identity}") if track.kind == rtc.TrackKind.KIND_AUDIO: - await self._close_audio_stream(participant.sid) - await self._callbacks.on_audio_track_unsubscribed(participant.sid) + await self._close_audio_stream(participant.identity) + await self._callbacks.on_audio_track_unsubscribed(participant.identity) elif track.kind == rtc.TrackKind.KIND_VIDEO: - await self._close_video_stream(participant.sid) - await self._callbacks.on_video_track_unsubscribed(participant.sid) + await self._close_video_stream(participant.identity) + await self._callbacks.on_video_track_unsubscribed(participant.identity) async def _close_audio_stream(self, participant_id: str) -> None: """Close a participant's owned audio stream and cancel its producer task. @@ -633,7 +650,7 @@ async def _close_all_streams(self) -> None: async def _async_on_data_received(self, data: rtc.DataPacket): """Handle data received events.""" # LiveKit delivers packets sent by a server SDK with no participant. - sender = data.participant.sid if data.participant else None + sender = data.participant.identity if data.participant else None await self._callbacks.on_data_received(data.data, sender) async def _async_on_connected(self): @@ -649,7 +666,7 @@ async def _async_on_disconnected(self, reason=None): async def _async_on_sip_dtmf_received(self, dtmf: rtc.SipDTMF): """Handle inbound SIP DTMF events from LiveKit telephony.""" participant = getattr(dtmf, "participant", None) - participant_id = getattr(participant, "sid", None) if participant else None + participant_id = getattr(participant, "identity", None) if participant else None data = { "tone": dtmf.digit, "digit": dtmf.digit, @@ -800,13 +817,17 @@ async def _teardown(self): async def push_app_message(self, message: Any, sender: str | None): """Push an application message as an urgent transport frame. + Broadcast both upstream and downstream so it reaches ``RTVIProcessor`` + regardless of where it sits relative to the transport in the pipeline. + Args: message: The message data to send. sender: ID of the message sender, or None if it was sent unattributed by a server SDK. """ - frame = LiveKitOutputTransportMessageUrgentFrame(message=message, participant_id=sender) - await self.push_frame(frame) + await self.broadcast_frame( + LiveKitInputTransportMessageFrame, message=message, participant_id=sender + ) async def _audio_in_task_handler(self): """Handle incoming audio frames from participants.""" @@ -1054,6 +1075,15 @@ class LiveKitTransport(BaseTransport): messaging, participant management, and room event handling for conversational AI applications. + Every ``participant_id`` surfaced by this transport (event args, frame + fields, ``get_participants()``, and the ``get_participant_metadata``/ + ``mute_participant``/``unmute_participant`` methods) is the participant's + LiveKit *identity* (``rtc.Participant.identity``) — the value set when + minting its access token, and what LiveKit itself keys + ``room.remote_participants`` by and expects in ``destination_identities``. + It is not the participant's *SID* (``rtc.Participant.sid``), a + per-connection session id that changes on every reconnect. + Event handlers available: - on_connected: Called when the bot connects to the room. @@ -1068,6 +1098,10 @@ class LiveKitTransport(BaseTransport): Args: (participant_id: str) - on_participant_left: Called when a participant leaves. Args: (participant_id: str, reason: str) + - on_client_connected: Called when a participant connects (alias for + on_participant_connected). Args: (participant: dict) + - on_client_disconnected: Called when a participant disconnects (alias for + on_participant_disconnected). Args: (participant: dict) - 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. @@ -1076,6 +1110,8 @@ class LiveKitTransport(BaseTransport): Args: (participant_id: str) - on_video_track_unsubscribed: Called when a video track is unsubscribed. Args: (participant_id: str) + - on_app_message: Called when data is received from a participant. RTVI-compatible version of on_data_received. + Args: (message: Any, sender: str) - on_data_received: Called when data is received. The participant ID is None for packets sent by a server SDK, which LiveKit delivers unattributed. Args: (data: bytes, participant_id: str | None) @@ -1142,10 +1178,13 @@ def __init__( 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_client_connected") + self._register_event_handler("on_client_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_app_message") self._register_event_handler("on_data_received") self._register_event_handler("on_first_participant_joined") self._register_event_handler("on_participant_left") @@ -1199,7 +1238,7 @@ def get_participants(self) -> list[str]: """Get list of participant IDs in the room. Returns: - List of participant IDs. + List of participant LiveKit identities. """ return self._client.get_participants() @@ -1207,7 +1246,7 @@ async def get_participant_metadata(self, participant_id: str) -> dict: """Get metadata for a specific participant. Args: - participant_id: ID of the participant to get metadata for. + participant_id: LiveKit identity of the participant to get metadata for. Returns: Dictionary containing participant metadata. @@ -1223,18 +1262,20 @@ async def set_metadata(self, metadata: str): await self._client.set_participant_metadata(metadata) async def mute_participant(self, participant_id: str): - """Mute a specific participant's audio tracks. + """Stop receiving a specific participant's audio. Args: - participant_id: ID of the participant to mute. + participant_id: LiveKit identity of the participant to stop + receiving audio from. """ await self._client.mute_participant(participant_id) async def unmute_participant(self, participant_id: str): - """Unmute a specific participant's audio tracks. + """Resume receiving a specific participant's audio. Args: - participant_id: ID of the participant to unmute. + participant_id: LiveKit identity of the participant to resume + receiving audio from. """ await self._client.unmute_participant(participant_id) @@ -1255,6 +1296,10 @@ async def _on_before_disconnect(self): async def _on_participant_connected(self, participant_id: str): """Handle participant connected events.""" await self._call_event_handler("on_participant_connected", participant_id) + # Also call on_client_connected for compatibility with other transports. + # Wrapped as a dict (matching Daily's Mapping[str, Any] shape) so + # drop-in bot templates reading client["id"] work across transports. + await self._call_event_handler("on_client_connected", {"id": participant_id}) if self._input: await self._input.push_frame(ClientConnectedFrame()) @@ -1262,17 +1307,12 @@ 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") + # Also call on_client_disconnected for compatibility with other transports + await self._call_event_handler("on_client_disconnected", {"id": participant_id}) 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: - # TODO: not a LiveKit participant attribute; this raises if reached. - for publication in participant.audio_tracks.values(): # pyright: ignore[reportAttributeAccessIssue] - 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.""" @@ -1281,13 +1321,6 @@ async def _on_audio_track_unsubscribed(self, participant_id: str): 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: - # TODO: not a LiveKit participant attribute; this raises if reached. - for publication in participant.video_tracks.values(): # pyright: ignore[reportAttributeAccessIssue] - 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.""" @@ -1295,8 +1328,21 @@ async def _on_video_track_unsubscribed(self, participant_id: str): async def _on_data_received(self, data: bytes, participant_id: str | None): """Handle data received events.""" - if self._input: - await self._input.push_app_message(data.decode(), participant_id) + try: + message = json.loads(data.decode()) + if not isinstance(message, dict): + logger.debug(f"{self} Ignoring non-object JSON data: {message!r}") + message = None + except (UnicodeDecodeError, json.JSONDecodeError) as e: + logger.debug(f"{self} Ignoring non-JSON data from {participant_id}: {e}") + message = None + + if message is not None: + if self._input: + await self._input.push_app_message(message, participant_id) + # RTVI compatibility: + await self._call_event_handler("on_app_message", message, participant_id) + # Backwards compatibility with older transports that used on_data_received for app messages await self._call_event_handler("on_data_received", data, participant_id) async def _on_dtmf_event(self, data: Any): diff --git a/tests/test_livekit_transport.py b/tests/test_livekit_transport.py index 37ece7518e..2055958da5 100644 --- a/tests/test_livekit_transport.py +++ b/tests/test_livekit_transport.py @@ -11,6 +11,7 @@ only starts when there is a consumer for the frames. """ +import json import unittest from unittest.mock import AsyncMock, MagicMock, patch @@ -74,7 +75,7 @@ def _create_mock_video_track(self): track.sid = "video-track-123" publication = MagicMock() participant = MagicMock() - participant.sid = "participant-456" + participant.identity = "participant-456" return track, publication, participant async def test_disabled_video_input_does_not_start_queue_producer(self): @@ -96,7 +97,7 @@ async def test_disabled_video_input_does_not_start_queue_producer(self): self.assertEqual(client._video_queue.qsize(), 0) # Track metadata should still be recorded - self.assertIn(participant.sid, client._video_tracks) + self.assertIn(participant.identity, client._video_tracks) # Callback should still fire for user code client._callbacks.on_video_track_subscribed.assert_called_once() @@ -115,7 +116,7 @@ async def test_enabled_video_input_starts_queue_producer(self): self.assertEqual(len(video_tasks), 1, "Video processing task should be started") # Track metadata should be recorded - self.assertIn(participant.sid, client._video_tracks) + self.assertIn(participant.identity, client._video_tracks) # Callback should fire client._callbacks.on_video_track_subscribed.assert_called_once() @@ -135,10 +136,10 @@ class TestLiveKitAudioStreamLeakOnUnsubscribe(unittest.IsolatedAsyncioTestCase): keeps pushing frames forever; N republishes → N concurrent producers interleave audio into the shared queue and downstream STT receives garbage. - The fix: store ``(stream, task)`` per ``participant.sid`` in + The fix: store ``(stream, task)`` per ``participant.identity`` in ``_audio_streams`` on subscribe, then ``aclose()`` + cancel on unsubscribe - and again on a re-subscribe for the same sid (to handle missed unsubscribe). - Symmetric for video. + and again on a re-subscribe for the same identity (to handle missed + unsubscribe). Symmetric for video. """ def _create_client(self, video_in_enabled: bool = False) -> LiveKitTransportClient: @@ -179,26 +180,24 @@ def _make_task(coro, name): client._task_manager.create_task.side_effect = _make_task return client - def _audio_track(self, sid: str = "audio-track-1", participant_sid: str = "p-1"): + def _audio_track(self, sid: str = "audio-track-1", participant_identity: str = "p-1"): track = MagicMock() track.kind = rtc.TrackKind.KIND_AUDIO track.sid = sid publication = MagicMock() publication.sid = sid participant = MagicMock() - participant.sid = participant_sid - participant.identity = "user" + participant.identity = participant_identity return track, publication, participant - def _video_track(self, sid: str = "video-track-1", participant_sid: str = "p-1"): + def _video_track(self, sid: str = "video-track-1", participant_identity: str = "p-1"): track = MagicMock() track.kind = rtc.TrackKind.KIND_VIDEO track.sid = sid publication = MagicMock() publication.sid = sid participant = MagicMock() - participant.sid = participant_sid - participant.identity = "user" + participant.identity = participant_identity return track, publication, participant async def test_audio_stream_registered_on_subscribe(self): @@ -211,8 +210,8 @@ async def test_audio_stream_registered_on_subscribe(self): with patch.object(rtc, "AudioStream", return_value=mock_stream): await client._async_on_track_subscribed(track, pub, participant) - self.assertIn(participant.sid, client._audio_streams) - stream, task = client._audio_streams[participant.sid] + self.assertIn(participant.identity, client._audio_streams) + stream, task = client._audio_streams[participant.identity] self.assertIs(stream, mock_stream) self.assertIsNotNone(task) @@ -225,13 +224,13 @@ async def test_audio_stream_closed_and_task_cancelled_on_unsubscribe(self): mock_stream.aclose = AsyncMock() with patch.object(rtc, "AudioStream", return_value=mock_stream): await client._async_on_track_subscribed(track, pub, participant) - _, task = client._audio_streams[participant.sid] + _, task = client._audio_streams[participant.identity] await client._async_on_track_unsubscribed(track, pub, participant) mock_stream.aclose.assert_awaited_once() task.cancel.assert_called_once() - self.assertNotIn(participant.sid, client._audio_streams) + self.assertNotIn(participant.identity, client._audio_streams) client._callbacks.on_audio_track_unsubscribed.assert_called_once() async def test_resubscribe_closes_previous_audio_stream(self): @@ -246,7 +245,7 @@ async def test_resubscribe_closes_previous_audio_stream(self): with patch.object(rtc, "AudioStream", return_value=first): await client._async_on_track_subscribed(track, pub, participant) - first_task = client._audio_streams[participant.sid][1] + first_task = client._audio_streams[participant.identity][1] # Republish without an explicit unsubscribe in between. with patch.object(rtc, "AudioStream", return_value=second): @@ -254,7 +253,7 @@ async def test_resubscribe_closes_previous_audio_stream(self): first.aclose.assert_awaited_once() first_task.cancel.assert_called_once() - self.assertIs(client._audio_streams[participant.sid][0], second) + self.assertIs(client._audio_streams[participant.identity][0], second) async def test_unsubscribe_without_subscribe_is_noop(self): """Unsubscribe for an unknown sid does not raise.""" @@ -273,11 +272,11 @@ async def test_video_stream_closed_on_unsubscribe(self): mock_stream.aclose = AsyncMock() with patch.object(rtc, "VideoStream", return_value=mock_stream): await client._async_on_track_subscribed(track, pub, participant) - self.assertIn(participant.sid, client._video_streams) + self.assertIn(participant.identity, client._video_streams) await client._async_on_track_unsubscribed(track, pub, participant) mock_stream.aclose.assert_awaited_once() - self.assertNotIn(participant.sid, client._video_streams) + self.assertNotIn(participant.identity, client._video_streams) @unittest.skipUnless(LIVEKIT_AVAILABLE, "livekit package not installed") @@ -315,7 +314,7 @@ async def test_sip_dtmf_forwards_digit_to_callback(self): """Room sip_dtmf_received events are normalized and forwarded.""" client = self._create_client() participant = MagicMock() - participant.sid = "sip-participant-1" + participant.identity = "sip-participant-1" dtmf = MagicMock() dtmf.digit = "5" dtmf.code = 5 @@ -382,5 +381,280 @@ async def test_transport_ignores_unsupported_dtmf_tone(self): transport._input.push_frame.assert_not_awaited() +@unittest.skipUnless(LIVEKIT_AVAILABLE, "livekit package not installed") +class TestLiveKitAppMessageInput(unittest.IsolatedAsyncioTestCase): + """Inbound JSON data messages (RTVI's wire channel) are parsed and + broadcast both directions as an ``InputTransportMessageFrame`` so + ``RTVIProcessor`` sees them wherever it sits in the pipeline. Non-object + JSON and non-JSON data are not pushed into the pipeline, but still fire + ``on_data_received`` for backwards compatibility. + """ + + def _make_transport(self): + from pipecat.transports.livekit.transport import LiveKitTransport + + transport = LiveKitTransport( + url="wss://test.livekit.cloud", + token="test-token", + room_name="test-room", + ) + input_transport = transport.input() + input_transport.push_frame = AsyncMock() + transport._call_event_handler = AsyncMock() + return transport, input_transport + + async def test_data_received_broadcasts_parsed_input_message_frame(self): + """A JSON data message is parsed and broadcast as an InputTransportMessageFrame.""" + from pipecat.frames.frames import InputTransportMessageFrame + from pipecat.processors.frame_processor import FrameDirection + + transport, input_transport = self._make_transport() + + rtvi_message = {"label": "rtvi-ai", "type": "client-ready", "id": "1", "data": {}} + await transport._on_data_received(json.dumps(rtvi_message).encode(), "participant-1") + + self.assertEqual(input_transport.push_frame.await_count, 2) + directions = set() + for call in input_transport.push_frame.await_args_list: + frame = call.args[0] + direction = call.args[1] if len(call.args) > 1 else FrameDirection.DOWNSTREAM + self.assertIsInstance(frame, InputTransportMessageFrame) + self.assertEqual(frame.message, rtvi_message) + self.assertEqual(frame.participant_id, "participant-1") + directions.add(direction) + # Broadcast both ways so RTVIProcessor sees it regardless of where it + # sits relative to the transport in the pipeline. + self.assertEqual(directions, {FrameDirection.DOWNSTREAM, FrameDirection.UPSTREAM}) + + transport._call_event_handler.assert_any_call( + "on_app_message", rtvi_message, "participant-1" + ) + + async def test_non_json_data_is_not_pushed_but_reported_for_compat(self): + """Non-JSON data doesn't crash, isn't pushed, but still reports on_data_received.""" + transport, input_transport = self._make_transport() + + await transport._on_data_received(b"not json", "participant-1") + + input_transport.push_frame.assert_not_awaited() + transport._call_event_handler.assert_awaited_once_with( + "on_data_received", b"not json", "participant-1" + ) + + async def test_non_object_json_is_not_pushed_but_reported_for_compat(self): + """A JSON value that isn't an object (str/number/bool/list) is ignored. + + ``RTVIProcessor`` calls ``.get("label")`` on the parsed message, so + pushing a non-dict would raise ``AttributeError`` deep in the pipeline. + """ + transport, input_transport = self._make_transport() + + await transport._on_data_received(json.dumps("hi").encode(), "participant-1") + + input_transport.push_frame.assert_not_awaited() + transport._call_event_handler.assert_awaited_once_with( + "on_data_received", json.dumps("hi").encode(), "participant-1" + ) + + +@unittest.skipUnless(LIVEKIT_AVAILABLE, "livekit package not installed") +class TestLiveKitClientConnectedAlias(unittest.IsolatedAsyncioTestCase): + """on_client_connected/on_client_disconnected match Daily's dict shape. + + Drop-in bot templates read ``client["id"]`` off these aliases, so the + payload needs to be a mapping, not a bare participant id string. + """ + + def _make_transport(self): + from pipecat.transports.livekit.transport import LiveKitTransport + + transport = LiveKitTransport( + url="wss://test.livekit.cloud", + token="test-token", + room_name="test-room", + ) + transport._input = MagicMock() + transport._input.push_frame = AsyncMock() + transport._call_event_handler = AsyncMock() + return transport + + async def test_on_client_connected_receives_dict(self): + transport = self._make_transport() + + await transport._on_participant_connected("participant-1") + + transport._call_event_handler.assert_any_call( + "on_client_connected", {"id": "participant-1"} + ) + + async def test_on_client_disconnected_receives_dict(self): + transport = self._make_transport() + + await transport._on_participant_disconnected("participant-1") + + transport._call_event_handler.assert_any_call( + "on_client_disconnected", {"id": "participant-1"} + ) + + +@unittest.skipUnless(LIVEKIT_AVAILABLE, "livekit package not installed") +class TestLiveKitParticipantIdentity(unittest.IsolatedAsyncioTestCase): + """The participant_id this transport hands out must be the LiveKit + identity, not the SID. + + Regression test (pipecat-ai/pipecat#5218): ``room.remote_participants`` + is keyed by identity and ``destination_identities`` expects identities, + but the transport used to emit ``participant.sid`` everywhere. Callers + couldn't feed the ``get_participants()``/event ``participant_id`` back + into ``get_participant_metadata``/``mute_participant``/ + ``unmute_participant``/targeted ``send_message`` — the lookup would + silently fail (``room.remote_participants.get(sid)`` returns ``None``). + """ + + def _create_client(self) -> LiveKitTransportClient: + params = LiveKitParams() + callbacks = LiveKitCallbacks( + on_connected=AsyncMock(), + on_disconnected=AsyncMock(), + on_before_disconnect=AsyncMock(), + on_participant_connected=AsyncMock(), + on_participant_disconnected=AsyncMock(), + on_audio_track_subscribed=AsyncMock(), + on_audio_track_unsubscribed=AsyncMock(), + on_video_track_subscribed=AsyncMock(), + on_video_track_unsubscribed=AsyncMock(), + on_data_received=AsyncMock(), + on_first_participant_joined=AsyncMock(), + on_dtmf_event=AsyncMock(), + ) + client = LiveKitTransportClient( + url="wss://test.livekit.cloud", + token="test-token", + room_name="test-room", + params=params, + callbacks=callbacks, + transport_name="test-transport", + ) + client._task_manager = MagicMock() + return client + + def _mock_room_with_participant( + self, client: LiveKitTransportClient, *, sid: str, identity: str + ): + publication = MagicMock() + publication.kind = rtc.TrackKind.KIND_AUDIO + publication.set_subscribed = MagicMock() + + participant = MagicMock() + participant.sid = sid + participant.identity = identity + participant.name = "Test User" + participant.metadata = "" + participant.track_publications = {"track-1": publication} + + room = MagicMock() + room.remote_participants = {identity: participant} + client._room = room + return participant, publication + + async def test_get_participants_returns_identity_not_sid(self): + client = self._create_client() + participant, _ = self._mock_room_with_participant( + client, sid="PA_serverSid", identity="repro-client" + ) + + self.assertEqual(client.get_participants(), ["repro-client"]) + + async def test_get_participant_metadata_resolves_id_from_get_participants(self): + """The id get_participants() hands out must work as a lookup key.""" + client = self._create_client() + self._mock_room_with_participant(client, sid="PA_serverSid", identity="repro-client") + + (participant_id,) = client.get_participants() + metadata = await client.get_participant_metadata(participant_id) + + self.assertEqual(metadata, {"id": "repro-client", "name": "Test User", "metadata": ""}) + + async def test_mute_participant_resolves_id_from_get_participants(self): + client = self._create_client() + _, publication = self._mock_room_with_participant( + client, sid="PA_serverSid", identity="repro-client" + ) + + (participant_id,) = client.get_participants() + await client.mute_participant(participant_id) + + publication.set_subscribed.assert_called_once_with(False) + + async def test_unmute_participant_resolves_id_from_get_participants(self): + client = self._create_client() + _, publication = self._mock_room_with_participant( + client, sid="PA_serverSid", identity="repro-client" + ) + + (participant_id,) = client.get_participants() + await client.unmute_participant(participant_id) + + publication.set_subscribed.assert_called_once_with(True) + + async def test_participant_connected_callback_receives_identity(self): + client = self._create_client() + participant = MagicMock() + participant.sid = "PA_serverSid" + participant.identity = "repro-client" + + await client._async_on_participant_connected(participant) + + client._callbacks.on_participant_connected.assert_awaited_once_with("repro-client") + + +@unittest.skipUnless(LIVEKIT_AVAILABLE, "livekit package not installed") +class TestLiveKitAudioTrackSubscribedHandler(unittest.TestCase): + """The top-level transport's on_audio/video_track_subscribed handlers + must not re-derive publications from nonexistent SDK attributes. + + Regression test: these used to look up ``participant.audio_tracks``/ + ``participant.video_tracks`` (removed from the SDK; ``track_publications`` + is the only such attribute now) and re-invoke the subscribe wrapper that + had already run for this exact track via the room event, redundantly. + """ + + def test_on_audio_track_subscribed_only_fires_event_handler(self): + import asyncio + + from pipecat.transports.livekit.transport import LiveKitTransport + + transport = LiveKitTransport( + url="wss://test.livekit.cloud", token="test-token", room_name="test-room" + ) + transport._call_event_handler = AsyncMock() + transport._client = MagicMock() + + asyncio.run(transport._on_audio_track_subscribed("participant-1")) + + transport._call_event_handler.assert_awaited_once_with( + "on_audio_track_subscribed", "participant-1" + ) + transport._client.room.remote_participants.get.assert_not_called() + + def test_on_video_track_subscribed_only_fires_event_handler(self): + import asyncio + + from pipecat.transports.livekit.transport import LiveKitTransport + + transport = LiveKitTransport( + url="wss://test.livekit.cloud", token="test-token", room_name="test-room" + ) + transport._call_event_handler = AsyncMock() + transport._client = MagicMock() + + asyncio.run(transport._on_video_track_subscribed("participant-1")) + + transport._call_event_handler.assert_awaited_once_with( + "on_video_track_subscribed", "participant-1" + ) + transport._client.room.remote_participants.get.assert_not_called() + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_runner_run.py b/tests/test_runner_run.py index 7ba72283ec..3ee9290b99 100644 --- a/tests/test_runner_run.py +++ b/tests/test_runner_run.py @@ -31,6 +31,14 @@ _verify_and_consume_ws_token, ) +try: + import jwt as _jwt # noqa: F401 + from livekit import api as _livekit_api # noqa: F401 + + LIVEKIT_AVAILABLE = True +except ImportError: + LIVEKIT_AVAILABLE = False + class TestRunnerRun(unittest.TestCase): def _capture_startup_message(self, args: argparse.Namespace) -> str: @@ -49,6 +57,7 @@ def test_transport_route_dependencies_maps_transports_to_modules(self): self.assertEqual(_transport_route_dependencies("plivo"), ("fastapi", "websockets")) self.assertEqual(_transport_route_dependencies("exotel"), ("fastapi", "websockets")) self.assertEqual(_transport_route_dependencies("vonage"), ()) + self.assertEqual(_transport_route_dependencies("livekit"), ("livekit.api",)) def test_transport_routes_enabled_maps_transports_to_dependency_checks(self): def module_available(module: str) -> bool: @@ -271,6 +280,7 @@ def routes_enabled(transport: str) -> bool: " → Open: http://localhost:7860\n" " → Enabled transports: telephony, websocket\n" " → Disabled transports: daily (install pipecat-ai[daily]), " + "livekit (install pipecat-ai[livekit]), " "webrtc (install pipecat-ai[webrtc]), " "moq (install pipecat-ai[moq])\n" " → Allowed origins: all (no restriction)\n" @@ -292,7 +302,7 @@ def test_startup_message_all_transports_omits_disabled_status_when_all_enabled(s "\n" "🚀 Bot ready!\n" " → Open: http://localhost:7860\n" - " → Enabled transports: daily, webrtc, telephony, websocket, moq\n" + " → Enabled transports: daily, livekit, webrtc, telephony, websocket, moq\n" " → Allowed origins: all (no restriction)\n" "\n" ), @@ -336,6 +346,58 @@ def test_startup_message_telephony_keeps_provider_endpoint_details(self): self.assertIn(" → WebSocket: ws://localhost:7860/ws\n", output) +@unittest.skipUnless(LIVEKIT_AVAILABLE, "livekit package not installed") +class TestLiveKitRunnerRoutes(unittest.TestCase): + def test_livekit_credentials_raises_when_unconfigured(self): + from pipecat.runner.livekit import livekit_credentials + + with patch.dict("os.environ", {}, clear=True): + with self.assertRaisesRegex(Exception, "LIVEKIT_URL"): + livekit_credentials() + + def test_livekit_credentials_returns_configured_values(self): + from pipecat.runner.livekit import livekit_credentials + + env = { + "LIVEKIT_URL": "wss://test.livekit.cloud", + "LIVEKIT_API_KEY": "key", + "LIVEKIT_API_SECRET": "secret", + } + with patch.dict("os.environ", env, clear=True): + url, api_key, api_secret = livekit_credentials() + self.assertEqual((url, api_key, api_secret), (env["LIVEKIT_URL"], "key", "secret")) + + def test_generate_session_tokens_suffixes_identity_with_session_id(self): + """Regression: fixed identities collide across concurrent sessions sharing + a room (e.g. a fixed LIVEKIT_ROOM_NAME), evicting the earlier participant. + """ + import uuid + + import jwt + + from pipecat.runner.livekit import generate_session_tokens + + agent_token_1, user_token_1 = generate_session_tokens( + "shared-room", str(uuid.uuid4()), "key", "secret" + ) + agent_token_2, user_token_2 = generate_session_tokens( + "shared-room", str(uuid.uuid4()), "key", "secret" + ) + + def identity(token: str) -> str: + return jwt.decode(token, options={"verify_signature": False})["sub"] + + identities = { + identity(agent_token_1), + identity(user_token_1), + identity(agent_token_2), + identity(user_token_2), + } + self.assertEqual( + len(identities), 4, "every session/role pair should have a unique identity" + ) + + class TestWsAuthTokens(unittest.TestCase): """Unit tests for the HMAC WebSocket session token helpers."""