consistent error class ServerError, test suite

livekit-api/livekit/api/__init__.py

@@ -36,7 +36,13 @@
 from livekit.protocol.connector_whatsapp import *
 from livekit.protocol.connector_twilio import *
 
-from .twirp_client import TwirpError, TwirpErrorCode
+from .twirp_client import (
+    ServerError,
+    ServerErrorCode,
+    SipCallError,
+    TwirpError,
+    TwirpErrorCode,
+)
 from .livekit_api import LiveKitAPI
 from .access_token import (
     InferenceGrants,
@@ -64,6 +70,9 @@
     "AccessToken",
     "TokenVerifier",
     "WebhookReceiver",
+    "ServerError",
+    "ServerErrorCode",
     "TwirpError",
     "TwirpErrorCode",
+    "SipCallError",
 ]

livekit-api/livekit/api/_dial_timeout.py

@@ -2,15 +2,12 @@
 
 from typing import Optional, Union
 
-from livekit.protocol.connector_whatsapp import AcceptWhatsAppCallRequest
 from livekit.protocol.sip import CreateSIPParticipantRequest, TransferSIPParticipantRequest
 
-# Requests that carry wait_until_answered / ringing_timeout and share the
-# phone-dialing timeout behavior.
+# Requests that carry ringing_timeout and share the phone-dialing timeout behavior.
 DialRequest = Union[
     CreateSIPParticipantRequest,
     TransferSIPParticipantRequest,
-    AcceptWhatsAppCallRequest,
 ]
 """@private"""
 

livekit-api/livekit/api/_service.py

@@ -20,18 +20,22 @@ def __init__(
         self._client = TwirpClient(session, host, "livekit", failover=failover)
         self.api_key = api_key
         self.api_secret = api_secret
+        # A pre-signed token set by LiveKitAPI for token auth; sent verbatim,
+        # skipping per-call signing. Per-service constructors stay key/secret-only.
+        self._token: str | None = None
 
     def _auth_header(
         self, grants: VideoGrants | None, sip: SIPGrants | None = None
     ) -> dict[str, str]:
+        # A pre-signed token is sent verbatim; the caller is responsible for its grants.
+        if self._token:
+            return {AUTHORIZATION: "Bearer {}".format(self._token)}
+
         tok = AccessToken(self.api_key, self.api_secret)
         if grants:
             tok.with_grants(grants)
         if sip is not None:
             tok.with_sip_grants(sip)
 
         token = tok.to_jwt()
-
-        headers = {}
-        headers[AUTHORIZATION] = "Bearer {}".format(token)
-        return headers
+        return {AUTHORIZATION: "Bearer {}".format(token)}

livekit-api/livekit/api/connector_service.py

@@ -119,18 +119,17 @@ async def accept_whatsapp_call(
         Args:
             request: AcceptWhatsAppCallRequest containing call parameters and SDP
             timeout: Optional request timeout in seconds. When the request waits
-                for an answer (wait_until_answered), it defaults to the standard
-                ring window; set it above the ringing_timeout passed to
-                dial_whatsapp_call (the two calls are separate, so the SDK can't
-                derive it).
+                for the inbound party to join (wait_until_answered), it defaults
+                to the standard ring window.
 
         Returns:
             AcceptWhatsAppCallResponse with the room name
         """
         client_timeout: Optional[aiohttp.ClientTimeout] = None
         if request.wait_until_answered:
-            # Accept can block until the call is answered, so default to the
-            # standard ring window; the caller overrides via timeout.
+            # Waiting for the inbound party to join can block, so default the
+            # request timeout to the standard ring window; the caller overrides
+            # via timeout.
             client_timeout = aiohttp.ClientTimeout(
                 total=timeout if timeout else DEFAULT_RINGING_TIMEOUT
             )

livekit-api/livekit/api/livekit_api.py

@@ -29,28 +29,41 @@ def __init__(
         api_key: Optional[str] = None,
         api_secret: Optional[str] = None,
         *,
+        token: Optional[str] = None,
         timeout: Optional[aiohttp.ClientTimeout] = None,
         session: Optional[aiohttp.ClientSession] = None,
         failover: bool = True,
     ):
         """Create a new LiveKitAPI instance.
 
+        Authenticate with an API key and secret (recommended for backend use).
+        For token auth (client-side use, where the API secret must not be
+        exposed), prefer the :meth:`with_token` constructor.
+
         Args:
             url: LiveKit server URL (read from `LIVEKIT_URL` environment variable if not provided)
             api_key: API key (read from `LIVEKIT_API_KEY` environment variable if not provided)
             api_secret: API secret (read from `LIVEKIT_API_SECRET` environment variable if not provided)
+            token: Pre-signed access token (read from `LIVEKIT_TOKEN` environment variable if not provided)
             timeout: Request timeout (default: 10 seconds)
             session: aiohttp.ClientSession instance to use for requests, if not provided, a new one will be created
         """
         url = url or os.getenv("LIVEKIT_URL")
-        api_key = api_key or os.getenv("LIVEKIT_API_KEY")
-        api_secret = api_secret or os.getenv("LIVEKIT_API_SECRET")
+
+        # Only fall back to environment credentials when none were provided
+        # explicitly, so an ambient LIVEKIT_TOKEN can't silently override an
+        # explicit api_key/secret (or vice versa).
+        if not token and not api_key and not api_secret:
+            token = os.getenv("LIVEKIT_TOKEN")
+        if not token and not api_key and not api_secret:
+            api_key = os.getenv("LIVEKIT_API_KEY")
+            api_secret = os.getenv("LIVEKIT_API_SECRET")
 
         if not url:
             raise ValueError("url must be set")
 
-        if not api_key or not api_secret:
-            raise ValueError("api_key and api_secret must be set")
+        if not token and (not api_key or not api_secret):
+            raise ValueError("either token, or api_key and api_secret, must be set")
 
         self._custom_session = True
         self._session = session
@@ -60,14 +73,51 @@ def __init__(
                 timeout = aiohttp.ClientTimeout(total=10)
             self._session = aiohttp.ClientSession(timeout=timeout)
 
-        self._room = RoomService(self._session, url, api_key, api_secret, failover)
-        self._ingress = IngressService(self._session, url, api_key, api_secret, failover)
-        self._egress = EgressService(self._session, url, api_key, api_secret, failover)
-        self._sip = SipService(self._session, url, api_key, api_secret, failover)
-        self._agent_dispatch = AgentDispatchService(
-            self._session, url, api_key, api_secret, failover
-        )
-        self._connector = ConnectorService(self._session, url, api_key, api_secret, failover)
+        # In token mode there is no key/secret; pass empty strings and rely on
+        # the token, injected into each service below.
+        key = api_key or ""
+        secret = api_secret or ""
+        self._room = RoomService(self._session, url, key, secret, failover)
+        self._ingress = IngressService(self._session, url, key, secret, failover)
+        self._egress = EgressService(self._session, url, key, secret, failover)
+        self._sip = SipService(self._session, url, key, secret, failover)
+        self._agent_dispatch = AgentDispatchService(self._session, url, key, secret, failover)
+        self._connector = ConnectorService(self._session, url, key, secret, failover)
+
+        if token:
+            for svc in (
+                self._room,
+                self._ingress,
+                self._egress,
+                self._sip,
+                self._agent_dispatch,
+                self._connector,
+            ):
+                svc._token = token
+
+    @classmethod
+    def with_token(
+        cls,
+        token: str,
+        url: Optional[str] = None,
+        *,
+        timeout: Optional[aiohttp.ClientTimeout] = None,
+        session: Optional[aiohttp.ClientSession] = None,
+        failover: bool = True,
+    ) -> "LiveKitAPI":
+        """Create a LiveKitAPI authenticated with a pre-signed token.
+
+        The token is sent verbatim and must already carry the grants for the calls
+        it's used with. Since it needs no secret, this is suitable for client-side
+        use. `url` falls back to the `LIVEKIT_URL` environment variable.
+
+        Args:
+            token: Pre-signed access token
+            url: LiveKit server URL (read from `LIVEKIT_URL` if not provided)
+            timeout: Request timeout (default: 10 seconds)
+            session: aiohttp.ClientSession to use; a new one is created if omitted
+        """
+        return cls(url, token=token, timeout=timeout, session=session, failover=failover)
 
     @property
     def agent_dispatch(self) -> AgentDispatchService:

livekit-api/livekit/api/sip_service.py

@@ -36,6 +36,7 @@
     SIPMediaConfig,
 )
 from ._service import Service
+from .twirp_client import SipCallError, ServerError
 from ._dial_timeout import (
     dial_timeout as _dial_timeout,
     pin_ringing_timeout as _pin_ringing_timeout,
@@ -46,6 +47,14 @@
 """@private"""
 
 
+def _as_sip_error(err: ServerError) -> ServerError:
+    """Surface a SIP dialing failure as a SipCallError so callers can branch on
+    the SIP status; other failures (auth, validation) are returned unchanged."""
+    if "sip_status_code" in err.metadata:
+        return SipCallError.from_server_error(err)
+    return err
+
+
 class SipService(Service):
     """Client for LiveKit SIP Service API
 
@@ -806,14 +815,17 @@ async def create_sip_participant(
         if outbound_trunk_config:
             create.trunk = outbound_trunk_config
 
-        return await self._client.request(
-            SVC,
-            "CreateSIPParticipant",
-            create,
-            self._auth_header(VideoGrants(), sip=SIPGrants(call=True)),
-            SIPParticipantInfo,
-            timeout=client_timeout,
-        )
+        try:
+            return await self._client.request(
+                SVC,
+                "CreateSIPParticipant",
+                create,
+                self._auth_header(VideoGrants(), sip=SIPGrants(call=True)),
+                SIPParticipantInfo,
+                timeout=client_timeout,
+            )
+        except ServerError as e:
+            raise _as_sip_error(e) from None
 
     async def transfer_sip_participant(
         self,
@@ -837,20 +849,23 @@ async def transfer_sip_participant(
         # timeout doesn't depend on the server's default.
         _pin_ringing_timeout(transfer)
         client_timeout = aiohttp.ClientTimeout(total=_dial_timeout(timeout, transfer))
-        return await self._client.request(
-            SVC,
-            "TransferSIPParticipant",
-            transfer,
-            self._auth_header(
-                VideoGrants(
-                    room_admin=True,
-                    room=transfer.room_name,
+        try:
+            return await self._client.request(
+                SVC,
+                "TransferSIPParticipant",
+                transfer,
+                self._auth_header(
+                    VideoGrants(
+                        room_admin=True,
+                        room=transfer.room_name,
+                    ),
+                    sip=SIPGrants(call=True),
                 ),
-                sip=SIPGrants(call=True),
-            ),
-            SIPParticipantInfo,
-            timeout=client_timeout,
-        )
+                SIPParticipantInfo,
+                timeout=client_timeout,
+            )
+        except ServerError as e:
+            raise _as_sip_error(e) from None
 
     def _admin_headers(self) -> dict[str, str]:
         return self._auth_header(VideoGrants(), sip=SIPGrants(admin=True))