consistent error class ServerError, test suite

livekit-api/livekit/api/__init__.py

@@ -36,7 +36,7 @@
 from livekit.protocol.connector_whatsapp import *
 from livekit.protocol.connector_twilio import *
 
-from .twirp_client import TwirpError, TwirpErrorCode
+from .twirp_client import ServerError, SipCallError, TwirpError, TwirpErrorCode
 from .livekit_api import LiveKitAPI
 from .access_token import (
     InferenceGrants,
@@ -64,6 +64,8 @@
     "AccessToken",
     "TokenVerifier",
     "WebhookReceiver",
+    "ServerError",
     "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,35 @@ 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")
+        token = token or os.getenv("LIVEKIT_TOKEN")
         api_key = api_key or os.getenv("LIVEKIT_API_KEY")
         api_secret = api_secret or 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 +67,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))

livekit-api/livekit/api/twirp_client.py

@@ -28,16 +28,20 @@
     origin_of,
     pick_next,
 )
+from .version import __version__
 
 DEFAULT_PREFIX = "twirp"
 
 logger = logging.getLogger("livekit")
 
+# Identifies the SDK and version to the server on every request.
+_USER_AGENT = f"livekit-server-sdk-python/{__version__}"
+
 # Shared across all clients in the process so the region list is fetched once.
 _REGION_CACHE = RegionCache()
 
 
-class TwirpError(Exception):
+class ServerError(Exception):
     def __init__(
         self,
         code: str,
@@ -66,17 +70,62 @@ def status(self) -> int:
 
     @property
     def metadata(self) -> Dict[str, str]:
-        """Twirp metadata"""
+        """Server-provided error metadata"""
         return self._metadata
 
     def __str__(self) -> str:
-        result = f"TwirpError(code={self.code}, message={self.message}, status={self.status}"
+        result = f"ServerError(code={self.code}, message={self.message}, status={self.status}"
         if self.metadata:
             result += f", metadata={self.metadata}"
         result += ")"
         return result
 
 
+class SipCallError(ServerError):
+    """A :class:`ServerError` from a SIP dialing call (``create_sip_participant`` /
+    ``transfer_sip_participant``) that failed with a SIP response status. The SIP
+    code and reason are exposed as properties; any other error metadata remains
+    available via :attr:`metadata`."""
+
+    @property
+    def sip_status_code(self) -> Optional[int]:
+        """The SIP response code of the failed call, e.g. 486 (Busy Here)."""
+        raw = self.metadata.get("sip_status_code")
+        return int(raw) if raw is not None else None
+
+    @property
+    def sip_status(self) -> Optional[str]:
+        """The SIP reason phrase of the failed call, e.g. "Busy Here"."""
+        return self.metadata.get("sip_status")
+
+    @classmethod
+    def from_server_error(cls, err: ServerError) -> "SipCallError":
+        return cls(err.code, err.message, status=err.status, metadata=err.metadata)
+
+    def __str__(self) -> str:
+        code = self.metadata.get("sip_status_code")
+        if code is None:
+            return super().__str__()
+        # A clear, SIP-specific representation, including any extra metadata.
+        reason = self.metadata.get("sip_status")
+        result = f"SIP call failed: {code}"
+        if reason:
+            result += f" {reason}"
+        result += f" ({self.code})"
+        extra = {
+            k: v
+            for k, v in self.metadata.items()
+            if k not in ("sip_status_code", "sip_status", "error_details")
+        }
+        if extra:
+            result += " [" + ", ".join(f"{k}={v}" for k, v in extra.items()) + "]"
+        return result
+
+
+# Deprecated alias for :class:`ServerError`, kept for backwards compatibility.
+TwirpError = ServerError
+
+
 class TwirpErrorCode:
     CANCELED = "canceled"
     UNKNOWN = "unknown"
@@ -145,8 +194,9 @@ async def request(
         headers intact — against the next untried region, with exponential
         backoff. A 4xx is returned immediately."""
         path = f"{self.prefix}/{self.pkg}.{service}/{method}"
-        forward_headers = dict(headers)  # for the discovery fetch (no content-type yet)
         headers = dict(headers)
+        headers["User-Agent"] = _USER_AGENT
+        forward_headers = dict(headers)  # for the discovery fetch (no content-type yet)
         headers["Content-Type"] = "application/protobuf"
         serialized_data = data.SerializeToString()
 
@@ -183,7 +233,7 @@ async def request(
                         error_data = {}
                     if resp.status < 500:
                         # 4xx is terminal.
-                        raise self._twirp_error(error_data, resp.status)
+                        raise self._server_error(error_data, resp.status)
                     retryable_status = resp.status
             except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                 transport_exc = e
@@ -200,7 +250,7 @@ async def request(
             if next_origin is None:
                 if transport_exc is not None:
                     raise transport_exc
-                raise self._twirp_error(error_data, retryable_status or 500)
+                raise self._server_error(error_data, retryable_status or 500)
 
             reason = transport_exc if transport_exc is not None else f"status {retryable_status}"
             logger.warning(
@@ -216,8 +266,8 @@ async def request(
         raise RuntimeError("failover loop exited without returning")  # unreachable
 
     @staticmethod
-    def _twirp_error(error_data: Dict, status: int) -> "TwirpError":
-        return TwirpError(
+    def _server_error(error_data: Dict, status: int) -> "ServerError":
+        return ServerError(
             error_data.get("code", "unknown"),
             error_data.get("msg", ""),
             status=status,