Module livekit.plugins.lemonslice.api
Classes
class LemonSliceAPI (api_key: NotGivenOr[str] = NOT_GIVEN,
api_url: NotGivenOr[str] = NOT_GIVEN,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0),
session: aiohttp.ClientSession | None = None)-
Expand source code
class LemonSliceAPI: def __init__( self, api_key: NotGivenOr[str] = NOT_GIVEN, api_url: NotGivenOr[str] = NOT_GIVEN, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, session: aiohttp.ClientSession | None = None, ) -> None: """ Initializes the LemonSliceAPI client. Args: api_key: Your LemonSlice API key. If not provided, it is read from the LEMONSLICE_API_KEY environment variable. api_url: The base URL of the LemonSlice API. conn_options: Connection options for the aiohttp session. session: An optional existing aiohttp.ClientSession to use for requests. """ ls_api_key = api_key if utils.is_given(api_key) else os.getenv("LEMONSLICE_API_KEY") if not ls_api_key: raise LemonSliceException("LEMONSLICE_API_KEY must be set") self._api_key = ls_api_key self._api_url = api_url or DEFAULT_API_URL self._conn_options = conn_options self._session = session self._owns_session = session is None async def __aenter__(self) -> LemonSliceAPI: if self._owns_session: self._session = aiohttp.ClientSession() return self async def __aexit__( self, exc_type: type | None, exc_val: Exception | None, exc_tb: Any ) -> None: if self._owns_session and self._session and not self._session.closed: await self._session.close() async def start_agent_session( self, *, livekit_url: str, livekit_token: str, livekit_session_id: str, agent_id: NotGivenOr[str] = NOT_GIVEN, agent_image_url: NotGivenOr[str] = NOT_GIVEN, agent_image: NotGivenOr[Image.Image] = NOT_GIVEN, agent_prompt: NotGivenOr[str] = NOT_GIVEN, agent_idle_prompt: NotGivenOr[str] = NOT_GIVEN, idle_timeout: NotGivenOr[int] = NOT_GIVEN, audio_sample_rate: NotGivenOr[int] = NOT_GIVEN, model: NotGivenOr[str] = NOT_GIVEN, extra_payload: NotGivenOr[dict[str, Any]] = NOT_GIVEN, ) -> str: """ Initiates a new LemonSlice agent session. Args: livekit_url: The LiveKit Cloud server URL. livekit_token: The LiveKit access token for the agent. livekit_session_id: LiveKit room session ID (room SID). agent_id: The ID of the LemonSlice agent to add to the session. agent_image_url: The URL of the image to use as the agent's avatar. agent_image: A PIL image to upload and use as the agent's avatar. Sent to LemonSlice as a multipart image upload. agent_prompt: A prompt that subtly influences the avatar's movements and expressions while responding. agent_idle_prompt: A prompt that subtly influences the avatar's movements and expressions while idle. idle_timeout: The idle timeout, in seconds. audio_sample_rate: Sample rate of the audio sent to the avatar. One of 16000, 22050, 24000, 44100 or 48000. model: The LemonSlice model to run the avatar on. Uses the default model when not given. extra_payload: Additional payload to include in the request. Returns: The unique session ID for the LemonSlice agent session. """ given_sources = [ source for source in (agent_id, agent_image_url, agent_image) if utils.is_given(source) ] if len(given_sources) == 0: raise LemonSliceException("Missing one of agent_id, agent_image_url or agent_image") if len(given_sources) > 1: raise LemonSliceException( "Only one of agent_id, agent_image_url or agent_image can be provided" ) payload: dict[str, Any] = { "transport_type": "livekit", "properties": { "livekit_url": livekit_url, "livekit_token": livekit_token, "livekit_session_id": livekit_session_id, }, } image_bytes: bytes | None = None if utils.is_given(agent_id): payload["agent_id"] = agent_id if utils.is_given(agent_image_url): payload["agent_image_url"] = agent_image_url if utils.is_given(agent_prompt): payload["agent_prompt"] = agent_prompt if utils.is_given(agent_idle_prompt): payload["agent_idle_prompt"] = agent_idle_prompt if utils.is_given(idle_timeout): payload["idle_timeout"] = idle_timeout if utils.is_given(audio_sample_rate): payload["audio_sample_rate"] = audio_sample_rate if utils.is_given(model): payload["model"] = model if utils.is_given(extra_payload): payload.update(extra_payload) if utils.is_given(agent_image): image_bytes = _encode_image(agent_image) response_data = await self._post(payload, image_bytes=image_bytes) session_id = response_data["session_id"] logger.debug(f"LemonSlice Session ID = {session_id}") return session_id # type: ignore async def join_meeting( self, session_id: str, *, meeting_url: str, livekit_url: str, broadcast_token: str, bot_name: NotGivenOr[str] = NOT_GIVEN, ) -> JoinMeetingResult: """Add an active avatar session to an external video meeting. Supports Zoom, Google Meet, Microsoft Teams, and Webex. Args: session_id: LemonSlice agent session ID. meeting_url: URL of the external meeting to join. livekit_url: LiveKit server URL for the agent room. broadcast_token: LiveKit token used to subscribe to avatar media. bot_name: Optional display name for the bot in the meeting. Returns: JoinMeetingResult with relay WebSocket URL and meeting bot ID. """ payload: dict[str, Any] = { "session_id": session_id, "meeting_url": meeting_url, "livekit_url": livekit_url, "broadcast_token": broadcast_token, } if utils.is_given(bot_name) and bot_name: payload["bot_name"] = bot_name url = f"{self._api_url.rstrip('/')}/{session_id}/join-meeting" data = await self._post(payload, url=url) return JoinMeetingResult( websocket_url=str(data["websocket_url"]), meeting_bot_id=str(data["meeting_bot_id"]), ) async def leave_meeting( self, session_id: str, *, meeting_bot_id: str, ) -> None: """Remove the avatar from an external meeting. Args: session_id: LemonSlice agent session ID. meeting_bot_id: Meeting bot ID returned by join_meeting(). """ url = f"{self._api_url.rstrip('/')}/{session_id}/leave-meeting" await self._post( {"meeting_bot_id": meeting_bot_id}, url=url, ) async def _post( self, payload: dict[str, Any], *, url: str | None = None, image_bytes: bytes | None = None, ) -> dict[str, Any]: """ Make a POST request to the LemonSlice API with retry logic. Args: payload: JSON payload for the request. url: Optional URL override. image_bytes: Optional PNG-encoded image. Returns: Response data as a dictionary Raises: APIConnectionError: If the request fails after all retries """ session = self._session or aiohttp.ClientSession() try: for i in range(self._conn_options.max_retry + 1): try: headers = {"X-API-Key": self._api_key} request_kwargs: dict[str, Any] if image_bytes is not None: form = aiohttp.FormData() form.add_field( "payload", json.dumps(payload), content_type="application/json" ) # Upload the image using multipart form.add_field( "image", image_bytes, filename="image.png", content_type="image/png", ) request_kwargs = {"data": form} else: headers["Content-Type"] = "application/json" request_kwargs = {"json": payload} async with session.post( url or self._api_url, headers=headers, timeout=aiohttp.ClientTimeout( total=60.0, sock_connect=self._conn_options.timeout, ), **request_kwargs, ) as response: if not response.ok: text = await response.text() raise APIStatusError( "LemonSlice Server returned an error", status_code=response.status, body=text, ) return await response.json() # type: ignore except Exception as e: if isinstance(e, APIStatusError): logger.error( "LemonSlice API returned an error", extra={ "status_code": e.status_code, "body": e.body, }, ) if not e.retryable: raise e elif isinstance(e, APIConnectionError): logger.warning("failed to call LemonSlice api", extra={"error": str(e)}) else: logger.exception("failed to call lemonslice api") if i < self._conn_options.max_retry: await asyncio.sleep(self._conn_options._interval_for_retry(i)) finally: if not self._session: # if we created the session, we close it await session.close() raise APIConnectionError("Failed to call LemonSlice API after all retries")Initializes the LemonSliceAPI client.
Args
api_key- Your LemonSlice API key. If not provided, it is read from the LEMONSLICE_API_KEY environment variable.
api_url- The base URL of the LemonSlice API.
conn_options- Connection options for the aiohttp session.
session- An optional existing aiohttp.ClientSession to use for requests.
Methods
async def join_meeting(self,
session_id: str,
*,
meeting_url: str,
livekit_url: str,
broadcast_token: str,
bot_name: NotGivenOr[str] = NOT_GIVEN) ‑> JoinMeetingResult-
Expand source code
async def join_meeting( self, session_id: str, *, meeting_url: str, livekit_url: str, broadcast_token: str, bot_name: NotGivenOr[str] = NOT_GIVEN, ) -> JoinMeetingResult: """Add an active avatar session to an external video meeting. Supports Zoom, Google Meet, Microsoft Teams, and Webex. Args: session_id: LemonSlice agent session ID. meeting_url: URL of the external meeting to join. livekit_url: LiveKit server URL for the agent room. broadcast_token: LiveKit token used to subscribe to avatar media. bot_name: Optional display name for the bot in the meeting. Returns: JoinMeetingResult with relay WebSocket URL and meeting bot ID. """ payload: dict[str, Any] = { "session_id": session_id, "meeting_url": meeting_url, "livekit_url": livekit_url, "broadcast_token": broadcast_token, } if utils.is_given(bot_name) and bot_name: payload["bot_name"] = bot_name url = f"{self._api_url.rstrip('/')}/{session_id}/join-meeting" data = await self._post(payload, url=url) return JoinMeetingResult( websocket_url=str(data["websocket_url"]), meeting_bot_id=str(data["meeting_bot_id"]), )Add an active avatar session to an external video meeting.
Supports Zoom, Google Meet, Microsoft Teams, and Webex.
Args
session_id- LemonSlice agent session ID.
meeting_url- URL of the external meeting to join.
livekit_url- LiveKit server URL for the agent room.
broadcast_token- LiveKit token used to subscribe to avatar media.
bot_name- Optional display name for the bot in the meeting.
Returns
JoinMeetingResult with relay WebSocket URL and meeting bot ID.
async def leave_meeting(self, session_id: str, *, meeting_bot_id: str) ‑> None-
Expand source code
async def leave_meeting( self, session_id: str, *, meeting_bot_id: str, ) -> None: """Remove the avatar from an external meeting. Args: session_id: LemonSlice agent session ID. meeting_bot_id: Meeting bot ID returned by join_meeting(). """ url = f"{self._api_url.rstrip('/')}/{session_id}/leave-meeting" await self._post( {"meeting_bot_id": meeting_bot_id}, url=url, )Remove the avatar from an external meeting.
Args
session_id- LemonSlice agent session ID.
meeting_bot_id- Meeting bot ID returned by join_meeting().
async def start_agent_session(self,
*,
livekit_url: str,
livekit_token: str,
livekit_session_id: str,
agent_id: NotGivenOr[str] = NOT_GIVEN,
agent_image_url: NotGivenOr[str] = NOT_GIVEN,
agent_image: NotGivenOr[Image.Image] = NOT_GIVEN,
agent_prompt: NotGivenOr[str] = NOT_GIVEN,
agent_idle_prompt: NotGivenOr[str] = NOT_GIVEN,
idle_timeout: NotGivenOr[int] = NOT_GIVEN,
audio_sample_rate: NotGivenOr[int] = NOT_GIVEN,
model: NotGivenOr[str] = NOT_GIVEN,
extra_payload: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> str-
Expand source code
async def start_agent_session( self, *, livekit_url: str, livekit_token: str, livekit_session_id: str, agent_id: NotGivenOr[str] = NOT_GIVEN, agent_image_url: NotGivenOr[str] = NOT_GIVEN, agent_image: NotGivenOr[Image.Image] = NOT_GIVEN, agent_prompt: NotGivenOr[str] = NOT_GIVEN, agent_idle_prompt: NotGivenOr[str] = NOT_GIVEN, idle_timeout: NotGivenOr[int] = NOT_GIVEN, audio_sample_rate: NotGivenOr[int] = NOT_GIVEN, model: NotGivenOr[str] = NOT_GIVEN, extra_payload: NotGivenOr[dict[str, Any]] = NOT_GIVEN, ) -> str: """ Initiates a new LemonSlice agent session. Args: livekit_url: The LiveKit Cloud server URL. livekit_token: The LiveKit access token for the agent. livekit_session_id: LiveKit room session ID (room SID). agent_id: The ID of the LemonSlice agent to add to the session. agent_image_url: The URL of the image to use as the agent's avatar. agent_image: A PIL image to upload and use as the agent's avatar. Sent to LemonSlice as a multipart image upload. agent_prompt: A prompt that subtly influences the avatar's movements and expressions while responding. agent_idle_prompt: A prompt that subtly influences the avatar's movements and expressions while idle. idle_timeout: The idle timeout, in seconds. audio_sample_rate: Sample rate of the audio sent to the avatar. One of 16000, 22050, 24000, 44100 or 48000. model: The LemonSlice model to run the avatar on. Uses the default model when not given. extra_payload: Additional payload to include in the request. Returns: The unique session ID for the LemonSlice agent session. """ given_sources = [ source for source in (agent_id, agent_image_url, agent_image) if utils.is_given(source) ] if len(given_sources) == 0: raise LemonSliceException("Missing one of agent_id, agent_image_url or agent_image") if len(given_sources) > 1: raise LemonSliceException( "Only one of agent_id, agent_image_url or agent_image can be provided" ) payload: dict[str, Any] = { "transport_type": "livekit", "properties": { "livekit_url": livekit_url, "livekit_token": livekit_token, "livekit_session_id": livekit_session_id, }, } image_bytes: bytes | None = None if utils.is_given(agent_id): payload["agent_id"] = agent_id if utils.is_given(agent_image_url): payload["agent_image_url"] = agent_image_url if utils.is_given(agent_prompt): payload["agent_prompt"] = agent_prompt if utils.is_given(agent_idle_prompt): payload["agent_idle_prompt"] = agent_idle_prompt if utils.is_given(idle_timeout): payload["idle_timeout"] = idle_timeout if utils.is_given(audio_sample_rate): payload["audio_sample_rate"] = audio_sample_rate if utils.is_given(model): payload["model"] = model if utils.is_given(extra_payload): payload.update(extra_payload) if utils.is_given(agent_image): image_bytes = _encode_image(agent_image) response_data = await self._post(payload, image_bytes=image_bytes) session_id = response_data["session_id"] logger.debug(f"LemonSlice Session ID = {session_id}") return session_id # type: ignoreInitiates a new LemonSlice agent session.
Args
livekit_url- The LiveKit Cloud server URL.
livekit_token- The LiveKit access token for the agent.
livekit_session_id- LiveKit room session ID (room SID).
agent_id- The ID of the LemonSlice agent to add to the session.
agent_image_url- The URL of the image to use as the agent's avatar.
agent_image- A PIL image to upload and use as the agent's avatar. Sent to LemonSlice as a multipart image upload.
agent_prompt- A prompt that subtly influences the avatar's movements and expressions while responding.
agent_idle_prompt- A prompt that subtly influences the avatar's movements and expressions while idle.
idle_timeout- The idle timeout, in seconds.
audio_sample_rate- Sample rate of the audio sent to the avatar. One of 16000, 22050, 24000, 44100 or 48000.
model- The LemonSlice model to run the avatar on. Uses the default model when not given.
extra_payload- Additional payload to include in the request.
Returns
The unique session ID for the LemonSlice agent session.
class LemonSliceException (*args, **kwargs)-
Expand source code
class LemonSliceException(Exception): """Exception for LemonSlice errors"""Exception for LemonSlice errors
Ancestors
- builtins.Exception
- builtins.BaseException