Add LiveKit to runner

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.

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`.

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.

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.
 

src/pipecat/runner/run.py

@@ -56,15 +56,16 @@ 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
 
 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,13 +161,15 @@ 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"),
     "moq": ("moq", "cryptography"),
 }
 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. "

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
     """