Module livekit.plugins.synthesia

Synthesia plugin for LiveKit Agents

See https://docs.livekit.io/agents/models/avatar/plugins/synthesia/ for more information.

Classes

class AvatarConfig (avatar_ids: Sequence[str])

Expand source code

@dataclass
class AvatarConfig:
    """The avatars to render in the room.

    ``avatar_ids`` holds one to five gallery ids of Synthesia avatars available
    to your workspace. The first id is the active avatar; the rest are
    available for swapping in during the session. Normalized to a tuple.
    """

    avatar_ids: Sequence[str]

    def __post_init__(self) -> None:
        if isinstance(self.avatar_ids, (str, bytes)):
            raise ValueError("avatar_ids must be a list of ids, not a single string")
        self.avatar_ids = tuple(self.avatar_ids)
        if not 1 <= len(self.avatar_ids) <= MAX_AVATAR_IDS:
            raise ValueError(
                f"avatar_ids must contain between 1 and {MAX_AVATAR_IDS} ids, "
                f"got {len(self.avatar_ids)}"
            )

The avatars to render in the room.

avatar_ids holds one to five gallery ids of Synthesia avatars available to your workspace. The first id is the active avatar; the rest are available for swapping in during the session. Normalized to a tuple.

Instance variables

var avatar_ids : Sequence[str]

class AvatarSession (avatar_config: AvatarConfig,
*,
api_key: str | None = None,
api_url: str | None = None,
join_timeout: float = 30.0,
avatar_participant_identity: str | None = None,
avatar_participant_name: str | None = None)

Expand source code

class AvatarSession(BaseAvatarSession):
    """A Synthesia interactive avatar for a LiveKit voice agent.

    Construct with up to five gallery ``avatar_ids`` and call ``start()`` before
    ``AgentSession.start()``. The first id is the active avatar and the rest are
    precomputed for a future mid-session swap. Credentials fall back to the
    ``SYNTHESIA_API_KEY`` and ``SYNTHESIA_API_URL`` environment variables.

    Logs when the room ends cleanly and when the avatar track drops
    unexpectedly mid-session.

    Pass ``avatar_participant_identity`` to override the LiveKit identity the
    avatar joins under. It must be unique per concurrent avatar in a room, since
    LiveKit evicts an existing participant when a second joins with the same
    identity. It defaults to ``synthesia-avatar-agent``.
    """

    def __init__(
        self,
        avatar_config: AvatarConfig,
        *,
        api_key: str | None = None,
        api_url: str | None = None,
        join_timeout: float = DEFAULT_JOIN_TIMEOUT,
        avatar_participant_identity: str | None = None,
        avatar_participant_name: str | None = None,
    ) -> None:
        super().__init__()

        if avatar_participant_identity is not None and not avatar_participant_identity.strip():
            raise SynthesiaError("avatar_participant_identity must be a non-empty string")
        if avatar_participant_name is not None and not avatar_participant_name.strip():
            raise SynthesiaError("avatar_participant_name must be a non-empty string")
        self._avatar_identity = avatar_participant_identity or AVATAR_IDENTITY
        self._avatar_name = avatar_participant_name or AVATAR_NAME

        key = api_key or os.environ.get("SYNTHESIA_API_KEY")
        if not key:
            raise SynthesiaError(
                "a Synthesia API key is required: pass api_key or set SYNTHESIA_API_KEY"
            )

        url = api_url or os.environ.get("SYNTHESIA_API_URL") or DEFAULT_API_URL

        self._config = SessionConfig(
            avatar_ids=tuple(avatar_config.avatar_ids),
            api_key=key,
            api_url=url,
            join_timeout=join_timeout,
        )
        self._state = _State.IDLE
        # Guards against re-entrant start() calls. Distinct from _state: while a
        # start is awaiting a prior teardown the lifecycle is still its old
        # state, so this cannot be folded into _state.
        self._starting = False
        self._close_done = asyncio.Event()
        self._session_id: str | None = None
        self._teardown_task: asyncio.Task[None] | None = None
        # The exact sink this session installed, tracked by identity rather than
        # type or chain position: AgentSession.start() can wrap it in
        # TranscriptSynchronizer/RecorderAudioOutput, and a caller may already
        # have their own DataStreamAudioOutput installed before start() runs.
        self._audio_output: DataStreamAudioOutput | None = None

    @property
    def avatar_identity(self) -> str:
        return self._avatar_identity

    @property
    def provider(self) -> str:
        return "synthesia"

    async def start(
        self,
        agent_session: AgentSession,
        room: rtc.Room,
        *,
        livekit_url: str | None = None,
        livekit_api_key: str | None = None,
        livekit_api_secret: str | None = None,
    ) -> None:
        if self._starting:
            raise SynthesiaError("start() is already in progress")
        self._starting = True
        try:
            if self._teardown_task is not None:
                teardown = self._teardown_task
                await asyncio.shield(asyncio.gather(teardown, return_exceptions=True))
                if not teardown.cancelled() and teardown.exception() is not None:
                    raise SynthesiaError(
                        "the previous avatar session did not tear down cleanly; not restarting"
                    ) from teardown.exception()
            if self._state is _State.STARTED:
                return

            url = livekit_url or os.environ.get("LIVEKIT_URL")
            lk_key = livekit_api_key or os.environ.get("LIVEKIT_API_KEY")
            lk_secret = livekit_api_secret or os.environ.get("LIVEKIT_API_SECRET")
            # A blank key or secret still mints a syntactically valid token that
            # Synthesia accepts, since it cannot verify a signature made with the
            # developer's own secret. LiveKit only rejects it once the avatar
            # tries to join, so catch it here instead.
            if not _present(url) or not _present(lk_key) or not _present(lk_secret):
                raise SynthesiaError(
                    "LiveKit url, API key, and API secret are required: pass them or "
                    "set LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET"
                )
            url = _to_ws_url(url)
            if not url.startswith(("ws://", "wss://")):
                raise SynthesiaError(f"livekit_url {url!r} is not a ws:// or wss:// URL")

            self._state = _State.IDLE
            self._close_done.clear()
            self._teardown_task = None

            try:
                await super().start(agent_session, room)
                token = self._mint_token(room=room, lk_key=lk_key, lk_secret=lk_secret)
                client = SynthesiaAPI(api_key=self._config.api_key, api_url=self._config.api_url)
                launch_conn = dataclasses.replace(
                    DEFAULT_API_CONNECT_OPTIONS,
                    timeout=self._config.join_timeout,
                    max_retry=0,
                )
                response = await client.start_session(
                    StartSessionRequest(
                        avatar_ids=list(self._config.avatar_ids),
                        livekit_url=url,
                        lk_token=token,
                    ),
                    conn_options=launch_conn,
                )
                self._session_id = response.session_id

                # TODO: confirm the avatar worker's expected audio sample rate and
                # pass sample_rate explicitly once it is verified against the real
                # worker.
                audio_output = DataStreamAudioOutput(
                    room,
                    destination_identity=self.avatar_identity,
                    wait_remote_track=rtc.TrackKind.KIND_VIDEO,
                )
                # replace_audio_tail keeps any wrapper AgentSession.start() adds
                # later (TranscriptSynchronizer, RecorderAudioOutput) attached.
                agent_session.output.replace_audio_tail(audio_output)
                self._audio_output = audio_output

                await self.wait_for_join(timeout=self._config.join_timeout)
            except asyncio.TimeoutError as e:
                await self.aclose()
                await self._discard_partial_start()
                raise SynthesiaError(
                    f"avatar did not join within {self._config.join_timeout}s",
                    type=ErrorType.TIMEOUT,
                ) from e
            except BaseException:
                await self.aclose()
                await self._discard_partial_start()
                raise

            if self._state is _State.CLOSED:
                await self._close_done.wait()
                await self._discard_partial_start()
                raise SynthesiaError("avatar session was closed while starting")

            room.on("disconnected", self._on_room_disconnected)
            room.on("track_unpublished", self._on_track_unpublished)
            room.on("participant_disconnected", self._on_participant_disconnected)
            self._state = _State.STARTED
        finally:
            self._starting = False

    async def _discard_partial_start(self) -> None:
        # An aclose() that ran concurrently with start() cannot see the audio
        # output and session id that start() set after it completed, so a failed
        # start cleans them up itself, closing the audio output's background
        # tasks rather than just dropping the reference. Only the sink this
        # session installed is touched: it may now be wrapped by
        # AgentSession.start() (TranscriptSynchronizer, RecorderAudioOutput), and
        # the caller may have had their own DataStreamAudioOutput before this
        # session ever ran, so neither the current chain head nor its type says
        # what to close.
        audio_output, self._audio_output = self._audio_output, None
        if audio_output is not None and hasattr(audio_output, "aclose"):
            await audio_output.aclose()
        self._session_id = None

    async def swap_avatar(self, avatar_id: str, *, timeout: float = DEFAULT_SWAP_TIMEOUT) -> str:
        """Switch the rendered avatar mid-session.

        ``avatar_id`` must be one of the ids passed in ``AvatarConfig``, or
        ``"default"`` for the first id in that list. Returns the now-active
        avatar id once the swap has taken effect. Raises a ``SynthesiaError``
        of type ``UNKNOWN_AVATAR`` for an id that was not precomputed,
        ``CONNECTION`` if the worker cannot be reached, and no ``type`` if the
        worker rejects the swap.
        """
        if self._state is not _State.STARTED or self._ending() or self._room is None:
            raise SynthesiaError("swap_avatar() requires a started avatar session")
        if avatar_id != "default" and avatar_id not in self._config.avatar_ids:
            raise SynthesiaError(
                f"avatar {avatar_id!r} was not in initial list of avatar_ids",
                type=ErrorType.UNKNOWN_AVATAR,
            )

        try:
            raw = await self._room.local_participant.perform_rpc(
                destination_identity=self.avatar_identity,
                method="swapAvatar",
                payload=json.dumps({"avatar_id": avatar_id}),
                response_timeout=timeout,
            )
        except Exception as e:
            raise SynthesiaError(f"avatar swap RPC failed: {e}", type=ErrorType.CONNECTION) from e

        try:
            response = json.loads(raw)
        except ValueError as e:
            raise SynthesiaError("avatar swap returned a malformed response") from e
        avatar_id_result = response.get("avatar_id") if isinstance(response, dict) else None
        if (
            not isinstance(response, dict)
            or response.get("error")
            or not isinstance(avatar_id_result, str)
        ):
            detail = response.get("error") if isinstance(response, dict) else None
            raise SynthesiaError(f"avatar swap failed: {detail or raw}")
        return avatar_id_result

    def _mint_token(self, *, room: rtc.Room, lk_key: str, lk_secret: str) -> str:
        # Synthesia rejects a token whose publish-on-behalf attribute is empty,
        # since the worker has no agent to publish the avatar's audio for. Prefer
        # the job context over room.local_participant: inside a job, the room may
        # not have finished connecting yet when start() runs. Standalone callers
        # (no job context) connect the room themselves before calling start(), so
        # room.local_participant is safe there.
        job_ctx = get_job_context(required=False)
        agent_identity = (
            job_ctx.local_participant_identity
            if job_ctx is not None
            else room.local_participant.identity
        )
        if not _present(agent_identity):
            raise SynthesiaError(
                "the room's local participant has no identity; connect the room "
                "before starting the avatar session"
            )

        token = (
            api.AccessToken(lk_key, lk_secret)
            .with_identity(self.avatar_identity)
            .with_name(self._avatar_name)
            .with_kind("agent")
            .with_grants(
                api.VideoGrants(
                    room_join=True,
                    room=room.name,
                    can_publish=True,
                    can_subscribe=True,
                    can_publish_data=True,
                )
            )
            .with_attributes({ATTRIBUTE_PUBLISH_ON_BEHALF: agent_identity})
            .with_ttl(TOKEN_TTL)
        )
        return token.to_jwt()

    def _on_room_disconnected(self, *args: object) -> None:
        if self._ending():
            return
        logger.info("avatar session ended")
        self._begin_teardown()

    def _on_track_unpublished(
        self,
        publication: rtc.RemoteTrackPublication,
        participant: rtc.RemoteParticipant,
    ) -> None:
        if (
            participant.identity == self.avatar_identity
            and publication.kind == rtc.TrackKind.KIND_VIDEO
        ):
            self._report_avatar_lost()

    def _on_participant_disconnected(self, participant: rtc.RemoteParticipant) -> None:
        if participant.identity == self.avatar_identity:
            self._report_avatar_lost()

    def _report_avatar_lost(self) -> None:
        if self._ending():
            return
        # The avatar leaving while the room is still up is a worker crash. If the
        # room itself is going down, this is a clean end and _on_room_disconnected
        # reports it instead.
        if self._room is not None and not self._room.isconnected():
            return
        logger.warning("avatar left the room unexpectedly")
        self._begin_teardown()

    def _ending(self) -> bool:
        return self._state is _State.CLOSED or self._teardown_task is not None

    def _begin_teardown(self) -> None:
        if self._teardown_task is None and self._state is not _State.CLOSED:
            self._teardown_task = asyncio.create_task(self.aclose())
            self._teardown_task.add_done_callback(self._on_teardown_done)

    def _on_teardown_done(self, task: asyncio.Task[None]) -> None:
        if task.cancelled():
            return
        exc = task.exception()
        if exc is not None:
            logger.error("avatar teardown failed", exc_info=exc)

    async def aclose(self) -> None:
        if self._state is _State.CLOSED:
            await self._close_done.wait()
            return
        self._state = _State.CLOSED
        self._session_id = None

        try:
            # Only the sink this session installed is closed, by identity: it may
            # now be wrapped by AgentSession.start() (TranscriptSynchronizer,
            # RecorderAudioOutput), and the caller may have had their own
            # DataStreamAudioOutput before this session ever ran, so neither the
            # current chain head nor its type says what to close. The wrapper
            # chain itself is left in place; there is no public API to remove a
            # tail sink from it without supplying a replacement.
            audio_output, self._audio_output = self._audio_output, None
            if audio_output is not None and hasattr(audio_output, "aclose"):
                await audio_output.aclose()

            if self._room is not None:
                self._room.off("disconnected", self._on_room_disconnected)
                self._room.off("track_unpublished", self._on_track_unpublished)
                self._room.off("participant_disconnected", self._on_participant_disconnected)

            await super().aclose()
        finally:
            self._close_done.set()

A Synthesia interactive avatar for a LiveKit voice agent.

Construct with up to five gallery avatar_ids and call start() before AgentSession.start(). The first id is the active avatar and the rest are precomputed for a future mid-session swap. Credentials fall back to the SYNTHESIA_API_KEY and SYNTHESIA_API_URL environment variables.

Logs when the room ends cleanly and when the avatar track drops unexpectedly mid-session.

Pass avatar_participant_identity to override the LiveKit identity the avatar joins under. It must be unique per concurrent avatar in a room, since LiveKit evicts an existing participant when a second joins with the same identity. It defaults to synthesia-avatar-agent.

Ancestors

• livekit.agents.voice.avatar._types.AvatarSession
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop avatar_identity : str

Expand source code

@property
def avatar_identity(self) -> str:
    return self._avatar_identity

The participant identifier of the avatar

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return "synthesia"

The provider of the avatar

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    if self._state is _State.CLOSED:
        await self._close_done.wait()
        return
    self._state = _State.CLOSED
    self._session_id = None

    try:
        # Only the sink this session installed is closed, by identity: it may
        # now be wrapped by AgentSession.start() (TranscriptSynchronizer,
        # RecorderAudioOutput), and the caller may have had their own
        # DataStreamAudioOutput before this session ever ran, so neither the
        # current chain head nor its type says what to close. The wrapper
        # chain itself is left in place; there is no public API to remove a
        # tail sink from it without supplying a replacement.
        audio_output, self._audio_output = self._audio_output, None
        if audio_output is not None and hasattr(audio_output, "aclose"):
            await audio_output.aclose()

        if self._room is not None:
            self._room.off("disconnected", self._on_room_disconnected)
            self._room.off("track_unpublished", self._on_track_unpublished)
            self._room.off("participant_disconnected", self._on_participant_disconnected)

        await super().aclose()
    finally:
        self._close_done.set()

async def start(self,
agent_session: AgentSession,
room: rtc.Room,
*,
livekit_url: str | None = None,
livekit_api_key: str | None = None,
livekit_api_secret: str | None = None) ‑> None

Expand source code

async def start(
    self,
    agent_session: AgentSession,
    room: rtc.Room,
    *,
    livekit_url: str | None = None,
    livekit_api_key: str | None = None,
    livekit_api_secret: str | None = None,
) -> None:
    if self._starting:
        raise SynthesiaError("start() is already in progress")
    self._starting = True
    try:
        if self._teardown_task is not None:
            teardown = self._teardown_task
            await asyncio.shield(asyncio.gather(teardown, return_exceptions=True))
            if not teardown.cancelled() and teardown.exception() is not None:
                raise SynthesiaError(
                    "the previous avatar session did not tear down cleanly; not restarting"
                ) from teardown.exception()
        if self._state is _State.STARTED:
            return

        url = livekit_url or os.environ.get("LIVEKIT_URL")
        lk_key = livekit_api_key or os.environ.get("LIVEKIT_API_KEY")
        lk_secret = livekit_api_secret or os.environ.get("LIVEKIT_API_SECRET")
        # A blank key or secret still mints a syntactically valid token that
        # Synthesia accepts, since it cannot verify a signature made with the
        # developer's own secret. LiveKit only rejects it once the avatar
        # tries to join, so catch it here instead.
        if not _present(url) or not _present(lk_key) or not _present(lk_secret):
            raise SynthesiaError(
                "LiveKit url, API key, and API secret are required: pass them or "
                "set LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET"
            )
        url = _to_ws_url(url)
        if not url.startswith(("ws://", "wss://")):
            raise SynthesiaError(f"livekit_url {url!r} is not a ws:// or wss:// URL")

        self._state = _State.IDLE
        self._close_done.clear()
        self._teardown_task = None

        try:
            await super().start(agent_session, room)
            token = self._mint_token(room=room, lk_key=lk_key, lk_secret=lk_secret)
            client = SynthesiaAPI(api_key=self._config.api_key, api_url=self._config.api_url)
            launch_conn = dataclasses.replace(
                DEFAULT_API_CONNECT_OPTIONS,
                timeout=self._config.join_timeout,
                max_retry=0,
            )
            response = await client.start_session(
                StartSessionRequest(
                    avatar_ids=list(self._config.avatar_ids),
                    livekit_url=url,
                    lk_token=token,
                ),
                conn_options=launch_conn,
            )
            self._session_id = response.session_id

            # TODO: confirm the avatar worker's expected audio sample rate and
            # pass sample_rate explicitly once it is verified against the real
            # worker.
            audio_output = DataStreamAudioOutput(
                room,
                destination_identity=self.avatar_identity,
                wait_remote_track=rtc.TrackKind.KIND_VIDEO,
            )
            # replace_audio_tail keeps any wrapper AgentSession.start() adds
            # later (TranscriptSynchronizer, RecorderAudioOutput) attached.
            agent_session.output.replace_audio_tail(audio_output)
            self._audio_output = audio_output

            await self.wait_for_join(timeout=self._config.join_timeout)
        except asyncio.TimeoutError as e:
            await self.aclose()
            await self._discard_partial_start()
            raise SynthesiaError(
                f"avatar did not join within {self._config.join_timeout}s",
                type=ErrorType.TIMEOUT,
            ) from e
        except BaseException:
            await self.aclose()
            await self._discard_partial_start()
            raise

        if self._state is _State.CLOSED:
            await self._close_done.wait()
            await self._discard_partial_start()
            raise SynthesiaError("avatar session was closed while starting")

        room.on("disconnected", self._on_room_disconnected)
        room.on("track_unpublished", self._on_track_unpublished)
        room.on("participant_disconnected", self._on_participant_disconnected)
        self._state = _State.STARTED
    finally:
        self._starting = False

async def swap_avatar(self, avatar_id: str, *, timeout: float = 15.0) ‑> str

Expand source code

async def swap_avatar(self, avatar_id: str, *, timeout: float = DEFAULT_SWAP_TIMEOUT) -> str:
    """Switch the rendered avatar mid-session.

    ``avatar_id`` must be one of the ids passed in ``AvatarConfig``, or
    ``"default"`` for the first id in that list. Returns the now-active
    avatar id once the swap has taken effect. Raises a ``SynthesiaError``
    of type ``UNKNOWN_AVATAR`` for an id that was not precomputed,
    ``CONNECTION`` if the worker cannot be reached, and no ``type`` if the
    worker rejects the swap.
    """
    if self._state is not _State.STARTED or self._ending() or self._room is None:
        raise SynthesiaError("swap_avatar() requires a started avatar session")
    if avatar_id != "default" and avatar_id not in self._config.avatar_ids:
        raise SynthesiaError(
            f"avatar {avatar_id!r} was not in initial list of avatar_ids",
            type=ErrorType.UNKNOWN_AVATAR,
        )

    try:
        raw = await self._room.local_participant.perform_rpc(
            destination_identity=self.avatar_identity,
            method="swapAvatar",
            payload=json.dumps({"avatar_id": avatar_id}),
            response_timeout=timeout,
        )
    except Exception as e:
        raise SynthesiaError(f"avatar swap RPC failed: {e}", type=ErrorType.CONNECTION) from e

    try:
        response = json.loads(raw)
    except ValueError as e:
        raise SynthesiaError("avatar swap returned a malformed response") from e
    avatar_id_result = response.get("avatar_id") if isinstance(response, dict) else None
    if (
        not isinstance(response, dict)
        or response.get("error")
        or not isinstance(avatar_id_result, str)
    ):
        detail = response.get("error") if isinstance(response, dict) else None
        raise SynthesiaError(f"avatar swap failed: {detail or raw}")
    return avatar_id_result

Switch the rendered avatar mid-session.

avatar_id must be one of the ids passed in AvatarConfig, or "default" for the first id in that list. Returns the now-active avatar id once the swap has taken effect. Raises a SynthesiaError of type UNKNOWN_AVATAR for an id that was not precomputed, CONNECTION if the worker cannot be reached, and no type if the worker rejects the swap.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class ErrorType (*args, **kwds)

Expand source code

class ErrorType(enum.Enum):
    """What went wrong, for callers that want to branch on it."""

    AUTH = "auth"
    """The API key is invalid, expired, or lacks the scope this endpoint requires."""

    FEATURE_NOT_IN_PLAN = "feature_not_in_plan"
    """The workspace's plan does not include interactive avatars."""

    INVALID_ROOM_TOKEN = "invalid_room_token"
    """The room token cannot produce a joined session.

    Distinct from ``AUTH``, which concerns the Synthesia API key. Raised for a
    token that is malformed or missing the attribute naming the agent the
    avatar publishes audio for.
    """

    LIVEKIT_CREDENTIALS_REJECTED = "livekit_credentials_rejected"
    """The LiveKit project the token was signed for would not accept it.

    The token itself is well formed, so the API key and secret it was minted
    with are the thing to check, not the Synthesia credentials.
    """

    INVALID_SESSION_REQUEST = "invalid_session_request"
    """The backend rejected the session request payload."""

    UNKNOWN_AVATAR = "unknown_avatar"
    """Avatar is not in the gallery or not accessible to the workspace."""

    QUOTA_EXCEEDED = "quota_exceeded"
    """The workspace's session quota is exhausted."""

    RATE_LIMITED = "rate_limited"
    """The request was throttled. ``retry_after`` may carry the back-off."""

    CONCURRENCY_LIMIT = "concurrency_limit"
    """Every concurrent-session slot is in use.

    The API sends no ``Retry-After`` for this today, so ``retry_after`` is
    normally ``None``; it is honoured if one ever arrives.
    """

    TIMEOUT = "timeout"
    """The avatar did not join within ``join_timeout``."""

    CONNECTION = "connection"
    """No usable response after retries.

    The last attempt either failed to connect or was answered with a 5xx;
    ``status`` and ``request_id`` describe that answer when there was one.
    """

What went wrong, for callers that want to branch on it.

Ancestors

• enum.Enum

Class variables

var AUTH

The API key is invalid, expired, or lacks the scope this endpoint requires.

var CONCURRENCY_LIMIT

Every concurrent-session slot is in use.

The API sends no Retry-After for this today, so retry_after is normally None; it is honoured if one ever arrives.

var CONNECTION

No usable response after retries.

The last attempt either failed to connect or was answered with a 5xx; status and request_id describe that answer when there was one.

var FEATURE_NOT_IN_PLAN

The workspace's plan does not include interactive avatars.

var INVALID_ROOM_TOKEN

The room token cannot produce a joined session.

Distinct from AUTH, which concerns the Synthesia API key. Raised for a token that is malformed or missing the attribute naming the agent the avatar publishes audio for.

var INVALID_SESSION_REQUEST

The backend rejected the session request payload.

var LIVEKIT_CREDENTIALS_REJECTED

The LiveKit project the token was signed for would not accept it.

The token itself is well formed, so the API key and secret it was minted with are the thing to check, not the Synthesia credentials.

var QUOTA_EXCEEDED

The workspace's session quota is exhausted.

var RATE_LIMITED

The request was throttled. retry_after may carry the back-off.

var TIMEOUT

The avatar did not join within join_timeout.

var UNKNOWN_AVATAR

Avatar is not in the gallery or not accessible to the workspace.

class SynthesiaError (message: str,
*,
type: ErrorType | None = None,
body: object | None = None,
retryable: bool | None = None,
retry_after: float | None = None,
status: int | None = None,
request_id: str | None = None)

Expand source code

class SynthesiaError(APIError):
    """Every error the Synthesia plugin raises. Also a :class:`~livekit.agents.APIError`.

    - ``type`` identifies what went wrong; see :class:`ErrorType`.
    - ``retryable`` reports whether retrying the same call could plausibly
      succeed, and defaults based on ``type`` unless overridden.
    - ``retry_after`` carries the server-provided back-off in seconds for a
      ``RATE_LIMITED`` or ``CONCURRENCY_LIMIT`` error, when the backend
      supplied one.
    - ``status`` is the HTTP status the API answered, or ``None`` when no
      answer arrived.
    - ``request_id`` is the API's ``requestId`` when the body carried one.
      Quote it when contacting support.
    """

    def __init__(
        self,
        message: str,
        *,
        type: ErrorType | None = None,
        body: object | None = None,
        retryable: bool | None = None,
        retry_after: float | None = None,
        status: int | None = None,
        request_id: str | None = None,
    ) -> None:
        if not isinstance(message, str):
            message = _stringify(message)
        self.type = type
        self.retry_after = retry_after
        self.status = status
        self.request_id = request_id
        if retryable is None:
            retryable = type in _RETRYABLE_TYPES
        super().__init__(message, body=body, retryable=retryable)

Every error the Synthesia plugin raises. Also a :class:~livekit.agents.APIError.

• type identifies what went wrong; see :class:ErrorType.
• retryable reports whether retrying the same call could plausibly succeed, and defaults based on type unless overridden.
• retry_after carries the server-provided back-off in seconds for a RATE_LIMITED or CONCURRENCY_LIMIT error, when the backend supplied one.
• status is the HTTP status the API answered, or None when no answer arrived.
• request_id is the API's requestId when the body carried one. Quote it when contacting support.

Ancestors

• livekit.agents._exceptions.APIError
• builtins.Exception
• builtins.BaseException