Media Streams Providers¶
VoiceChannel delegates the real-time media transport to a pluggable
VoiceProvider. The default one, ConversationRelayProvider, is documented
under Channels.
The providers below bridge Twilio
Media Streams
(<Connect><Stream>) to a speech-to-speech model API, relaying audio between
Twilio's WebSocket and the model's. Each needs the optional websockets
dependency:
pip install "tac[server,gpt-live]" # GPT-Live
pip install "tac[server,openai-realtime]" # Realtime
Twilio's bidirectional <Stream> only ever carries 8kHz G.711 u-law audio, so
each provider exports the constant to put in your session_config — model APIs
describe that same audio with different schemas, so the constants are not
interchangeable.
OpenAI GPT-Live¶
tac.channels.voice.media_streams.gpt_live ¶
GPTLiveProvider: bridges Twilio Media Streams to OpenAI's GPT-Live API.
TWILIO_AUDIO_FORMAT_FOR_GPT_LIVE
module-attribute
¶
TWILIO_AUDIO_FORMAT_FOR_GPT_LIVE: dict[str, Any] = {
"type": "audio/pcmu",
"rate": 8000,
}
GPT_LIVE_SESSION_ID_METADATA_KEY
module-attribute
¶
GPT_LIVE_SESSION_ID_METADATA_KEY = 'gpt_live_session_id'
GPTLiveProvider ¶
GPTLiveProvider(*args: Any, **kwargs: Any)
Bases: MediaStreamsOpenAIProvider[_CallState]
VoiceProvider bridging Twilio Media Streams to OpenAI's GPT-Live API.
Example
channel = VoiceChannel(tac, config=GPTLiveProviderConfig(default_session_config=...))
OpenAI's id for the GPT-Live session behind a call is exposed on the
session under GPT_LIVE_SESSION_ID_METADATA_KEY — quote it to OpenAI
support when reporting a session:
session.metadata[GPT_LIVE_SESSION_ID_METADATA_KEY] # e.g. "live_123"
initiate_outbound_conversation
async
¶
initiate_outbound_conversation(
options: InitiateVoiceConversationOptions,
) -> InitiateVoiceConversationResult
Initiate an outbound voice conversation.
Places an outbound call with inline TwiML that connects to a Media
Stream. Unlike inbound, there's no local session yet at this point —
it's created when Twilio's WebSocket start event arrives, the
same as _register_call does for inbound.
TwiML fields are merged per-field — see TwiMLBuilderMediaStreams.build.
The WebSocket URL is derived from TACConfig.voice_public_domain +
TACConfig.voice_websocket_path, unless overridden per-call via
options.websocket_url.
Pass InitiateVoiceConversationOptionsGPTLive with session_config
set to override the default for this call.
handle_websocket
async
¶
handle_websocket(websocket: WebSocketProtocol) -> None
Drive one Twilio Media Stream connection from accept to disconnect.
Races the Twilio read against the GPT-Live model-event reader so that if the model side disconnects first, we stop pumping caller audio into a dead socket and tear the call down immediately instead of leaving the caller connected to silence.
GPTLiveProviderConfig
pydantic-model
¶
Bases: MediaStreamsOpenAIProviderConfig
Configuration for GPTLiveProvider.
Fields:
-
memory_mode(MemoryMode) -
default_twiml_options(VoiceTwiMLOptionsMediaStreams | None) -
openai_api_key(str | None) -
tools(list[TACTool]) -
welcome_instruction(str | None) -
default_session_config(dict[str, Any] | None) -
on_inbound_call_session_config(Callable[[TwiMLRequest], Awaitable[dict[str, Any] | None]] | None)
Validators:
-
_validate_openai_api_key
tools
pydantic-field
¶
tools: list[TACTool]
Executable TACTool implementations, looked up by name to run Responses-delegated tool calls. This alone does not tell the model these tools exist — also add each tool's to_realtime_format() schema to default_session_config['delegation']['responses']['tools'].
welcome_instruction
pydantic-field
¶
welcome_instruction: str | None = None
If set, sent verbatim as a session.commentary.append once session.started arrives. Word it as an instruction, not just a greeting, e.g. 'Greet the caller immediately using: Hi, how can I help you today?' — a bare greeting won't make the model speak first.
default_session_config
pydantic-field
¶
default_session_config: dict[str, Any] | None = None
The session.start payload's 'session' body, sent once the model connects — used for any call that doesn't supply its own via on_inbound_call_session_config or InitiateVoiceConversationOptionsGPTLive. Must include model (e.g. 'gpt-live-1'). Set audio.format to TWILIO_AUDIO_FORMAT_FOR_GPT_LIVE and, for tool calling, delegation = {'type': 'responses', 'responses': {'model': ..., 'tools': [...]}}.
on_inbound_call_session_config
pydantic-field
¶
on_inbound_call_session_config: (
Callable[
[TwiMLRequest], Awaitable[dict[str, Any] | None]
]
| None
) = None
Per-inbound-call override for default_session_config, called with the TwiMLRequest. Its return value is used verbatim (not merged with default_session_config); return None to fall back to it.
InitiateVoiceConversationOptionsGPTLive
pydantic-model
¶
Bases: InitiateVoiceConversationOptions
Outbound options for GPTLiveProvider, adding a per-call session_config.
Fields:
-
to(str) -
websocket_url(str | None) -
twiml_options(VoiceTwiMLOptions | None) -
call_options(CallOptions | None) -
session_config(dict[str, Any] | None)
session_config
pydantic-field
¶
session_config: dict[str, Any] | None = None
Used verbatim in place of GPTLiveProviderConfig.default_session_config for this call.
OpenAI Realtime¶
tac.channels.voice.media_streams.openai_realtime ¶
OpenAIRealtimeProvider: bridges Twilio Media Streams to OpenAI's Realtime API.
TWILIO_AUDIO_FORMAT_FOR_REALTIME
module-attribute
¶
TWILIO_AUDIO_FORMAT_FOR_REALTIME: dict[str, Any] = {
"type": "audio/pcmu"
}
OpenAIRealtimeProvider ¶
OpenAIRealtimeProvider(
channel: VoiceChannel,
tac_config: TACConfig,
config: MediaStreamsOpenAIProviderConfig,
)
Bases: MediaStreamsOpenAIProvider[_CallState]
VoiceProvider bridging Twilio Media Streams to OpenAI Realtime.
Example
channel = VoiceChannel(tac, config=OpenAIRealtimeProviderConfig(default_session_config=...))
initiate_outbound_conversation
async
¶
initiate_outbound_conversation(
options: InitiateVoiceConversationOptions,
) -> InitiateVoiceConversationResult
Initiate an outbound voice conversation.
Places an outbound call with inline TwiML that connects to a Media
Stream. Unlike inbound, there's no local session yet at this point —
it's created when Twilio's WebSocket start event arrives, the
same as _register_call does for inbound.
TwiML fields are merged per-field — see TwiMLBuilderMediaStreams.build.
The WebSocket URL is derived from TACConfig.voice_public_domain +
TACConfig.voice_websocket_path, unless overridden per-call via
options.websocket_url.
Pass InitiateVoiceConversationOptionsOpenAIRealtime with
session_config set to override the default for this call.
handle_websocket
async
¶
handle_websocket(websocket: WebSocketProtocol) -> None
Drive one Twilio Media Stream connection from accept to disconnect.
Races the Twilio read against the OpenAI model-event reader so that if the model side disconnects first, we stop pumping caller audio into a dead socket and tear the call down immediately instead of leaving the caller connected to silence.
OpenAIRealtimeProviderConfig
pydantic-model
¶
Bases: MediaStreamsOpenAIProviderConfig
Configuration for OpenAIRealtimeProvider.
Fields:
-
memory_mode(MemoryMode) -
default_twiml_options(VoiceTwiMLOptionsMediaStreams | None) -
openai_api_key(str | None) -
tools(list[TACTool]) -
welcome_greeting_response(dict[str, Any] | None) -
default_session_config(dict[str, Any] | None) -
on_inbound_call_session_config(Callable[[TwiMLRequest], Awaitable[dict[str, Any] | None]] | None)
Validators:
-
_validate_openai_api_key
tools
pydantic-field
¶
tools: list[TACTool]
Executable TACTool implementations, looked up by name to run mid-call tool requests. This alone does not tell the model these tools exist — also add each tool's to_realtime_format() schema to default_session_config['tools'].
welcome_greeting_response
pydantic-field
¶
welcome_greeting_response: dict[str, Any] | None = None
If set, sent verbatim as response.create's 'response' payload when the call connects — e.g. {'instructions': 'Hi there!'}. No SDK-added wrapping text or language assumption.
default_session_config
pydantic-field
¶
default_session_config: dict[str, Any] | None = None
The session.update payload's 'session' body, sent once the model connects — used for any call that doesn't supply its own via on_inbound_call_session_config or InitiateVoiceConversationOptionsOpenAIRealtime. See https://developers.openai.com/api/reference/resources/realtime/client-events#session.update for the schema. If using tools, its 'tools' entry must separately list each tool's to_realtime_format() schema — this config is passed to OpenAI as-is, with no tool schemas merged in.
on_inbound_call_session_config
pydantic-field
¶
on_inbound_call_session_config: (
Callable[
[TwiMLRequest], Awaitable[dict[str, Any] | None]
]
| None
) = None
Per-inbound-call override for default_session_config, called with the TwiMLRequest. Its return value is used verbatim (not merged with default_session_config); return None to fall back to it. Outbound calls don't use this — see InitiateVoiceConversationOptionsOpenAIRealtime.
InitiateVoiceConversationOptionsOpenAIRealtime
pydantic-model
¶
Bases: InitiateVoiceConversationOptions
Outbound options for OpenAIRealtimeProvider, adding a per-call session_config.
Fields:
-
to(str) -
websocket_url(str | None) -
twiml_options(VoiceTwiMLOptions | None) -
call_options(CallOptions | None) -
session_config(dict[str, Any] | None)
session_config
pydantic-field
¶
session_config: dict[str, Any] | None = None
Used verbatim in place of OpenAIRealtimeProviderConfig.default_session_config for this call.
Shared¶
Used by every Media Streams provider.
tac.channels.voice.media_streams.twiml ¶
TwiML generation for Media Streams (<Connect><Stream>) providers.
See https://www.twilio.com/docs/voice/twiml/stream for the <Stream> verb.
generate_twiml ¶
generate_twiml(
websocket_url: str | None = None,
options: VoiceTwiMLOptionsMediaStreams
| dict[str, Any]
| None = None,
) -> str
Generate TwiML that connects the call to a bidirectional Media Stream.
The WebSocket URL may be passed positionally or as options.websocket_url
(positional wins when both are given), so a caller can pass everything in
one object: generate_twiml(options=VoiceTwiMLOptionsMediaStreams(websocket_url=...)).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_url
|
str | None
|
Public |
None
|
options
|
VoiceTwiMLOptionsMediaStreams | dict[str, Any] | None
|
Optional |
None
|
Returns:
| Type | Description |
|---|---|
str
|
TwiML XML string ready to return to Twilio. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no WebSocket URL is provided via either source. |
tac.models.voice ¶
Pydantic models for Twilio ConversationRelay Voice WebSocket messages.
VoiceTwiMLOptionsMediaStreams
pydantic-model
¶
Bases: VoiceTwiMLOptions
Options for the TwiML inside <Connect><Stream>.
Fields map to the attributes documented at
https://www.twilio.com/docs/voice/twiml/stream (the <Stream> verb)
and https://www.twilio.com/docs/voice/twiml/connect (the <Connect>
verb it's nested in). track is omitted — bidirectional <Connect>
streams only ever carry inbound_track, so it isn't settable.
Fields:
-
websocket_url(str | None) -
custom_parameters(dict[str, Any] | None) -
name(str | None) -
status_callback(str | None) -
status_callback_method(Literal['GET', 'POST'] | None) -
action_url(str | None) -
action_method(Literal['GET', 'POST'] | None)
websocket_url
pydantic-field
¶
websocket_url: str | None = None
Public WebSocket URL for Twilio's Media Stream (the
custom_parameters
pydantic-field
¶
custom_parameters: dict[str, Any] | None = None
Custom parameters emitted as
name
pydantic-field
¶
name: str | None = None
Friendly name for the Stream (the
status_callback
pydantic-field
¶
status_callback: str | None = None
Absolute URL Twilio posts to when the stream starts, stops, or errors (StreamSid/StreamName/StreamEvent/StreamError/Timestamp params).
status_callback_method
pydantic-field
¶
status_callback_method: Literal["GET", "POST"] | None = None
HTTP method for status_callback. Defaults to POST on Twilio.
action_url
pydantic-field
¶
action_url: str | None = None
URL Twilio requests when the
action_method
pydantic-field
¶
action_method: Literal['GET', 'POST'] | None = None
HTTP method for action_url. Defaults to POST on Twilio.
tac.models.stream ¶
Pydantic models for Twilio Media Streams (<Connect><Stream>) WebSocket messages.
StreamStartMessage
pydantic-model
¶
Bases: BaseModel
The start field of Twilio's Media Stream start event.
Config:
default:{'populate_by_name': True}
Fields:
-
stream_sid(str) -
call_sid(str) -
media_format(dict[str, Any] | None) -
custom_parameters(dict[str, str])