Skip to content

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:

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:

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:

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:

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 wss:// URL of the WebSocket endpoint Twilio should stream call audio to (the <Stream url=...> attribute). Optional if options.websocket_url is set.

None
options VoiceTwiMLOptionsMediaStreams | dict[str, Any] | None

Optional VoiceTwiMLOptionsMediaStreams (or dict).

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 pydantic-field

websocket_url: str | None = None

Public WebSocket URL for Twilio's Media Stream (the attribute). Leave None to use the URL derived from TACConfig.voice_public_domain + voice_websocket_path.

custom_parameters pydantic-field

custom_parameters: dict[str, Any] | None = None

Custom parameters emitted as children of . They arrive back in the WebSocket 'start' event under start.customParameters.

name pydantic-field

name: str | None = None

Friendly name for the Stream (the attribute). Must be unique per call; arrives back in the WebSocket 'start' event.

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 verb completes (the attribute), with standard call parameters.

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])