Use ThunderPhone from Pipecat
Run a ThunderPhone voice agent as the speech-to-speech service inside a Pipecat pipeline, with your own transport and telephony.
Pipecat is an open-source framework for building voice agents out of composable services. ThunderPhone plugs in as a speech-to-speech LLM service: Pipecat sends caller audio, ThunderPhone returns the agent's voice, transcripts and function calls, and Pipecat's transport (Daily, LiveKit, Twilio, WebRTC, a local microphone) carries the audio. Speech recognition, the language model, the voice, turn-taking, 47 languages and tools all run on ThunderPhone.
Calls made this way appear in call history and are billed at your product's per-minute rate like any other realtime call. There is no subscription and no ThunderPhone phone number involved.
Install
pip install pipecat-thunderphone
export THUNDERPHONE_API_KEY=sk_live_... # a secret API keyThe package wraps Pipecat's OpenAI Realtime service, because ThunderPhone's
Realtime WebSocket speaks the same protocol. It
needs pipecat-ai 1.8 or newer.
Run a saved agent
Everything the agent does (prompt, voice, engine, languages, tools, greeting, silence check-ins) is configured on ThunderPhone. The pipeline only moves audio.
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat_thunderphone import ThunderPhoneRealtimeLLMService
llm = ThunderPhoneRealtimeLLMService(agent_id=12)
context = LLMContext()
aggregators = LLMContextAggregatorPair(context)
pipeline = Pipeline([
transport.input(),
aggregators.user(),
llm,
aggregators.assistant(),
transport.output(),
])
worker = PipelineWorker(pipeline, params=PipelineParams(allow_interruptions=True))A saved agent opens the call on its own schedule, so the service does not
request an opening response from Pipecat. Pass greet_on_connect=True if you
want one anyway.
Configure the session inline
Without agent_id, instructions and tools come from the Pipecat context, the
same way they do for OpenAI. product picks the engine and voice picks the
ThunderPhone voice.
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.services.llm_service import FunctionCallParams
llm = ThunderPhoneRealtimeLLMService(product="bolt", voice="olivia", language="es")
async def check_availability(params: FunctionCallParams):
slots = await calendar.free_slots(params.arguments["date"])
await params.result_callback({"slots": slots})
llm.register_function("check_availability", check_availability)
context = LLMContext(
messages=[{"role": "system", "content": "You are Acme Dental's receptionist."}],
tools=ToolsSchema(standard_tools=[
FunctionSchema(
name="check_availability",
description="Free appointment slots on a date",
properties={"date": {"type": "string"}},
required=["date"],
)
]),
)Inline sessions are client-steered: the agent speaks first because Pipecat requests a response when the context arrives, and stays quiet during silence unless you append a message or request another response.
Options
| Argument | Meaning |
|---|---|
api_key | Secret key (sk_live_...). Defaults to THUNDERPHONE_API_KEY. |
agent_id | Run a saved agent. Mutually exclusive with product and voice. |
product | 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. |
from_number, to_number | Numbers to record on the call when the pipeline fronts a phone line. |
live_transcripts | Stream caller transcript fragments mid-utterance (billed extra). |
greet_on_connect | Request a first response as soon as the session is ready. |
end_task_on_call_ended | Push EndWorkerFrame when ThunderPhone ends the call (default on). |
Events
ThunderPhone adds platform events on top of the realtime protocol. The service delivers them to handlers:
@llm.event_handler("on_call_ended")
async def on_call_ended(service, event):
print("call ended:", event["reason"], "call id:", service.call_id)service.call_id is set once the session is live. Use it with
GET /v1/calls/{call_id} to fetch the recording,
transcript and grade after the call.
Limits
- Turn detection is server-side and always on; Pipecat-driven turns
(
turn_detection=False) are not supported. - Functions registered on the service run for inline sessions only. A saved agent executes its own tools on ThunderPhone.
- Audio is 16-bit mono PCM at 24 kHz in both directions.
- Video frames are ignored.