ThunderPhone 2.0 is live.Self-serve, from 2¢/min.Read the announcement

Connect tools & data

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 key

The 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

ArgumentMeaning
api_keySecret key (sk_live_...). Defaults to THUNDERPHONE_API_KEY.
agent_idRun a saved agent. Mutually exclusive with product and voice.
productEngine for inline sessions: spark, bolt or storm.
voiceThunderPhone voice name for inline sessions.
languagePrimary language hint for inline sessions, e.g. es.
from_number, to_numberNumbers to record on the call when the pipeline fronts a phone line.
live_transcriptsStream caller transcript fragments mid-utterance (billed extra).
greet_on_connectRequest a first response as soon as the session is ready.
end_task_on_call_endedPush 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.