-
Notifications
You must be signed in to change notification settings - Fork 122
consistent error class ServerError, test suite #744
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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") | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🔴 Environment token silently overrides explicitly-provided API key and secret credentials A token read from the Impact: Users with Mechanism: env-var token wins over explicit key/secret due to unconditional precedenceAt The fix should either: (a) only read Prompt for agentsWas this helpful? React with 👍 or 👎 to provide feedback.
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should we only fall back to |
||
| 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: | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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 | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: should we add a try/except for the |
||
|
|
||
| @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, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🚩 accept_whatsapp_call timeout lacks the RINGING_TIMEOUT_MARGIN used by SIP service
The
accept_whatsapp_callmethod usesDEFAULT_RINGING_TIMEOUT(30s) as the request timeout whenwait_until_answeredis set, without addingRINGING_TIMEOUT_MARGIN(2s). In contrast, the SIP service'screate_sip_participantusesring + RINGING_TIMEOUT_MARGINto ensure the HTTP request outlasts the ringing window. If the server takes exactly 30s to ring, the connector request could time out at the boundary. This is pre-existing behavior (unchanged by this PR) but the inconsistency is worth noting since the PR cleaned up theDialRequestunion that previously includedAcceptWhatsAppCallRequest.Was this helpful? React with 👍 or 👎 to provide feedback.