Module livekit.api

LiveKit Server APIs for Python

pip install livekit-api

Manage rooms, participants, egress, ingress, SIP, and Agent dispatch.

Primary entry point is LiveKitAPI.

See https://docs.livekit.io/reference/server/server-apis for more information.

Sub-modules

livekit.api.access_token

livekit.api.agent_dispatch_service

livekit.api.connector_service

livekit.api.egress_service

livekit.api.ingress_service

livekit.api.livekit_api

livekit.api.room_service

livekit.api.sip_service

livekit.api.twirp_client

livekit.api.version

livekit.api.webhook

Classes

class AccessToken (api_key: str | None = None, api_secret: str | None = None)

Expand source code

class AccessToken:
    ParticipantKind = Literal["standard", "egress", "ingress", "sip", "agent"]

    def __init__(
        self,
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
    ) -> None:
        api_key = api_key or os.getenv("LIVEKIT_API_KEY")
        api_secret = api_secret or os.getenv("LIVEKIT_API_SECRET")

        if not api_key or not api_secret:
            raise ValueError("api_key and api_secret must be set")

        self.api_key = api_key  # iss
        self.api_secret = api_secret
        self.claims = Claims()

        # default jwt claims
        self.identity = ""  # sub
        self.ttl = DEFAULT_TTL  # exp

    def with_ttl(self, ttl: datetime.timedelta) -> "AccessToken":
        self.ttl = ttl
        return self

    def with_grants(self, grants: VideoGrants) -> "AccessToken":
        self.claims.video = grants
        return self

    def with_sip_grants(self, grants: SIPGrants) -> "AccessToken":
        self.claims.sip = grants
        return self

    def with_observability_grants(self, grants: ObservabilityGrants) -> "AccessToken":
        self.claims.observability = grants
        return self

    def with_inference_grants(self, grants: InferenceGrants) -> "AccessToken":
        self.claims.inference = grants
        return self

    def with_identity(self, identity: str) -> "AccessToken":
        self.identity = identity
        return self

    def with_kind(self, kind: ParticipantKind) -> "AccessToken":
        self.claims.kind = kind
        return self

    def with_name(self, name: str) -> "AccessToken":
        self.claims.name = name
        return self

    def with_metadata(self, metadata: str) -> "AccessToken":
        self.claims.metadata = metadata
        return self

    def with_attributes(self, attributes: dict[str, str]) -> "AccessToken":
        self.claims.attributes = attributes
        return self

    def with_sha256(self, sha256: str) -> "AccessToken":
        self.claims.sha256 = sha256
        return self

    def with_room_preset(self, preset: str) -> "AccessToken":
        self.claims.room_preset = preset
        return self

    def with_room_config(self, config: RoomConfiguration) -> "AccessToken":
        self.claims.room_config = config
        return self

    def to_jwt(self) -> str:
        video = self.claims.video
        if video and video.room_join and (not self.identity or not video.room):
            raise ValueError("identity and room must be set when joining a room")

        # we want to exclude None values from the token
        jwt_claims = self.claims.asdict()
        jwt_claims.update(
            {
                "sub": self.identity,
                "iss": self.api_key,
                "nbf": calendar.timegm(datetime.datetime.now(datetime.timezone.utc).utctimetuple()),
                "exp": calendar.timegm(
                    (datetime.datetime.now(datetime.timezone.utc) + self.ttl).utctimetuple()
                ),
            }
        )
        return jwt.encode(jwt_claims, self.api_secret, algorithm="HS256")

Class variables

var ParticipantKind

Methods

def to_jwt(self) ‑> str

Expand source code

def to_jwt(self) -> str:
    video = self.claims.video
    if video and video.room_join and (not self.identity or not video.room):
        raise ValueError("identity and room must be set when joining a room")

    # we want to exclude None values from the token
    jwt_claims = self.claims.asdict()
    jwt_claims.update(
        {
            "sub": self.identity,
            "iss": self.api_key,
            "nbf": calendar.timegm(datetime.datetime.now(datetime.timezone.utc).utctimetuple()),
            "exp": calendar.timegm(
                (datetime.datetime.now(datetime.timezone.utc) + self.ttl).utctimetuple()
            ),
        }
    )
    return jwt.encode(jwt_claims, self.api_secret, algorithm="HS256")

def with_attributes(self, attributes: dict[str, str]) ‑> AccessToken

Expand source code

def with_attributes(self, attributes: dict[str, str]) -> "AccessToken":
    self.claims.attributes = attributes
    return self

def with_grants(self,
grants: VideoGrants) ‑> AccessToken

Expand source code

def with_grants(self, grants: VideoGrants) -> "AccessToken":
    self.claims.video = grants
    return self

def with_identity(self, identity: str) ‑> AccessToken

Expand source code

def with_identity(self, identity: str) -> "AccessToken":
    self.identity = identity
    return self

def with_inference_grants(self,
grants: InferenceGrants) ‑> AccessToken

Expand source code

def with_inference_grants(self, grants: InferenceGrants) -> "AccessToken":
    self.claims.inference = grants
    return self

def with_kind(self, kind: Literal['standard', 'egress', 'ingress', 'sip', 'agent']) ‑> AccessToken

Expand source code

def with_kind(self, kind: ParticipantKind) -> "AccessToken":
    self.claims.kind = kind
    return self

def with_metadata(self, metadata: str) ‑> AccessToken

Expand source code

def with_metadata(self, metadata: str) -> "AccessToken":
    self.claims.metadata = metadata
    return self

def with_name(self, name: str) ‑> AccessToken

Expand source code

def with_name(self, name: str) -> "AccessToken":
    self.claims.name = name
    return self

def with_observability_grants(self,
grants: ObservabilityGrants) ‑> AccessToken

Expand source code

def with_observability_grants(self, grants: ObservabilityGrants) -> "AccessToken":
    self.claims.observability = grants
    return self

def with_room_config(self, config: room.RoomConfiguration) ‑> AccessToken

Expand source code

def with_room_config(self, config: RoomConfiguration) -> "AccessToken":
    self.claims.room_config = config
    return self

def with_room_preset(self, preset: str) ‑> AccessToken

Expand source code

def with_room_preset(self, preset: str) -> "AccessToken":
    self.claims.room_preset = preset
    return self

def with_sha256(self, sha256: str) ‑> AccessToken

Expand source code

def with_sha256(self, sha256: str) -> "AccessToken":
    self.claims.sha256 = sha256
    return self

def with_sip_grants(self,
grants: SIPGrants) ‑> AccessToken

Expand source code

def with_sip_grants(self, grants: SIPGrants) -> "AccessToken":
    self.claims.sip = grants
    return self

def with_ttl(self, ttl: datetime.timedelta) ‑> AccessToken

Expand source code

def with_ttl(self, ttl: datetime.timedelta) -> "AccessToken":
    self.ttl = ttl
    return self

class InferenceGrants (perform: bool = False)

Expand source code

@dataclass
class InferenceGrants:
    # perform inference
    perform: bool = False

InferenceGrants(perform: bool = False)

Instance variables

var perform : bool

class LiveKitAPI (url: str | None = None,
api_key: str | None = None,
api_secret: str | None = None,
*,
token: str | None = None,
timeout: aiohttp.client.ClientTimeout | None = None,
session: aiohttp.client.ClientSession | None = None,
failover: bool = True)

Expand source code

class LiveKitAPI:
    """LiveKit Server API Client

    This class is the main entrypoint, which exposes all services.

    Usage:

    ```python
    from livekit import api
    lkapi = api.LiveKitAPI()
    rooms = await lkapi.room.list_rooms(api.proto_room.ListRoomsRequest(names=['test-room']))
    ```
    """

    def __init__(
        self,
        url: Optional[str] = None,
        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")

        # 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 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
        if not self._session:
            self._custom_session = False
            if not timeout:
                timeout = aiohttp.ClientTimeout(total=10)
            self._session = aiohttp.ClientSession(timeout=timeout)

        # 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:
        """Instance of the AgentDispatchService"""
        return self._agent_dispatch

    @property
    def room(self) -> RoomService:
        """Instance of the RoomService"""
        return self._room

    @property
    def ingress(self) -> IngressService:
        """Instance of the IngressService"""
        return self._ingress

    @property
    def egress(self) -> EgressService:
        """Instance of the EgressService"""
        return self._egress

    @property
    def sip(self) -> SipService:
        """Instance of the SipService"""
        return self._sip

    @property
    def connector(self) -> ConnectorService:
        """Instance of the ConnectorService"""
        return self._connector

    async def aclose(self) -> None:
        """Close the API client

        Call this before your application exits or when the API client is no longer needed."""
        # we do not close custom sessions, that's up to the caller
        if not self._custom_session and self._session is not None:
            await self._session.close()

    async def __aenter__(self) -> "LiveKitAPI":
        """@private

        Support for `async with`"""
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """@private

        Support for `async with`"""
        await self.aclose()

LiveKit Server API Client

This class is the main entrypoint, which exposes all services.

Usage:

from livekit import api
lkapi = api.LiveKitAPI()
rooms = await lkapi.room.list_rooms(api.proto_room.ListRoomsRequest(names=['test-room']))

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

Static methods

def with_token(token: str,
url: str | None = None,
*,
timeout: aiohttp.client.ClientTimeout | None = None,
session: aiohttp.client.ClientSession | None = 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

Instance variables

prop agent_dispatch : AgentDispatchService

Expand source code

@property
def agent_dispatch(self) -> AgentDispatchService:
    """Instance of the AgentDispatchService"""
    return self._agent_dispatch

Instance of the AgentDispatchService

prop connector : ConnectorService

Expand source code

@property
def connector(self) -> ConnectorService:
    """Instance of the ConnectorService"""
    return self._connector

Instance of the ConnectorService

prop egress : EgressService

Expand source code

@property
def egress(self) -> EgressService:
    """Instance of the EgressService"""
    return self._egress

Instance of the EgressService

prop ingress : IngressService

Expand source code

@property
def ingress(self) -> IngressService:
    """Instance of the IngressService"""
    return self._ingress

Instance of the IngressService

prop room : RoomService

Expand source code

@property
def room(self) -> RoomService:
    """Instance of the RoomService"""
    return self._room

Instance of the RoomService

prop sip : SipService

Expand source code

@property
def sip(self) -> SipService:
    """Instance of the SipService"""
    return self._sip

Instance of the SipService

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """Close the API client

    Call this before your application exits or when the API client is no longer needed."""
    # we do not close custom sessions, that's up to the caller
    if not self._custom_session and self._session is not None:
        await self._session.close()

Close the API client

Call this before your application exits or when the API client is no longer needed.

class ObservabilityGrants (write: bool = False)

Expand source code

@dataclass
class ObservabilityGrants:
    # write grants to publish observability data
    write: bool = False

ObservabilityGrants(write: bool = False)

Instance variables

var write : bool

class SIPGrants (admin: bool = False, call: bool = False)

Expand source code

@dataclass
class SIPGrants:
    # manage sip resources
    admin: bool = False
    # make outbound calls
    call: bool = False

SIPGrants(admin: bool = False, call: bool = False)

Instance variables

var admin : bool

var call : bool

class ServerError (code: str, msg: str, *, status: int, metadata: Dict[str, str] | None = None)

Expand source code

class ServerError(Exception):
    def __init__(
        self,
        code: str,
        msg: str,
        *,
        status: int,
        metadata: Optional[Dict[str, str]] = None,
    ) -> None:
        self._code = code
        self._msg = msg
        self._status = status
        self._metadata = metadata or {}

    @property
    def code(self) -> str:
        return self._code

    @property
    def message(self) -> str:
        return self._msg

    @property
    def status(self) -> int:
        """HTTP status code"""
        return self._status

    @property
    def metadata(self) -> Dict[str, str]:
        """Server-provided error metadata"""
        return self._metadata

    def __str__(self) -> str:
        result = f"ServerError(code={self.code}, message={self.message}, status={self.status}"
        if self.metadata:
            result += f", metadata={self.metadata}"
        result += ")"
        return result

Common base class for all non-exit exceptions.

Ancestors

• builtins.Exception
• builtins.BaseException

Subclasses

• SipCallError

Instance variables

prop code : str

Expand source code

@property
def code(self) -> str:
    return self._code

prop message : str

Expand source code

@property
def message(self) -> str:
    return self._msg

prop metadata : Dict[str, str]

Expand source code

@property
def metadata(self) -> Dict[str, str]:
    """Server-provided error metadata"""
    return self._metadata

Server-provided error metadata

prop status : int

Expand source code

@property
def status(self) -> int:
    """HTTP status code"""
    return self._status

HTTP status code

class TwirpError (code: str, msg: str, *, status: int, metadata: Dict[str, str] | None = None)

Expand source code

class ServerError(Exception):
    def __init__(
        self,
        code: str,
        msg: str,
        *,
        status: int,
        metadata: Optional[Dict[str, str]] = None,
    ) -> None:
        self._code = code
        self._msg = msg
        self._status = status
        self._metadata = metadata or {}

    @property
    def code(self) -> str:
        return self._code

    @property
    def message(self) -> str:
        return self._msg

    @property
    def status(self) -> int:
        """HTTP status code"""
        return self._status

    @property
    def metadata(self) -> Dict[str, str]:
        """Server-provided error metadata"""
        return self._metadata

    def __str__(self) -> str:
        result = f"ServerError(code={self.code}, message={self.message}, status={self.status}"
        if self.metadata:
            result += f", metadata={self.metadata}"
        result += ")"
        return result

Common base class for all non-exit exceptions.

Ancestors

• builtins.Exception
• builtins.BaseException

Subclasses

• SipCallError

Instance variables

prop code : str

Expand source code

@property
def code(self) -> str:
    return self._code

prop message : str

Expand source code

@property
def message(self) -> str:
    return self._msg

prop metadata : Dict[str, str]

Expand source code

@property
def metadata(self) -> Dict[str, str]:
    """Server-provided error metadata"""
    return self._metadata

Server-provided error metadata

prop status : int

Expand source code

@property
def status(self) -> int:
    """HTTP status code"""
    return self._status

HTTP status code

class ServerErrorCode

Expand source code

class ServerErrorCode:
    CANCELED = "canceled"
    UNKNOWN = "unknown"
    INVALID_ARGUMENT = "invalid_argument"
    MALFORMED = "malformed"
    DEADLINE_EXCEEDED = "deadline_exceeded"
    NOT_FOUND = "not_found"
    BAD_ROUTE = "bad_route"
    ALREADY_EXISTS = "already_exists"
    PERMISSION_DENIED = "permission_denied"
    UNAUTHENTICATED = "unauthenticated"
    RESOURCE_EXHAUSTED = "resource_exhausted"
    FAILED_PRECONDITION = "failed_precondition"
    ABORTED = "aborted"
    OUT_OF_RANGE = "out_of_range"
    UNIMPLEMENTED = "unimplemented"
    INTERNAL = "internal"
    UNAVAILABLE = "unavailable"
    DATA_LOSS = "dataloss"

Class variables

var ABORTED

var ALREADY_EXISTS

var BAD_ROUTE

var CANCELED

var DATA_LOSS

var DEADLINE_EXCEEDED

var FAILED_PRECONDITION

var INTERNAL

var INVALID_ARGUMENT

var MALFORMED

var NOT_FOUND

var OUT_OF_RANGE

var PERMISSION_DENIED

var RESOURCE_EXHAUSTED

var UNAUTHENTICATED

var UNAVAILABLE

var UNIMPLEMENTED

var UNKNOWN

class TwirpErrorCode

Expand source code

class ServerErrorCode:
    CANCELED = "canceled"
    UNKNOWN = "unknown"
    INVALID_ARGUMENT = "invalid_argument"
    MALFORMED = "malformed"
    DEADLINE_EXCEEDED = "deadline_exceeded"
    NOT_FOUND = "not_found"
    BAD_ROUTE = "bad_route"
    ALREADY_EXISTS = "already_exists"
    PERMISSION_DENIED = "permission_denied"
    UNAUTHENTICATED = "unauthenticated"
    RESOURCE_EXHAUSTED = "resource_exhausted"
    FAILED_PRECONDITION = "failed_precondition"
    ABORTED = "aborted"
    OUT_OF_RANGE = "out_of_range"
    UNIMPLEMENTED = "unimplemented"
    INTERNAL = "internal"
    UNAVAILABLE = "unavailable"
    DATA_LOSS = "dataloss"

Class variables

var ABORTED

var ALREADY_EXISTS

var BAD_ROUTE

var CANCELED

var DATA_LOSS

var DEADLINE_EXCEEDED

var FAILED_PRECONDITION

var INTERNAL

var INVALID_ARGUMENT

var MALFORMED

var NOT_FOUND

var OUT_OF_RANGE

var PERMISSION_DENIED

var RESOURCE_EXHAUSTED

var UNAUTHENTICATED

var UNAVAILABLE

var UNIMPLEMENTED

var UNKNOWN

class SipCallError (code: str, msg: str, *, status: int, metadata: Dict[str, str] | None = None)

Expand source code

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")
        if raw is None:
            return None
        try:
            return int(raw)
        except ValueError:
            return 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

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.

Ancestors

• ServerError
• builtins.Exception
• builtins.BaseException

Static methods

def from_server_error(err: ServerError) ‑> SipCallError

Instance variables

prop sip_status : str | None

Expand source code

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

The SIP reason phrase of the failed call, e.g. "Busy Here".

prop sip_status_code : int | None

Expand source code

@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")
    if raw is None:
        return None
    try:
        return int(raw)
    except ValueError:
        return None

The SIP response code of the failed call, e.g. 486 (Busy Here).

Inherited members

• ServerError:
  • metadata
  • status

class TokenVerifier (api_key: str | None = None,
api_secret: str | None = None,
*,
leeway: datetime.timedelta = datetime.timedelta(seconds=60))

Expand source code

class TokenVerifier:
    def __init__(
        self,
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
        *,
        leeway: datetime.timedelta = DEFAULT_LEEWAY,
    ) -> None:
        api_key = api_key or os.getenv("LIVEKIT_API_KEY")
        api_secret = api_secret or os.getenv("LIVEKIT_API_SECRET")

        self.api_key = api_key
        self.api_secret = api_secret
        self._leeway = leeway

    def verify(self, token: str, *, verify_signature: bool = True) -> Claims:
        if verify_signature and (not self.api_key or not self.api_secret):
            raise ValueError("api_key and api_secret must be set")

        claims = jwt.decode(
            token,
            key=self.api_secret or "",
            issuer=self.api_key or "",
            algorithms=["HS256"],
            leeway=self._leeway.total_seconds(),
            options={"verify_signature": verify_signature},
        )
        video_dict = claims.get("video", dict())
        video_dict = {camel_to_snake(k): v for k, v in video_dict.items()}
        video_dict = {k: v for k, v in video_dict.items() if k in VideoGrants.__dataclass_fields__}
        video = VideoGrants(**video_dict)

        sip_dict = claims.get("sip", dict())
        sip_dict = {camel_to_snake(k): v for k, v in sip_dict.items()}
        sip_dict = {k: v for k, v in sip_dict.items() if k in SIPGrants.__dataclass_fields__}
        sip = SIPGrants(**sip_dict)

        inference_dict = claims.get("inference", dict())
        inference_dict = {camel_to_snake(k): v for k, v in inference_dict.items()}
        inference_dict = {
            k: v for k, v in inference_dict.items() if k in InferenceGrants.__dataclass_fields__
        }
        inference = InferenceGrants(**inference_dict)

        grant_claims = Claims(
            identity=claims.get("sub", ""),
            name=claims.get("name", ""),
            video=video,
            sip=sip,
            inference=inference,
            attributes=claims.get("attributes", {}),
            metadata=claims.get("metadata", ""),
            sha256=claims.get("sha256", ""),
        )

        if claims.get("roomPreset"):
            grant_claims.room_preset = claims.get("roomPreset")
        room_config = claims.get("roomConfig")
        if room_config:
            grant_claims.room_config = ParseDict(
                room_config,
                RoomConfiguration(),
                ignore_unknown_fields=True,
            )

        return grant_claims

Methods

def verify(self, token: str, *, verify_signature: bool = True) ‑> Claims

Expand source code

def verify(self, token: str, *, verify_signature: bool = True) -> Claims:
    if verify_signature and (not self.api_key or not self.api_secret):
        raise ValueError("api_key and api_secret must be set")

    claims = jwt.decode(
        token,
        key=self.api_secret or "",
        issuer=self.api_key or "",
        algorithms=["HS256"],
        leeway=self._leeway.total_seconds(),
        options={"verify_signature": verify_signature},
    )
    video_dict = claims.get("video", dict())
    video_dict = {camel_to_snake(k): v for k, v in video_dict.items()}
    video_dict = {k: v for k, v in video_dict.items() if k in VideoGrants.__dataclass_fields__}
    video = VideoGrants(**video_dict)

    sip_dict = claims.get("sip", dict())
    sip_dict = {camel_to_snake(k): v for k, v in sip_dict.items()}
    sip_dict = {k: v for k, v in sip_dict.items() if k in SIPGrants.__dataclass_fields__}
    sip = SIPGrants(**sip_dict)

    inference_dict = claims.get("inference", dict())
    inference_dict = {camel_to_snake(k): v for k, v in inference_dict.items()}
    inference_dict = {
        k: v for k, v in inference_dict.items() if k in InferenceGrants.__dataclass_fields__
    }
    inference = InferenceGrants(**inference_dict)

    grant_claims = Claims(
        identity=claims.get("sub", ""),
        name=claims.get("name", ""),
        video=video,
        sip=sip,
        inference=inference,
        attributes=claims.get("attributes", {}),
        metadata=claims.get("metadata", ""),
        sha256=claims.get("sha256", ""),
    )

    if claims.get("roomPreset"):
        grant_claims.room_preset = claims.get("roomPreset")
    room_config = claims.get("roomConfig")
    if room_config:
        grant_claims.room_config = ParseDict(
            room_config,
            RoomConfiguration(),
            ignore_unknown_fields=True,
        )

    return grant_claims

class VideoGrants (room_create: bool | None = None,
room_list: bool | None = None,
room_record: bool | None = None,
room_admin: bool | None = None,
room_join: bool | None = None,
room: str = '',
destination_room: str | None = None,
can_publish: bool = True,
can_subscribe: bool = True,
can_publish_data: bool = True,
can_publish_sources: List[str] | None = None,
can_update_own_metadata: bool | None = None,
ingress_admin: bool | None = None,
hidden: bool | None = None,
recorder: bool | None = None,
agent: bool | None = None,
can_manage_agent_session: bool | None = None)

Expand source code

@dataclass
class VideoGrants:
    # actions on rooms
    room_create: Optional[bool] = None
    room_list: Optional[bool] = None
    room_record: Optional[bool] = None

    # actions on a particular room
    room_admin: Optional[bool] = None
    room_join: Optional[bool] = None
    room: str = ""

    # allows forwarding participant to room
    destination_room: Optional[str] = None

    # permissions within a room
    can_publish: bool = True
    can_subscribe: bool = True
    can_publish_data: bool = True

    # TrackSource types that a participant may publish.
    # When set, it supersedes CanPublish. Only sources explicitly set here can be
    # published
    can_publish_sources: Optional[List[str]] = None

    # by default, a participant is not allowed to update its own metadata
    can_update_own_metadata: Optional[bool] = None

    # actions on ingresses
    ingress_admin: Optional[bool] = None  # applies to all ingress

    # participant is not visible to other participants (useful when making bots)
    hidden: Optional[bool] = None

    # [deprecated] indicates to the room that current participant is a recorder
    recorder: Optional[bool] = None

    # indicates that the holder can register as an Agent framework worker
    agent: Optional[bool] = None

    # allow participant to manage an agent session via RemoteSession
    can_manage_agent_session: Optional[bool] = None

VideoGrants(room_create: Optional[bool] = None, room_list: Optional[bool] = None, room_record: Optional[bool] = None, room_admin: Optional[bool] = None, room_join: Optional[bool] = None, room: str = '', destination_room: Optional[str] = None, can_publish: bool = True, can_subscribe: bool = True, can_publish_data: bool = True, can_publish_sources: Optional[List[str]] = None, can_update_own_metadata: Optional[bool] = None, ingress_admin: Optional[bool] = None, hidden: Optional[bool] = None, recorder: Optional[bool] = None, agent: Optional[bool] = None, can_manage_agent_session: Optional[bool] = None)

Instance variables

var agent : bool | None

var can_manage_agent_session : bool | None

var can_publish : bool

var can_publish_data : bool

var can_publish_sources : List[str] | None

var can_subscribe : bool

var can_update_own_metadata : bool | None

var destination_room : str | None

var hidden : bool | None

var ingress_admin : bool | None

var recorder : bool | None

var room : str

var room_admin : bool | None

var room_create : bool | None

var room_join : bool | None

var room_list : bool | None

var room_record : bool | None

class WebhookReceiver (token_verifier: TokenVerifier)

Expand source code

class WebhookReceiver:
    def __init__(self, token_verifier: TokenVerifier):
        self._verifier = token_verifier

    def receive(self, body: str, auth_token: str) -> WebhookEvent:
        claims = self._verifier.verify(auth_token)
        if claims.sha256 is None:
            raise Exception("sha256 was not found in the token")

        body_hash = hashlib.sha256(body.encode()).digest()
        claims_hash = base64.b64decode(claims.sha256)

        if body_hash != claims_hash:
            raise Exception("hash mismatch")

        return Parse(body, WebhookEvent(), ignore_unknown_fields=True)

Methods

def receive(self, body: str, auth_token: str) ‑> WebhookEvent

Expand source code

def receive(self, body: str, auth_token: str) -> WebhookEvent:
    claims = self._verifier.verify(auth_token)
    if claims.sha256 is None:
        raise Exception("sha256 was not found in the token")

    body_hash = hashlib.sha256(body.encode()).digest()
    claims_hash = base64.b64decode(claims.sha256)

    if body_hash != claims_hash:
        raise Exception("hash mismatch")

    return Parse(body, WebhookEvent(), ignore_unknown_fields=True)