Module livekit.plugins.thunderphone
ThunderPhone plugin for LiveKit Agents
ThunderPhone voice agents as a realtime (speech-to-speech) model.
Sub-modules
livekit.plugins.thunderphone.realtime
Classes
class RealtimeModel (*,
api_key: str | None = None,
agent_id: int | str | None = None,
model: str | None = None,
voice: NotGivenOr[str] = NOT_GIVEN,
language: str | None = None,
caller_number: str | None = None,
callee_number: str | None = None,
live_transcripts: bool = False,
base_url: str = 'wss://api.thunderphone.com/v1/realtime',
http_session: Any = None,
conn_options: APIConnectOptions | None = None,
**kwargs: Any)-
Expand source code
class RealtimeModel(_openai.RealtimeModel): """Run a ThunderPhone voice agent as the realtime model of an AgentSession. Two ways to use it: * **Saved agent** — pass ``agent_id``. The agent's prompt, voice, engine, languages, tools, greeting and silence handling come from ThunderPhone; the LiveKit Agent's instructions and tools are not sent. * **Inline** — omit ``agent_id``. Instructions and tools come from the LiveKit Agent, ``model`` picks the engine (``spark``, ``bolt`` or ``storm``) and ``voice`` the ThunderPhone voice. Function tools run in the LiveKit agent. An inline call freezes its instructions and tools once it starts, so a handoff to an Agent configured differently starts a new call; with a saved agent, handoffs keep the call. Every call ends with ThunderPhone closing the socket; the session does not reconnect, because a reconnect would start a new, separately billed call. """ def __init__( self, *, api_key: str | None = None, agent_id: int | str | None = None, model: str | None = None, voice: NotGivenOr[str] = NOT_GIVEN, language: str | None = None, caller_number: str | None = None, callee_number: str | None = None, live_transcripts: bool = False, base_url: str = DEFAULT_BASE_URL, http_session: Any = None, conn_options: APIConnectOptions | None = None, **kwargs: Any, ) -> None: """Create a ThunderPhone realtime model. Args: api_key: ThunderPhone secret key (``sk_live_...``). Defaults to the ``THUNDERPHONE_API_KEY`` environment variable. agent_id: Id of a saved ThunderPhone agent. Mutually exclusive with ``model`` and ``voice``. model: Engine for inline sessions: ``"spark"``, ``"bolt"`` or ``"storm"``. voice: ThunderPhone voice name for inline sessions. language: Primary language hint for inline sessions, e.g. ``"es"``. caller_number: Phone number the call came from, when the session fronts a phone call. Shown as the call's "from" number in ThunderPhone call history. callee_number: Phone number that was dialed. Shown as the call's "to" number. live_transcripts: Stream caller transcript fragments while the caller is still speaking (billed extra). Off by default: one transcript per turn. base_url: WebSocket endpoint. Defaults to the ThunderPhone API. http_session: Optional aiohttp session to reuse. conn_options: Connection options. Retries default to 0: a ThunderPhone session is a call, and reconnecting would start a new one. **kwargs: Passed to ``livekit.plugins.openai.realtime.RealtimeModel`` (``turn_detection``, ``modalities``, ...). ThunderPhone transcribes the caller with the call's own engine, so ``input_audio_transcription`` is not sent; ``None`` still turns off the caller transcripts. """ key = (api_key or os.environ.get(API_KEY_ENV, "")).strip() if not key.startswith(API_KEY_PREFIX): raise ValueError( f"ThunderPhone needs a secret API key starting with '{API_KEY_PREFIX}' " f"(pass api_key= or set {API_KEY_ENV})." ) if agent_id is not None and (model or is_given(voice)): raise ValueError( "agent_id sessions take their engine and voice from the saved agent; " "drop model/voice or drop agent_id." ) if agent_id is not None and not str(agent_id).strip(): # The URL would drop it and the server would start an inline call, # while the plugin strips the agent's instructions and tools. raise ValueError("agent_id must not be empty") self._tp_agent_id = str(agent_id).strip() if agent_id is not None else None self._tp_engine = model self._tp_live_transcripts = bool(live_transcripts) url = _build_url( base_url, agent_id=self._tp_agent_id, product=model, language=language, from_number=caller_number, to_number=callee_number, ) # The framework recycles an OpenAI socket after max_session_duration # (20 min by default) by closing it and reconnecting. On ThunderPhone # that would hang up and start a second, separately billed call. kwargs["max_session_duration"] = None super().__init__( model=SERVER_MODEL, voice=voice if is_given(voice) else "default", base_url=url, api_key=key, http_session=http_session, conn_options=conn_options or APIConnectOptions(max_retry=0), **kwargs, ) self._provider_label = "ThunderPhone" # An inline call freezes its instructions and tools once it starts, so a # handoff to an agent configured differently gets a new session (and a # new call). A saved agent owns its instructions and tools and the # plugin never sends them, so updates are no-ops and every handoff keeps # the call. Turn taking is ThunderPhone's and always on; the framework # must not run its own VAD or turn detector against it. agent_mode = self._tp_agent_id is not None self._capabilities = dataclasses.replace( self._capabilities, mutable_instructions=agent_mode, mutable_tools=agent_mode, turn_detection=True, can_disable_turn_detection=False, ) @property def model(self) -> str: # The engine when one was chosen; a saved agent's engine lives on ThunderPhone. return self._tp_engine or SERVER_MODEL @property def provider(self) -> str: return "ThunderPhone" def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession: if turn_detection_disabled: logger.warning( "ThunderPhone turn detection is server-side and always on; " "turn_detection_disabled is ignored" ) sess = RealtimeSession(self) self._sessions.add(sess) return sessRun a ThunderPhone voice agent as the realtime model of an AgentSession.
Two ways to use it:
- Saved agent — pass
agent_id. The agent's prompt, voice, engine, languages, tools, greeting and silence handling come from ThunderPhone; the LiveKit Agent's instructions and tools are not sent. - Inline — omit
agent_id. Instructions and tools come from the LiveKit Agent,modelpicks the engine (spark,boltorstorm) andvoicethe ThunderPhone voice. Function tools run in the LiveKit agent.
An inline call freezes its instructions and tools once it starts, so a handoff to an Agent configured differently starts a new call; with a saved agent, handoffs keep the call. Every call ends with ThunderPhone closing the socket; the session does not reconnect, because a reconnect would start a new, separately billed call.
Create a ThunderPhone realtime model.
Args
api_key- ThunderPhone secret key (
sk_live_…). Defaults to theTHUNDERPHONE_API_KEYenvironment variable. agent_id- Id of a saved ThunderPhone agent. Mutually exclusive with
modelandvoice. model- Engine for inline sessions:
"spark","bolt"or"storm". voice- ThunderPhone voice name for inline sessions.
language- Primary language hint for inline sessions, e.g.
"es". caller_number- Phone number the call came from, when the session fronts a phone call. Shown as the call's "from" number in ThunderPhone call history.
callee_number- Phone number that was dialed. Shown as the call's "to" number.
live_transcripts- Stream caller transcript fragments while the caller is still speaking (billed extra). Off by default: one transcript per turn.
base_url- WebSocket endpoint. Defaults to the ThunderPhone API.
http_session- Optional aiohttp session to reuse.
conn_options- Connection options. Retries default to 0: a ThunderPhone session is a call, and reconnecting would start a new one.
**kwargs- Passed to
RealtimeModel(turn_detection,modalities, …). ThunderPhone transcribes the caller with the call's own engine, soinput_audio_transcriptionis not sent;Nonestill turns off the caller transcripts.
Ancestors
- livekit.plugins.openai.realtime.realtime_model.RealtimeModel
- livekit.agents.llm._realtime.openai.RealtimeModel
- livekit.agents.llm.realtime.RealtimeModel
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: # The engine when one was chosen; a saved agent's engine lives on ThunderPhone. return self._tp_engine or SERVER_MODEL prop provider : str-
Expand source code
@property def provider(self) -> str: return "ThunderPhone"
Methods
def session(self, *, turn_detection_disabled: bool = False) ‑> RealtimeSession-
Expand source code
def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession: if turn_detection_disabled: logger.warning( "ThunderPhone turn detection is server-side and always on; " "turn_detection_disabled is ignored" ) sess = RealtimeSession(self) self._sessions.add(sess) return sessCreate a new session, optionally with server-side turn detection disabled.
turn_detection_disabledis honored only by plugins reportingcan_disable_turn_detection; the model itself is left unchanged and reusable.
- Saved agent — pass
class RealtimeSession (realtime_model: RealtimeModel)-
Expand source code
class RealtimeSession(_openai.RealtimeSession): """One ThunderPhone call. Emits, in addition to the framework events: * ``thunderphone_call_event`` with every ``call.*`` event (transfer, keypad, speech ignored, ...); * ``thunderphone_call_ended`` with the ``call.ended`` event, after which the session closes itself. ``call_id`` holds the ThunderPhone call id once the session is live, for fetching the recording, transcript and grade afterwards. """ def __init__(self, realtime_model: RealtimeModel) -> None: self._tp_model = realtime_model self._tp_agent_mode = realtime_model._tp_agent_id is not None # Inline sessions: hold session.update until the agent has supplied its # instructions and tools, then send one update that starts the call. self._tp_holding = not self._tp_agent_mode self._tp_pending_session: dict[str, Any] = {} self._tp_call_ended = False self._tp_warned_reply_instructions = False # Function call ids longer than OpenAI's 32-character cap come back # from the framework hashed; map the hash back to the id the server # expects on the function_call_output. self._tp_call_ids: dict[str, str] = {} self.call_id: int | None = None super().__init__(realtime_model) self.on("openai_server_event_received", self._tp_on_server_event) # ----------------------------------------------------------- outgoing def send_event(self, event: Any) -> None: payload = ( event.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=False) if isinstance(event, BaseModel) else dict(event) ) if payload.get("type") == "session.update": payload = self._tp_shape_session_update(payload) if self._tp_holding: _merge(self._tp_pending_session, payload.get("session") or {}) return elif self._tp_holding and payload.get("type") != "input_audio_buffer.append": # Anything but audio is meaningful only once the call exists; # flush the configuration first so ordering is preserved. self._tp_flush() if payload.get("type") == "response.create": response = payload.get("response") if isinstance(response, dict): for key in _RESPONSE_OVERRIDE_KEYS: response.pop(key, None) if payload.get("type") == "conversation.item.create": item = payload.get("item") if isinstance(item, dict) and item.get("type") == "function_call_output": call_id = item.get("call_id") if isinstance(call_id, str): item["call_id"] = self._tp_call_ids.get(call_id, call_id) super().send_event(payload) def _tp_shape_session_update(self, payload: dict[str, Any]) -> dict[str, Any]: session = dict(payload.get("session") or {}) session["model"] = SERVER_MODEL for key in _OPENAI_ONLY_KEYS: session.pop(key, None) audio = session.get("audio") leg = audio.get("input") if isinstance(audio, dict) else None if isinstance(leg, dict): # ThunderPhone transcribes the caller with the call's own engine. leg.pop("transcription", None) if self._tp_agent_mode: # A saved agent is already running with its own prompt, tools, # voice, transcription and turn-taking; the server rejects any # change to those. Keep only what a client may still set. for key in _AGENT_OWNED_KEYS: session.pop(key, None) session.pop("output_modalities", None) audio = session.get("audio") if isinstance(audio, dict): for direction in ("input", "output"): leg = audio.get(direction) if isinstance(leg, dict): audio[direction] = {k: v for k, v in leg.items() if k == "format"} else: config = dict(session.get("config") or {}) # Opt into ThunderPhone's call.* events (hang-up reason, transfer, keypad). config.setdefault("call_events", True) session["config"] = config if self._tp_model._tp_live_transcripts: session["live_transcripts"] = True payload["session"] = session return payload def _tp_flush(self) -> None: if not self._tp_holding: return self._tp_holding = False if self._tp_pending_session: super().send_event( { "event_id": utils.shortuuid("session_update_"), "type": "session.update", "session": self._tp_pending_session, } ) self._tp_pending_session = {} def generate_reply( self, *, instructions: NotGivenOr[str] = NOT_GIVEN, tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN, tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN, ) -> asyncio.Future[llm.GenerationCreatedEvent]: if is_given(instructions) and not self._tp_warned_reply_instructions: self._tp_warned_reply_instructions = True logger.warning( "ThunderPhone ignores per-reply instructions; " "replies use the session's instructions" ) return super().generate_reply( instructions=instructions, tool_choice=tool_choice, tools=tools ) async def _update_session( self, *, instructions: NotGivenOr[str] = NOT_GIVEN, chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN, tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN, ) -> None: # The framework sends instructions, then chat history, then tools. On # ThunderPhone the first session.update starts the call and freezes # instructions and tools, so send both before any conversation item. if is_given(instructions): try: await self.update_instructions(instructions) except llm.RealtimeError: logger.exception("failed to update the instructions") if is_given(tools): try: await self.update_tools(tools) except llm.RealtimeError: logger.exception("failed to update the tools") self._tp_flush() if is_given(chat_ctx): try: await self.update_chat_ctx(chat_ctx) except llm.RealtimeError: logger.exception("failed to update the chat_ctx") # ----------------------------------------------------------- incoming def _emit_transcription_metrics(self, event: Any) -> None: # The base names the model after its OpenAI transcription config, which # ThunderPhone never uses: the call's engine transcribes the caller and # reports usage as seconds of caller audio (never tokens). try: usage = UsageTranscriptTextUsageDuration.model_validate(event.usage) except ValidationError: return model = self._tp_model self.emit( "metrics_collected", STTMetrics( request_id=event.event_id, timestamp=time.time(), duration=0.0, label=model.label, audio_duration=usage.seconds, streamed=True, metadata=Metadata(model_name=model.model, model_provider=model.provider), ), ) def _tp_close_expected(self) -> bool: # After call.ended the server hangs up; after aclose() we do. return self._tp_call_ended or self._closing async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse: ws = await super()._create_ws_conn() return cast(aiohttp.ClientWebSocketResponse, _CallSocket(ws, self)) async def _run_ws(self, ws_conn: aiohttp.ClientWebSocketResponse) -> None: try: await super()._run_ws(ws_conn) except APIConnectionError: # The socket closed after the call ended or while closing, and our # send loop did not get there first (see _CallSocket). With the # outgoing channel closed the framework ends the session here. if not self._tp_close_expected(): raise def _tp_on_server_event(self, event: dict[str, Any]) -> None: event_type = event.get("type") if event_type == "response.created": self._tp_claim_response(event) return if event_type == "response.output_item.done": item = event.get("item") if isinstance(item, dict) and item.get("type") == "function_call": call_id = item.get("call_id") if isinstance(call_id, str): self._tp_call_ids[_shorten_call_id(call_id)] = call_id return if event_type in ("session.created", "session.updated"): session = event.get("session") or {} call_id = session.get("call_id") if isinstance(session, dict) else None if isinstance(call_id, int): self.call_id = call_id elif isinstance(call_id, str) and call_id.isdigit(): self.call_id = int(call_id) return if not isinstance(event_type, str) or not event_type.startswith("call."): return self.emit(CALL_EVENT, event) if event_type == "call.ended" and not self._tp_call_ended: self._tp_call_ended = True logger.info("ThunderPhone call ended", extra={"lk.pii.event": event}) self.emit(CALL_ENDED_EVENT, event) # The server closes the socket right after this event. Closing the # outgoing channel ends the send loop, which closes our side and # ends the session; _CallSocket holds the server's close back until # then so it is not read as a connection failure. Anything still # queued was meant for a call that no longer exists. self._msg_ch.close() while not self._msg_ch.empty(): self._msg_ch.recv_nowait() def _tp_claim_response(self, event: dict[str, Any]) -> None: # generate_reply() resolves its future by the client_event_id the # server echoes in response.metadata. ThunderPhone servers that # predate that echo start the response but never label it, so the # future times out while the audio plays anyway. Credit an unlabeled # response to the oldest pending reply request instead; the hook runs # on the raw dict before the framework parses it. response = event.get("response") if not isinstance(response, dict): return metadata = response.get("metadata") if isinstance(metadata, dict) and metadata.get("client_event_id"): return pending = next(iter(self._response_created_futures), None) if pending is None: return response["metadata"] = { **(metadata if isinstance(metadata, dict) else {}), "client_event_id": pending, }One ThunderPhone call.
Emits, in addition to the framework events:
thunderphone_call_eventwith everycall.*event (transfer, keypad, speech ignored, …);thunderphone_call_endedwith thecall.endedevent, after which the session closes itself.
call_idholds the ThunderPhone call id once the session is live, for fetching the recording, transcript and grade afterwards.Ancestors
- livekit.agents.llm._realtime.openai.RealtimeSession
- livekit.agents.llm.realtime.RealtimeSession
- abc.ABC
- EventEmitter
- typing.Generic
Methods
def generate_reply(self,
*,
instructions: NotGivenOr[str] = NOT_GIVEN,
tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN,
tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN) ‑> _asyncio.Future[livekit.agents.llm.realtime.GenerationCreatedEvent]-
Expand source code
def generate_reply( self, *, instructions: NotGivenOr[str] = NOT_GIVEN, tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN, tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN, ) -> asyncio.Future[llm.GenerationCreatedEvent]: if is_given(instructions) and not self._tp_warned_reply_instructions: self._tp_warned_reply_instructions = True logger.warning( "ThunderPhone ignores per-reply instructions; " "replies use the session's instructions" ) return super().generate_reply( instructions=instructions, tool_choice=tool_choice, tools=tools ) def send_event(self, event: Any) ‑> None-
Expand source code
def send_event(self, event: Any) -> None: payload = ( event.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=False) if isinstance(event, BaseModel) else dict(event) ) if payload.get("type") == "session.update": payload = self._tp_shape_session_update(payload) if self._tp_holding: _merge(self._tp_pending_session, payload.get("session") or {}) return elif self._tp_holding and payload.get("type") != "input_audio_buffer.append": # Anything but audio is meaningful only once the call exists; # flush the configuration first so ordering is preserved. self._tp_flush() if payload.get("type") == "response.create": response = payload.get("response") if isinstance(response, dict): for key in _RESPONSE_OVERRIDE_KEYS: response.pop(key, None) if payload.get("type") == "conversation.item.create": item = payload.get("item") if isinstance(item, dict) and item.get("type") == "function_call_output": call_id = item.get("call_id") if isinstance(call_id, str): item["call_id"] = self._tp_call_ids.get(call_id, call_id) super().send_event(payload)
Inherited members