Module scenario.voice.adapters.twilio
TwilioAgentAdapter: bidirectional real phone transport via Twilio Media Streams.
One adapter class serves both directions a Twilio number can participate in:
- Inbound —
wait_for_call()sets the number's voice webhook to our local server, blocks until a caller dials in, and opens a Media Streams WebSocket for the call. - Outbound —
place_call(to=...)originates a call via Twilio REST, then accepts the Media Streams WebSocket Twilio opens back to us.
connect() is direction-agnostic: resolve the number SID, start a
FastAPI webhook + Media Streams WS server, open a public URL (user-supplied
or via CloudflareTunnel). After connect(), call either
place_call() or wait_for_call().
Expand source code
"""
TwilioAgentAdapter: bidirectional real phone transport via Twilio Media Streams.
One adapter class serves both directions a Twilio number can participate in:
- **Inbound** — ``wait_for_call()`` sets the number's voice webhook to our
local server, blocks until a caller dials in, and opens a Media Streams
WebSocket for the call.
- **Outbound** — ``place_call(to=...)`` originates a call via Twilio REST,
then accepts the Media Streams WebSocket Twilio opens back to us.
``connect()`` is direction-agnostic: resolve the number SID, start a
FastAPI webhook + Media Streams WS server, open a public URL (user-supplied
or via ``CloudflareTunnel``). After ``connect()``, call either
``place_call()`` or ``wait_for_call()``.
"""
from __future__ import annotations
import asyncio
import logging
import time
from contextlib import suppress
from typing import Any, Callable, ClassVar, Literal, Optional
# The adapter is dormant after connect() and before a direction-specific kickoff
# method is called. Mode is dynamic — we don't take it at construction — because
# the same credentials + tunnel can serve either direction, and picking at
# first-use keeps the API small.
TwilioAdapterMode = Literal["idle", "answer", "call"]
#: How the media-stream session most recently ended. "none" until a session has
#: ever run (the disconnect-counters T3 enum, #775). Set by the media loop
#: (`_twilio_server.py`) at each of its three termination paths.
TwilioStreamEndedReason = Literal["stop", "close", "error", "none"]
from opentelemetry import trace as _otel_trace
from ...types import AgentRole
from ..adapter import VoiceAgentAdapter
from ..audio_chunk import AudioChunk
from ..capabilities import AdapterCapabilities
from .._telemetry import set_span_attributes, voice_span
from ._twilio_shared import (
TWILIO_FRAME_MS,
TwilioRESTHelper,
_redact_e164,
build_clear_frame,
build_media_frame,
iter_mulaw_frames,
pcm16_24k_to_mulaw8k,
validate_e164,
)
logger = logging.getLogger("scenario.voice.twilio")
#: A-leg <Say> anchor for outbound calls. Whisper hallucinates non-English
#: text on bare ``<Pause>`` silence (#465 in this PR), so we play one known-
#: good utterance at call setup, then hold the line for the Media Stream to
#: carry the real bidirectional conversation.
PLACE_CALL_A_LEG_SAY_TEXT = (
"Thank you for calling. "
"I will hold the line while you complete your scenario."
)
class TwilioAgentAdapter(VoiceAgentAdapter):
"""
Bidirectional Twilio Media Streams adapter.
Same class, same state, either direction:
adapter = TwilioAgentAdapter(
account_sid=..., auth_token=...,
phone_number="+14155551234",
)
async with adapter: # connect() / disconnect()
await adapter.place_call(to="+14155557777") # OR wait_for_call()
# ... scenario.run(...) feeds send_audio / recv_audio ...
The adapter is the *only* adapter with ``dtmf=True``. DTMF events
received from the callee surface via the ``on_dtmf`` callback set at
construction time. To send DTMF, use ``send_dtmf()``.
``interrupt(after_words=N)`` raises ``UnsupportedCapabilityError`` on this
adapter — Media Streams delivers raw audio without incremental
transcripts. Use ``interrupt(after=seconds)`` instead.
"""
capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
streaming_transcripts=False,
native_vad=False,
dtmf=True,
# Twilio Media Streams ``clear`` event drops all buffered outbound
# audio. Used by ``adapter.interrupt()`` already wired below.
interruption=True,
input_formats=["mulaw/8000"],
output_formats=["mulaw/8000"],
)
# ------------------------------------------------------------------ init
def __init__(
self,
*,
account_sid: str,
auth_token: str,
phone_number: str,
public_base_url: Optional[str] = None,
allowed_callers: Optional[list[str]] = None,
on_dtmf: Optional[Callable[[str], None]] = None,
http_port: int = 8765,
role: AgentRole = AgentRole.AGENT,
validate_signature: bool = True,
) -> None:
super().__init__()
validate_e164(phone_number)
self.account_sid = account_sid
self.auth_token = auth_token
self.phone_number = phone_number
self.public_base_url = public_base_url
self.allowed_callers = set(allowed_callers) if allowed_callers else None
self.on_dtmf = on_dtmf
self.http_port = http_port
self.role = role # type: ignore[misc]
# When True (default), the inbound /twilio/voice route requires a
# valid X-Twilio-Signature header before accepting the webhook
# body. The cloudflared tunnel URL is ephemeral and not
# guessable, but anyone who learns it could otherwise POST fake
# Twilio webhook events into the harness. Tests that don't have
# real Twilio credentials disable this with
# ``validate_signature=False``; production callers should leave
# it on.
self.validate_signature = validate_signature
if not validate_signature:
logger.warning(
"TwilioAgentAdapter: validate_signature=False — inbound "
"webhooks accept any payload without signature checks. "
"Use only in tests; do not deploy to production."
)
# Populated during connect(); None when disconnected.
self._rest: Optional[TwilioRESTHelper] = None
self._phone_number_sid: Optional[str] = None
self._prior_voice_url: Optional[str] = None
# Set by place_call() when it rewrites the callee's voice_url so
# B-leg's webhook lands on our harness. Restored in disconnect.
self._callee_phone_number_sid: Optional[str] = None
self._prior_callee_voice_url: Optional[str] = None
# Set by the first of wait_for_call()/place_call(); subsequent calls to
# the other method raise. "idle" after connect() before either fires.
self._mode: TwilioAdapterMode = "idle"
# Server / media stream state.
self._server_task: Optional[asyncio.Task] = None
self._server_shutdown: Optional[asyncio.Event] = None
self._call_sid: Optional[str] = None
self._stream_sid: Optional[str] = None
self._stream_connected: Optional[asyncio.Event] = None
self._stream_ws: Any = None # starlette WebSocket
self._inbound_queue: Optional[asyncio.Queue[AudioChunk]] = None
# Set True by the media-stream loop's terminal path (stop / socket close
# / throw) the moment it enqueues the end-of-call sentinel. Once the call
# has ended, recv_audio must keep draining the queue (and hand back the
# sentinel) WITHOUT re-asserting transport liveness — the real
# production wrapper nulls _stream_ws/_stream_sid synchronously right
# after the loop returns, so the drain's second recv_audio would
# otherwise hit _assert_stream_live and crash (#695). Reset on
# connect(), disconnect(), and at media-stream-loop entry (per-call
# scope — a second session on the same connected adapter must not
# inherit the previous session's flag).
self._stream_ended: bool = False
# Call-lifetime counters stamped onto ``voice.adapter.disconnect`` from
# inside disconnect() (#775 Tier 2b — mirrors ElevenLabs' pump-counter
# seam). Accumulated by the media loop (`_twilio_server.py`) and the
# `/twilio/voice` webhook handler; reset on connect()/disconnect() like
# `_stream_ended` above.
self._frames_received: int = 0
self._dtmf_received: int = 0
self._stream_ended_reason: TwilioStreamEndedReason = "none"
self._webhook_invocations: int = 0
self._webhook_rejected: int = 0
# ------------------------------------------------------------------ repr
def __repr__(self) -> str: # redact credentials
return (
f"TwilioAgentAdapter("
f"phone_number={self.phone_number!r}, "
f"account_sid='***', auth_token='***', "
f"public_base_url={self.public_base_url!r})"
)
# ------------------------------------------------------------------ lifecycle
async def connect(self) -> None:
"""Resolve number SID and start the FastAPI webhook + WS server.
Does NOT modify the Twilio account's ``voice_url``. That side-effect
only happens when ``wait_for_call()`` is invoked — callers (who will
use ``place_call()``) never overwrite their number's inbound webhook,
which makes caller-mode adapters safe to run against a shared pool of
Twilio numbers without clobbering anyone's prod webhook.
Idempotent: calling connect() on an already-connected adapter is a
no-op. This lets the scenario executor's auto-connect step
(``_voice_connect_all``) coexist with explicit harness-driven
connects (``TwilioHarness`` ``__aenter__``) that already brought
the adapter up before scenario.run() was called.
"""
if self._rest is not None:
return
if self.public_base_url is None:
raise RuntimeError(
"TwilioAgentAdapter: public_base_url is required. Wrap the "
"adapter in scenario.voice.testing.TwilioHarness, or supply "
"a stable public HTTPS URL that routes to this machine."
)
self._rest = TwilioRESTHelper(self.account_sid, self.auth_token)
self._phone_number_sid = self._rest.resolve_phone_number_sid(self.phone_number)
# Stamp Twilio-specific attrs onto the active ``voice.adapter.connect``
# span (opened by the executor connect loop). Base spans are name-owned;
# the adapter contributes attributes, never a parallel span name — mirror
# ElevenLabs' ``voice.elevenlabs.agent_id`` seam (``adapters/elevenlabs.py``).
# NOTE: no ``voice.twilio.direction`` here — connect() is direction-
# agnostic (``_mode`` stays "idle" until place_call()/wait_for_call(),
# after this span has already closed); direction is stamped on the NEW
# ``voice.adapter.dial`` span instead (see place_call/wait_for_call).
set_span_attributes(
_otel_trace.get_current_span(),
{
"voice.twilio.phone_number_sid": self._phone_number_sid,
"voice.twilio.validate_signature": self.validate_signature,
"voice.twilio.webhook_port": self.http_port,
},
)
self._stream_connected = asyncio.Event()
self._inbound_queue = asyncio.Queue()
self._server_shutdown = asyncio.Event()
self._stream_ended = False
self._frames_received = 0
self._dtmf_received = 0
self._stream_ended_reason = "none"
self._webhook_invocations = 0
self._webhook_rejected = 0
self._mode = "idle"
# Webhook server is its own unit — see _twilio_server.py. The
# adapter only orchestrates lifecycle; the routes, signature
# validation, and WS framing live in TwilioWebhookServer.
from ._twilio_server import TwilioWebhookServer
self._webhook_server: Optional[TwilioWebhookServer] = TwilioWebhookServer(self)
self._server_task = asyncio.create_task(self._run_server())
# Give uvicorn a beat to bind the port before Twilio hits it.
await asyncio.sleep(0.2)
async def disconnect(self) -> None:
"""Restore prior voice_url (answer mode only), tear down server.
Best-effort on errors. In caller mode we never touched the Twilio
number's voice_url, so there's nothing to restore.
"""
if self._rest is None:
return
# 1. Restore webhook first so Twilio doesn't keep hitting a dead URL.
# ``rest_restore_failed`` tracks whether either restore below raised —
# previously fully swallowed via ``suppress(Exception)``; now surfaced
# onto the disconnect span (#775 Tier 2b) so a Twilio REST outage during
# teardown is no longer invisible. The swallow behavior itself (a failed
# restore must never block disconnect()) is unchanged.
rest_restore_failed = False
if self._mode == "answer" and self._phone_number_sid is not None:
try:
prior = self._prior_voice_url or ""
self._rest.write_voice_url(self._phone_number_sid, prior)
logger.debug(
"TwilioAgentAdapter: restored voice_url=%r on %s",
prior,
self._phone_number_sid,
)
except Exception:
rest_restore_failed = True
# place_call() rewrites the CALLEE's voice_url to attach Media
# Streams to B-leg. Restore that too.
if self._mode == "call" and self._callee_phone_number_sid is not None:
try:
prior_b = self._prior_callee_voice_url or ""
self._rest.write_voice_url(self._callee_phone_number_sid, prior_b)
logger.debug(
"TwilioAgentAdapter: restored callee voice_url=%r on %s",
prior_b,
self._callee_phone_number_sid,
)
except Exception:
rest_restore_failed = True
# 2. Signal server to shut down, then wait for the task. This is what
# ACTUALLY closes a still-live media stream (uvicorn's graceful
# shutdown tears down the open WS), which is what lets the media
# loop's ``finally``/except-handlers (``_twilio_server.py``) make
# their FINAL writes to ``_frames_received`` / ``_stream_ended_reason``
# / etc. The counter stamp below MUST run after this await, not
# before — stamping first (the #775 review-caught bug) reads the
# pre-teardown snapshot and reports ``stream_ended_reason=="none"`` /
# undercounted frames for a call that was still live when disconnect()
# was invoked, even though the call did in fact end (via this very
# shutdown) moments later.
if self._server_shutdown is not None:
self._server_shutdown.set()
if self._server_task is not None:
with suppress(Exception):
await asyncio.wait_for(self._server_task, timeout=3.0)
# Stamp call-lifetime counters onto the active ``voice.adapter.disconnect``
# span (opened by the executor disconnect loop) — mirrors ElevenLabs'
# pump-counter stamp (``adapters/elevenlabs.py``). Sampled AFTER the
# shutdown-await above (not before) so a still-live call's FINAL
# counts/reason are captured, not a mid-call snapshot. The disconnect
# span stays the current span across the await (contextvars survive
# awaits within the same task), so ``get_current_span()`` here still
# targets it.
set_span_attributes(
_otel_trace.get_current_span(),
{
"voice.twilio.frames_received": self._frames_received,
"voice.twilio.dtmf_received": self._dtmf_received,
"voice.twilio.stream_ended_reason": self._stream_ended_reason,
"voice.twilio.webhook_invocations": self._webhook_invocations,
"voice.twilio.webhook_rejected": self._webhook_rejected,
"voice.twilio.rest_restore_failed": rest_restore_failed,
},
)
# 3. Reset state.
self._rest = None
self._phone_number_sid = None
self._prior_voice_url = None
self._callee_phone_number_sid = None
self._prior_callee_voice_url = None
self._mode = "idle"
self._server_task = None
self._server_shutdown = None
self._call_sid = None
self._stream_sid = None
self._stream_connected = None
self._stream_ws = None
if self._inbound_queue is not None:
# A recv_audio() task may already be blocked in get() on this queue.
# Wake that in-flight consumer before dropping our reference; setting
# _inbound_queue to None alone leaves it parked until its receive
# timeout expires (#757). An empty chunk is the adapter's existing
# end-of-stream signal and lets the shared drain finish cleanly.
self._inbound_queue.put_nowait(AudioChunk(data=b""))
self._inbound_queue = None
self._stream_ended = False
self._frames_received = 0
self._dtmf_received = 0
self._stream_ended_reason = "none"
self._webhook_invocations = 0
self._webhook_rejected = 0
# ------------------------------------------------------------------ direction
async def place_call(
self,
to: str,
*,
timeout: float = 120.0,
attach_stream_to_self: bool = True,
) -> None:
"""
Originate an outbound call from this adapter's Twilio number to ``to``.
Twilio's REST ``Calls.create`` runs TwiML on TWO legs of the
resulting call:
- **A-leg** (the originator, ``from_=self.phone_number``): runs the
inline TwiML passed via ``twiml=`` to ``Calls.create``. We use
``<Pause length=120>`` so the originator just holds the bridge
open while the demo runs.
- **B-leg** (the callee, ``to=``): when Twilio dials B and B picks
up, B's number's ``voice_url`` fires — that's where the bridge's
Media Streams attach. This is identical to the inbound demo's
flow: B's voice_url returns ``<Connect><Stream>``, the WS opens,
audio flows.
So ``place_call`` only makes sense when ``to`` is another Twilio
number on this account whose ``voice_url`` is set to OUR harness
webhook. To make that wiring automatic, ``place_call`` temporarily
rewrites B's ``voice_url`` for the duration of the call and
restores it on ``disconnect``. The harness on this adapter's own
number does NOT need to be answer-mode — the Stream attaches to
B's leg, NOT this adapter's leg, but B's webhook is hosted on this
adapter's local server, so the WS still lands here.
The bidirectional audio model is unchanged: ``send_audio`` writes
frames over the WS (B hears them and bridges to A), ``recv_audio``
reads inbound frames off the WS (whatever the bridge mixes from
both legs).
Limitation: ``to`` MUST be a phone number on this same Twilio
account. Calling an external PSTN endpoint (a real cell phone)
requires a different topology (``<Start><Stream>`` + ``<Dial>``)
which we don't implement here because the inline TwiML route on
the A-leg can't capture B's audio when B is external.
Default timeout 120s covers cloudflared cold-start latency.
Raises:
RuntimeError: If called after ``wait_for_call()`` (modes are
exclusive per adapter instance), or if ``to`` is not a
Twilio number on this account.
ValueError: If ``to`` is not in E.164 format.
asyncio.TimeoutError: If the media stream doesn't open within
``timeout`` seconds.
"""
self._assert_connected()
self._enter_mode("call")
validate_e164(to)
assert self.public_base_url is not None
assert self._rest is not None
assert self._stream_connected is not None
# NEW ``voice.adapter.dial`` span (#775 Tier 2a): self-instrumented,
# since no executor span is active by the time place_call()/
# wait_for_call() run (the ``voice.adapter.connect`` span already
# closed). Wraps the REST dial + the ``_stream_connected`` wait — the
# "I placed the call but no media ever streamed" failure surface.
with voice_span(
"voice.adapter.dial",
{
"voice.adapter.class": type(self).__name__,
"voice.twilio.direction": "outbound",
"voice.twilio.to": _redact_e164(to),
"voice.twilio.from": _redact_e164(self.phone_number),
},
) as _dial:
if attach_stream_to_self:
# Resolve B-leg's number SID and snapshot+rewrite its voice_url so
# B's leg attaches its Media Stream to our harness webhook. We own
# this number (same Twilio account); disconnect() will restore.
self._callee_phone_number_sid = self._rest.resolve_phone_number_sid(to)
self._prior_callee_voice_url = self._rest.read_voice_url(
self._callee_phone_number_sid
)
webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice"
self._rest.write_voice_url(self._callee_phone_number_sid, webhook_url)
logger.info(
"TwilioAgentAdapter: rewrote callee %s voice_url to %s",
_redact_e164(to),
webhook_url,
)
# A-leg TwiML: play a short deterministic <Say> line, then hold
# the bridge open. Twilio runs this on the originator side while
# B's webhook attaches the Media Stream.
#
# The <Say> gives the recording a known-good utterance to
# transcribe. A bare <Pause> alone produces 120s of line silence
# that Whisper has been observed to hallucinate as non-English
# text (issue #465 in this PR). The Say is a one-time anchor at
# call setup; the Media Stream carries the real bidirectional
# conversation that follows.
inline_a_leg_twiml = (
'<?xml version="1.0" encoding="UTF-8"?>'
"<Response>"
f'<Say voice="Polly.Joanna">{PLACE_CALL_A_LEG_SAY_TEXT}</Say>'
'<Pause length="120"/>'
"</Response>"
)
self._call_sid = self._rest.place_call(
to=to, from_=self.phone_number, twiml=inline_a_leg_twiml
)
logger.info(
"TwilioAgentAdapter: placed call %s from %s to %s",
self._call_sid,
_redact_e164(self.phone_number),
_redact_e164(to),
)
set_span_attributes(_dial, {"voice.twilio.call_sid": self._call_sid})
if attach_stream_to_self:
# Wait for OUR webhook to fire — only meaningful when we rewrote
# the callee's voice_url to point at us. In originator-only mode
# (attach_stream_to_self=False), there's no stream coming to us;
# the callee has its own harness which owns the stream.
_dial_wait_started = time.monotonic()
try:
await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout)
except asyncio.TimeoutError:
# The media stream never connected — the marquee "I placed
# the call but no media ever streamed" failure. Tag the
# outcome BEFORE re-raising; the raised TimeoutError still
# marks the span ERROR (voice_span's own exception handling).
_dial.set_attribute(
"voice.twilio.dial_outcome", "stream_connect_timeout"
)
raise
set_span_attributes(
_dial,
{
"voice.twilio.stream_connect_latency_ms": round(
(time.monotonic() - _dial_wait_started) * 1000
),
},
)
async def wait_for_call(self, timeout: float = 120.0) -> None:
"""
Block until someone dials in and the media stream is live.
In **default mode** (no conference_room): the number's ``voice_url``
is overwritten to point at our webhook so inbound calls reach us.
Caller (``place_call``) elsewhere will dial this number.
In **conference mode**: there's no inbound call to wait for — the
two-Twilio-number demo can't naturally have one side receive an
inbound call when both legs need conference TwiML. Instead, we
ORIGINATE an outbound call FROM this adapter's number TO itself
with inline TwiML that opens the capture stream and dials into
the shared conference room. This is the symmetric counterpart to
``place_call`` in conference mode — both adapters end up as
participants in the same room, exchanging audio via the bridge.
Default timeout 120s covers cloudflared cold-start, conference-room
formation, and Twilio's webhook ramp.
Raises:
RuntimeError: If called after ``place_call()``.
asyncio.TimeoutError: If nobody dials in within ``timeout``.
"""
self._assert_connected()
self._enter_mode("answer")
assert self.public_base_url is not None
assert self._rest is not None
assert self._phone_number_sid is not None
assert self._stream_connected is not None
# NEW ``voice.adapter.dial`` span (#775 Tier 2a) — see place_call()'s
# comment for the rationale (self-instrumented, no executor span active
# here).
with voice_span(
"voice.adapter.dial",
{
"voice.adapter.class": type(self).__name__,
"voice.twilio.direction": "inbound",
"voice.twilio.to": _redact_e164(self.phone_number),
},
) as _dial:
# Snapshot the prior webhook so we can restore it on disconnect, then
# point the number at our server. Only answer mode does this.
self._prior_voice_url = self._rest.read_voice_url(self._phone_number_sid)
webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice"
self._rest.write_voice_url(self._phone_number_sid, webhook_url)
logger.info("TwilioAgentAdapter: webhook set to %s", webhook_url)
_dial_wait_started = time.monotonic()
try:
await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout)
except asyncio.TimeoutError:
# Nobody dialed in — tag the outcome BEFORE re-raising; the
# raised TimeoutError still marks the span ERROR (voice_span's
# own exception handling).
_dial.set_attribute(
"voice.twilio.dial_outcome", "stream_connect_timeout"
)
raise
set_span_attributes(
_dial,
{
"voice.twilio.call_sid": self._call_sid,
"voice.twilio.stream_connect_latency_ms": round(
(time.monotonic() - _dial_wait_started) * 1000
),
},
)
def _enter_mode(self, mode: TwilioAdapterMode) -> None:
"""Transition idle → mode, or raise if already in a different mode.
Modes are exclusive per connected session: an adapter can place a call
or answer a call, not both. Disconnect + reconnect to reuse the
instance in the other direction.
"""
if self._mode == mode:
return # idempotent re-entry (e.g. retrying place_call after timeout)
if self._mode != "idle":
raise RuntimeError(
f"TwilioAgentAdapter: already in {self._mode!r} mode; cannot "
f"switch to {mode!r}. Disconnect and reconnect to reuse this "
f"adapter in the other direction."
)
self._mode = mode
# ------------------------------------------------------------------ I/O
async def send_audio(self, chunk: AudioChunk) -> None:
# Pace at real-time (one frame per TWILIO_FRAME_MS). Without pacing the
# whole utterance arrives in milliseconds, which trips bots' VAD into
# a clipped-utterance reading.
self._assert_stream_live()
ws = self._stream_ws
stream_sid = self._stream_sid
assert ws is not None and stream_sid is not None
mulaw = pcm16_24k_to_mulaw8k(chunk.data)
frame_secs = TWILIO_FRAME_MS / 1000
for frame in iter_mulaw_frames(mulaw):
if not frame:
continue
await ws.send_text(build_media_frame(stream_sid, frame))
await asyncio.sleep(frame_secs)
async def recv_audio(self, timeout: float) -> AudioChunk:
# Once the media-stream loop has ended (stop / socket close / throw),
# the production wrapper (run_stream_session in _twilio_server) nulls
# _stream_ws/_stream_sid synchronously right after the loop returns. The
# drain loop (_drain_agent_response) always makes a *second* recv_audio
# call after the first chunk — by then liveness has flipped. So the
# liveness assert only guards the genuinely-live-and-idle case; buffered
# audio and the end-of-call drain bypass it (#695).
#
# Two sentinel sources cooperate here, and BOTH are load-bearing: the
# loop's ``finally`` ENQUEUES one empty chunk — that is what wakes a
# consumer already blocked in ``queue.get()`` at the moment the call
# ends (flipping ``_stream_ended`` alone wakes nobody) — and this method
# SYNTHESIZES further empty chunks for every call after the queue has
# drained (the tail-silence probe, or any caller that keeps polling).
# Delete either and a hang comes back.
if self._inbound_queue is None:
# disconnect() tore the adapter down mid-drain — nothing left to
# read. In the current teardown order _assert_connected() raises
# first (_rest is nulled before _inbound_queue, synchronously), so
# the explicit raise below is a reordering guard, not a live path —
# it keeps this method's error honest if disconnect() is ever
# resequenced.
self._assert_connected()
raise RuntimeError(
"TwilioAgentAdapter: inbound queue is gone; adapter disconnected."
)
if not self._inbound_queue.empty():
# Buffered audio or the loop-enqueued terminal sentinel — drain it,
# live or not.
return await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout)
if self._stream_ended:
# Call ended and fully drained: synthesize another empty sentinel.
return AudioChunk(data=b"")
# Live call, nothing buffered yet: assert liveness, then wait.
self._assert_stream_live()
return await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout)
async def send_dtmf(self, tones: str) -> None:
"""Send DTMF digits on the live call (uses Twilio REST ``<Play digits>``)."""
if self._rest is None or self._call_sid is None:
raise RuntimeError("TwilioAgentAdapter: no active call; send_dtmf requires an in-progress call")
# Run blocking REST call off-thread so we don't stall the event loop.
await asyncio.to_thread(self._rest.send_dtmf_on_call, self._call_sid, tones)
async def interrupt(self) -> None:
"""Drop any buffered outbound audio on Twilio's side (``clear`` event)."""
self._assert_stream_live()
ws = self._stream_ws
stream_sid = self._stream_sid
assert ws is not None and stream_sid is not None
await ws.send_text(build_clear_frame(stream_sid))
# ------------------------------------------------------------------ server
async def _run_server(self) -> None:
"""Thin lifecycle wrapper; the real work lives in TwilioWebhookServer."""
assert self._webhook_server is not None
await self._webhook_server.run()
def _build_app(self) -> Any:
"""Test seam: build the FastAPI app for in-process exercise.
Production code never calls this — the server's ``run()`` builds
the app itself when uvicorn starts. Existing unit tests use
``TestClient(_build_app())`` to exercise the routes without
binding a port; the delegation keeps that test surface stable.
"""
assert self._webhook_server is not None
return self._webhook_server.build_app()
async def _media_stream_loop(self, ws: Any) -> None:
"""Test seam: kick off the Media Streams WS loop directly.
The two-adapter-bridge test in
``tests/voice/test_twilio_two_adapter_bridge.py`` uses this to
drive a loopback WS without going through the FastAPI route
wrapper. Production code reaches the loop via the ``/twilio/stream``
WebSocket handler defined in ``_twilio_server.build_app``.
"""
assert self._webhook_server is not None
await self._webhook_server.media_stream_loop(ws)
# ------------------------------------------------------------------ assertions
def _assert_connected(self) -> None:
if self._rest is None:
raise RuntimeError("TwilioAgentAdapter: not connected; call connect() or use `async with`.")
def _assert_stream_live(self) -> None:
self._assert_connected()
if self._stream_ws is None or self._stream_sid is None:
raise RuntimeError(
"TwilioAgentAdapter: no live media stream. Call place_call() or "
"wait_for_call() first."
)Global variables
var PLACE_CALL_A_LEG_SAY_TEXT-
A-leg
anchor for outbound calls. Whisper hallucinates non-English text on bare <Pause>silence (#465 in this PR), so we play one known- good utterance at call setup, then hold the line for the Media Stream to carry the real bidirectional conversation. var TwilioStreamEndedReason-
How the media-stream session most recently ended. "none" until a session has ever run (the disconnect-counters T3 enum, #775). Set by the media loop (
_twilio_server.py) at each of its three termination paths.
Classes
class TwilioAgentAdapter (*, account_sid: str, auth_token: str, phone_number: str, public_base_url: Optional[str] = None, allowed_callers: Optional[list[str]] = None, on_dtmf: Optional[Callable[[str], None]] = None, http_port: int = 8765, role: AgentRole = AgentRole.AGENT, validate_signature: bool = True)-
Bidirectional Twilio Media Streams adapter.
Same class, same state, either direction:
adapter = TwilioAgentAdapter( account_sid=..., auth_token=..., phone_number="+14155551234", ) async with adapter: # connect() / disconnect() await adapter.place_call(to="+14155557777") # OR wait_for_call() # ... scenario.run(...) feeds send_audio / recv_audio ...The adapter is the only adapter with
dtmf=True. DTMF events received from the callee surface via theon_dtmfcallback set at construction time. To send DTMF, usesend_dtmf().interrupt(after_words=N)raisesUnsupportedCapabilityErroron this adapter — Media Streams delivers raw audio without incremental transcripts. Useinterrupt(after=seconds)instead.Expand source code
class TwilioAgentAdapter(VoiceAgentAdapter): """ Bidirectional Twilio Media Streams adapter. Same class, same state, either direction: adapter = TwilioAgentAdapter( account_sid=..., auth_token=..., phone_number="+14155551234", ) async with adapter: # connect() / disconnect() await adapter.place_call(to="+14155557777") # OR wait_for_call() # ... scenario.run(...) feeds send_audio / recv_audio ... The adapter is the *only* adapter with ``dtmf=True``. DTMF events received from the callee surface via the ``on_dtmf`` callback set at construction time. To send DTMF, use ``send_dtmf()``. ``interrupt(after_words=N)`` raises ``UnsupportedCapabilityError`` on this adapter — Media Streams delivers raw audio without incremental transcripts. Use ``interrupt(after=seconds)`` instead. """ capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities( streaming_transcripts=False, native_vad=False, dtmf=True, # Twilio Media Streams ``clear`` event drops all buffered outbound # audio. Used by ``adapter.interrupt()`` already wired below. interruption=True, input_formats=["mulaw/8000"], output_formats=["mulaw/8000"], ) # ------------------------------------------------------------------ init def __init__( self, *, account_sid: str, auth_token: str, phone_number: str, public_base_url: Optional[str] = None, allowed_callers: Optional[list[str]] = None, on_dtmf: Optional[Callable[[str], None]] = None, http_port: int = 8765, role: AgentRole = AgentRole.AGENT, validate_signature: bool = True, ) -> None: super().__init__() validate_e164(phone_number) self.account_sid = account_sid self.auth_token = auth_token self.phone_number = phone_number self.public_base_url = public_base_url self.allowed_callers = set(allowed_callers) if allowed_callers else None self.on_dtmf = on_dtmf self.http_port = http_port self.role = role # type: ignore[misc] # When True (default), the inbound /twilio/voice route requires a # valid X-Twilio-Signature header before accepting the webhook # body. The cloudflared tunnel URL is ephemeral and not # guessable, but anyone who learns it could otherwise POST fake # Twilio webhook events into the harness. Tests that don't have # real Twilio credentials disable this with # ``validate_signature=False``; production callers should leave # it on. self.validate_signature = validate_signature if not validate_signature: logger.warning( "TwilioAgentAdapter: validate_signature=False — inbound " "webhooks accept any payload without signature checks. " "Use only in tests; do not deploy to production." ) # Populated during connect(); None when disconnected. self._rest: Optional[TwilioRESTHelper] = None self._phone_number_sid: Optional[str] = None self._prior_voice_url: Optional[str] = None # Set by place_call() when it rewrites the callee's voice_url so # B-leg's webhook lands on our harness. Restored in disconnect. self._callee_phone_number_sid: Optional[str] = None self._prior_callee_voice_url: Optional[str] = None # Set by the first of wait_for_call()/place_call(); subsequent calls to # the other method raise. "idle" after connect() before either fires. self._mode: TwilioAdapterMode = "idle" # Server / media stream state. self._server_task: Optional[asyncio.Task] = None self._server_shutdown: Optional[asyncio.Event] = None self._call_sid: Optional[str] = None self._stream_sid: Optional[str] = None self._stream_connected: Optional[asyncio.Event] = None self._stream_ws: Any = None # starlette WebSocket self._inbound_queue: Optional[asyncio.Queue[AudioChunk]] = None # Set True by the media-stream loop's terminal path (stop / socket close # / throw) the moment it enqueues the end-of-call sentinel. Once the call # has ended, recv_audio must keep draining the queue (and hand back the # sentinel) WITHOUT re-asserting transport liveness — the real # production wrapper nulls _stream_ws/_stream_sid synchronously right # after the loop returns, so the drain's second recv_audio would # otherwise hit _assert_stream_live and crash (#695). Reset on # connect(), disconnect(), and at media-stream-loop entry (per-call # scope — a second session on the same connected adapter must not # inherit the previous session's flag). self._stream_ended: bool = False # Call-lifetime counters stamped onto ``voice.adapter.disconnect`` from # inside disconnect() (#775 Tier 2b — mirrors ElevenLabs' pump-counter # seam). Accumulated by the media loop (`_twilio_server.py`) and the # `/twilio/voice` webhook handler; reset on connect()/disconnect() like # `_stream_ended` above. self._frames_received: int = 0 self._dtmf_received: int = 0 self._stream_ended_reason: TwilioStreamEndedReason = "none" self._webhook_invocations: int = 0 self._webhook_rejected: int = 0 # ------------------------------------------------------------------ repr def __repr__(self) -> str: # redact credentials return ( f"TwilioAgentAdapter(" f"phone_number={self.phone_number!r}, " f"account_sid='***', auth_token='***', " f"public_base_url={self.public_base_url!r})" ) # ------------------------------------------------------------------ lifecycle async def connect(self) -> None: """Resolve number SID and start the FastAPI webhook + WS server. Does NOT modify the Twilio account's ``voice_url``. That side-effect only happens when ``wait_for_call()`` is invoked — callers (who will use ``place_call()``) never overwrite their number's inbound webhook, which makes caller-mode adapters safe to run against a shared pool of Twilio numbers without clobbering anyone's prod webhook. Idempotent: calling connect() on an already-connected adapter is a no-op. This lets the scenario executor's auto-connect step (``_voice_connect_all``) coexist with explicit harness-driven connects (``TwilioHarness`` ``__aenter__``) that already brought the adapter up before scenario.run() was called. """ if self._rest is not None: return if self.public_base_url is None: raise RuntimeError( "TwilioAgentAdapter: public_base_url is required. Wrap the " "adapter in scenario.voice.testing.TwilioHarness, or supply " "a stable public HTTPS URL that routes to this machine." ) self._rest = TwilioRESTHelper(self.account_sid, self.auth_token) self._phone_number_sid = self._rest.resolve_phone_number_sid(self.phone_number) # Stamp Twilio-specific attrs onto the active ``voice.adapter.connect`` # span (opened by the executor connect loop). Base spans are name-owned; # the adapter contributes attributes, never a parallel span name — mirror # ElevenLabs' ``voice.elevenlabs.agent_id`` seam (``adapters/elevenlabs.py``). # NOTE: no ``voice.twilio.direction`` here — connect() is direction- # agnostic (``_mode`` stays "idle" until place_call()/wait_for_call(), # after this span has already closed); direction is stamped on the NEW # ``voice.adapter.dial`` span instead (see place_call/wait_for_call). set_span_attributes( _otel_trace.get_current_span(), { "voice.twilio.phone_number_sid": self._phone_number_sid, "voice.twilio.validate_signature": self.validate_signature, "voice.twilio.webhook_port": self.http_port, }, ) self._stream_connected = asyncio.Event() self._inbound_queue = asyncio.Queue() self._server_shutdown = asyncio.Event() self._stream_ended = False self._frames_received = 0 self._dtmf_received = 0 self._stream_ended_reason = "none" self._webhook_invocations = 0 self._webhook_rejected = 0 self._mode = "idle" # Webhook server is its own unit — see _twilio_server.py. The # adapter only orchestrates lifecycle; the routes, signature # validation, and WS framing live in TwilioWebhookServer. from ._twilio_server import TwilioWebhookServer self._webhook_server: Optional[TwilioWebhookServer] = TwilioWebhookServer(self) self._server_task = asyncio.create_task(self._run_server()) # Give uvicorn a beat to bind the port before Twilio hits it. await asyncio.sleep(0.2) async def disconnect(self) -> None: """Restore prior voice_url (answer mode only), tear down server. Best-effort on errors. In caller mode we never touched the Twilio number's voice_url, so there's nothing to restore. """ if self._rest is None: return # 1. Restore webhook first so Twilio doesn't keep hitting a dead URL. # ``rest_restore_failed`` tracks whether either restore below raised — # previously fully swallowed via ``suppress(Exception)``; now surfaced # onto the disconnect span (#775 Tier 2b) so a Twilio REST outage during # teardown is no longer invisible. The swallow behavior itself (a failed # restore must never block disconnect()) is unchanged. rest_restore_failed = False if self._mode == "answer" and self._phone_number_sid is not None: try: prior = self._prior_voice_url or "" self._rest.write_voice_url(self._phone_number_sid, prior) logger.debug( "TwilioAgentAdapter: restored voice_url=%r on %s", prior, self._phone_number_sid, ) except Exception: rest_restore_failed = True # place_call() rewrites the CALLEE's voice_url to attach Media # Streams to B-leg. Restore that too. if self._mode == "call" and self._callee_phone_number_sid is not None: try: prior_b = self._prior_callee_voice_url or "" self._rest.write_voice_url(self._callee_phone_number_sid, prior_b) logger.debug( "TwilioAgentAdapter: restored callee voice_url=%r on %s", prior_b, self._callee_phone_number_sid, ) except Exception: rest_restore_failed = True # 2. Signal server to shut down, then wait for the task. This is what # ACTUALLY closes a still-live media stream (uvicorn's graceful # shutdown tears down the open WS), which is what lets the media # loop's ``finally``/except-handlers (``_twilio_server.py``) make # their FINAL writes to ``_frames_received`` / ``_stream_ended_reason`` # / etc. The counter stamp below MUST run after this await, not # before — stamping first (the #775 review-caught bug) reads the # pre-teardown snapshot and reports ``stream_ended_reason=="none"`` / # undercounted frames for a call that was still live when disconnect() # was invoked, even though the call did in fact end (via this very # shutdown) moments later. if self._server_shutdown is not None: self._server_shutdown.set() if self._server_task is not None: with suppress(Exception): await asyncio.wait_for(self._server_task, timeout=3.0) # Stamp call-lifetime counters onto the active ``voice.adapter.disconnect`` # span (opened by the executor disconnect loop) — mirrors ElevenLabs' # pump-counter stamp (``adapters/elevenlabs.py``). Sampled AFTER the # shutdown-await above (not before) so a still-live call's FINAL # counts/reason are captured, not a mid-call snapshot. The disconnect # span stays the current span across the await (contextvars survive # awaits within the same task), so ``get_current_span()`` here still # targets it. set_span_attributes( _otel_trace.get_current_span(), { "voice.twilio.frames_received": self._frames_received, "voice.twilio.dtmf_received": self._dtmf_received, "voice.twilio.stream_ended_reason": self._stream_ended_reason, "voice.twilio.webhook_invocations": self._webhook_invocations, "voice.twilio.webhook_rejected": self._webhook_rejected, "voice.twilio.rest_restore_failed": rest_restore_failed, }, ) # 3. Reset state. self._rest = None self._phone_number_sid = None self._prior_voice_url = None self._callee_phone_number_sid = None self._prior_callee_voice_url = None self._mode = "idle" self._server_task = None self._server_shutdown = None self._call_sid = None self._stream_sid = None self._stream_connected = None self._stream_ws = None if self._inbound_queue is not None: # A recv_audio() task may already be blocked in get() on this queue. # Wake that in-flight consumer before dropping our reference; setting # _inbound_queue to None alone leaves it parked until its receive # timeout expires (#757). An empty chunk is the adapter's existing # end-of-stream signal and lets the shared drain finish cleanly. self._inbound_queue.put_nowait(AudioChunk(data=b"")) self._inbound_queue = None self._stream_ended = False self._frames_received = 0 self._dtmf_received = 0 self._stream_ended_reason = "none" self._webhook_invocations = 0 self._webhook_rejected = 0 # ------------------------------------------------------------------ direction async def place_call( self, to: str, *, timeout: float = 120.0, attach_stream_to_self: bool = True, ) -> None: """ Originate an outbound call from this adapter's Twilio number to ``to``. Twilio's REST ``Calls.create`` runs TwiML on TWO legs of the resulting call: - **A-leg** (the originator, ``from_=self.phone_number``): runs the inline TwiML passed via ``twiml=`` to ``Calls.create``. We use ``<Pause length=120>`` so the originator just holds the bridge open while the demo runs. - **B-leg** (the callee, ``to=``): when Twilio dials B and B picks up, B's number's ``voice_url`` fires — that's where the bridge's Media Streams attach. This is identical to the inbound demo's flow: B's voice_url returns ``<Connect><Stream>``, the WS opens, audio flows. So ``place_call`` only makes sense when ``to`` is another Twilio number on this account whose ``voice_url`` is set to OUR harness webhook. To make that wiring automatic, ``place_call`` temporarily rewrites B's ``voice_url`` for the duration of the call and restores it on ``disconnect``. The harness on this adapter's own number does NOT need to be answer-mode — the Stream attaches to B's leg, NOT this adapter's leg, but B's webhook is hosted on this adapter's local server, so the WS still lands here. The bidirectional audio model is unchanged: ``send_audio`` writes frames over the WS (B hears them and bridges to A), ``recv_audio`` reads inbound frames off the WS (whatever the bridge mixes from both legs). Limitation: ``to`` MUST be a phone number on this same Twilio account. Calling an external PSTN endpoint (a real cell phone) requires a different topology (``<Start><Stream>`` + ``<Dial>``) which we don't implement here because the inline TwiML route on the A-leg can't capture B's audio when B is external. Default timeout 120s covers cloudflared cold-start latency. Raises: RuntimeError: If called after ``wait_for_call()`` (modes are exclusive per adapter instance), or if ``to`` is not a Twilio number on this account. ValueError: If ``to`` is not in E.164 format. asyncio.TimeoutError: If the media stream doesn't open within ``timeout`` seconds. """ self._assert_connected() self._enter_mode("call") validate_e164(to) assert self.public_base_url is not None assert self._rest is not None assert self._stream_connected is not None # NEW ``voice.adapter.dial`` span (#775 Tier 2a): self-instrumented, # since no executor span is active by the time place_call()/ # wait_for_call() run (the ``voice.adapter.connect`` span already # closed). Wraps the REST dial + the ``_stream_connected`` wait — the # "I placed the call but no media ever streamed" failure surface. with voice_span( "voice.adapter.dial", { "voice.adapter.class": type(self).__name__, "voice.twilio.direction": "outbound", "voice.twilio.to": _redact_e164(to), "voice.twilio.from": _redact_e164(self.phone_number), }, ) as _dial: if attach_stream_to_self: # Resolve B-leg's number SID and snapshot+rewrite its voice_url so # B's leg attaches its Media Stream to our harness webhook. We own # this number (same Twilio account); disconnect() will restore. self._callee_phone_number_sid = self._rest.resolve_phone_number_sid(to) self._prior_callee_voice_url = self._rest.read_voice_url( self._callee_phone_number_sid ) webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice" self._rest.write_voice_url(self._callee_phone_number_sid, webhook_url) logger.info( "TwilioAgentAdapter: rewrote callee %s voice_url to %s", _redact_e164(to), webhook_url, ) # A-leg TwiML: play a short deterministic <Say> line, then hold # the bridge open. Twilio runs this on the originator side while # B's webhook attaches the Media Stream. # # The <Say> gives the recording a known-good utterance to # transcribe. A bare <Pause> alone produces 120s of line silence # that Whisper has been observed to hallucinate as non-English # text (issue #465 in this PR). The Say is a one-time anchor at # call setup; the Media Stream carries the real bidirectional # conversation that follows. inline_a_leg_twiml = ( '<?xml version="1.0" encoding="UTF-8"?>' "<Response>" f'<Say voice="Polly.Joanna">{PLACE_CALL_A_LEG_SAY_TEXT}</Say>' '<Pause length="120"/>' "</Response>" ) self._call_sid = self._rest.place_call( to=to, from_=self.phone_number, twiml=inline_a_leg_twiml ) logger.info( "TwilioAgentAdapter: placed call %s from %s to %s", self._call_sid, _redact_e164(self.phone_number), _redact_e164(to), ) set_span_attributes(_dial, {"voice.twilio.call_sid": self._call_sid}) if attach_stream_to_self: # Wait for OUR webhook to fire — only meaningful when we rewrote # the callee's voice_url to point at us. In originator-only mode # (attach_stream_to_self=False), there's no stream coming to us; # the callee has its own harness which owns the stream. _dial_wait_started = time.monotonic() try: await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout) except asyncio.TimeoutError: # The media stream never connected — the marquee "I placed # the call but no media ever streamed" failure. Tag the # outcome BEFORE re-raising; the raised TimeoutError still # marks the span ERROR (voice_span's own exception handling). _dial.set_attribute( "voice.twilio.dial_outcome", "stream_connect_timeout" ) raise set_span_attributes( _dial, { "voice.twilio.stream_connect_latency_ms": round( (time.monotonic() - _dial_wait_started) * 1000 ), }, ) async def wait_for_call(self, timeout: float = 120.0) -> None: """ Block until someone dials in and the media stream is live. In **default mode** (no conference_room): the number's ``voice_url`` is overwritten to point at our webhook so inbound calls reach us. Caller (``place_call``) elsewhere will dial this number. In **conference mode**: there's no inbound call to wait for — the two-Twilio-number demo can't naturally have one side receive an inbound call when both legs need conference TwiML. Instead, we ORIGINATE an outbound call FROM this adapter's number TO itself with inline TwiML that opens the capture stream and dials into the shared conference room. This is the symmetric counterpart to ``place_call`` in conference mode — both adapters end up as participants in the same room, exchanging audio via the bridge. Default timeout 120s covers cloudflared cold-start, conference-room formation, and Twilio's webhook ramp. Raises: RuntimeError: If called after ``place_call()``. asyncio.TimeoutError: If nobody dials in within ``timeout``. """ self._assert_connected() self._enter_mode("answer") assert self.public_base_url is not None assert self._rest is not None assert self._phone_number_sid is not None assert self._stream_connected is not None # NEW ``voice.adapter.dial`` span (#775 Tier 2a) — see place_call()'s # comment for the rationale (self-instrumented, no executor span active # here). with voice_span( "voice.adapter.dial", { "voice.adapter.class": type(self).__name__, "voice.twilio.direction": "inbound", "voice.twilio.to": _redact_e164(self.phone_number), }, ) as _dial: # Snapshot the prior webhook so we can restore it on disconnect, then # point the number at our server. Only answer mode does this. self._prior_voice_url = self._rest.read_voice_url(self._phone_number_sid) webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice" self._rest.write_voice_url(self._phone_number_sid, webhook_url) logger.info("TwilioAgentAdapter: webhook set to %s", webhook_url) _dial_wait_started = time.monotonic() try: await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout) except asyncio.TimeoutError: # Nobody dialed in — tag the outcome BEFORE re-raising; the # raised TimeoutError still marks the span ERROR (voice_span's # own exception handling). _dial.set_attribute( "voice.twilio.dial_outcome", "stream_connect_timeout" ) raise set_span_attributes( _dial, { "voice.twilio.call_sid": self._call_sid, "voice.twilio.stream_connect_latency_ms": round( (time.monotonic() - _dial_wait_started) * 1000 ), }, ) def _enter_mode(self, mode: TwilioAdapterMode) -> None: """Transition idle → mode, or raise if already in a different mode. Modes are exclusive per connected session: an adapter can place a call or answer a call, not both. Disconnect + reconnect to reuse the instance in the other direction. """ if self._mode == mode: return # idempotent re-entry (e.g. retrying place_call after timeout) if self._mode != "idle": raise RuntimeError( f"TwilioAgentAdapter: already in {self._mode!r} mode; cannot " f"switch to {mode!r}. Disconnect and reconnect to reuse this " f"adapter in the other direction." ) self._mode = mode # ------------------------------------------------------------------ I/O async def send_audio(self, chunk: AudioChunk) -> None: # Pace at real-time (one frame per TWILIO_FRAME_MS). Without pacing the # whole utterance arrives in milliseconds, which trips bots' VAD into # a clipped-utterance reading. self._assert_stream_live() ws = self._stream_ws stream_sid = self._stream_sid assert ws is not None and stream_sid is not None mulaw = pcm16_24k_to_mulaw8k(chunk.data) frame_secs = TWILIO_FRAME_MS / 1000 for frame in iter_mulaw_frames(mulaw): if not frame: continue await ws.send_text(build_media_frame(stream_sid, frame)) await asyncio.sleep(frame_secs) async def recv_audio(self, timeout: float) -> AudioChunk: # Once the media-stream loop has ended (stop / socket close / throw), # the production wrapper (run_stream_session in _twilio_server) nulls # _stream_ws/_stream_sid synchronously right after the loop returns. The # drain loop (_drain_agent_response) always makes a *second* recv_audio # call after the first chunk — by then liveness has flipped. So the # liveness assert only guards the genuinely-live-and-idle case; buffered # audio and the end-of-call drain bypass it (#695). # # Two sentinel sources cooperate here, and BOTH are load-bearing: the # loop's ``finally`` ENQUEUES one empty chunk — that is what wakes a # consumer already blocked in ``queue.get()`` at the moment the call # ends (flipping ``_stream_ended`` alone wakes nobody) — and this method # SYNTHESIZES further empty chunks for every call after the queue has # drained (the tail-silence probe, or any caller that keeps polling). # Delete either and a hang comes back. if self._inbound_queue is None: # disconnect() tore the adapter down mid-drain — nothing left to # read. In the current teardown order _assert_connected() raises # first (_rest is nulled before _inbound_queue, synchronously), so # the explicit raise below is a reordering guard, not a live path — # it keeps this method's error honest if disconnect() is ever # resequenced. self._assert_connected() raise RuntimeError( "TwilioAgentAdapter: inbound queue is gone; adapter disconnected." ) if not self._inbound_queue.empty(): # Buffered audio or the loop-enqueued terminal sentinel — drain it, # live or not. return await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout) if self._stream_ended: # Call ended and fully drained: synthesize another empty sentinel. return AudioChunk(data=b"") # Live call, nothing buffered yet: assert liveness, then wait. self._assert_stream_live() return await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout) async def send_dtmf(self, tones: str) -> None: """Send DTMF digits on the live call (uses Twilio REST ``<Play digits>``).""" if self._rest is None or self._call_sid is None: raise RuntimeError("TwilioAgentAdapter: no active call; send_dtmf requires an in-progress call") # Run blocking REST call off-thread so we don't stall the event loop. await asyncio.to_thread(self._rest.send_dtmf_on_call, self._call_sid, tones) async def interrupt(self) -> None: """Drop any buffered outbound audio on Twilio's side (``clear`` event).""" self._assert_stream_live() ws = self._stream_ws stream_sid = self._stream_sid assert ws is not None and stream_sid is not None await ws.send_text(build_clear_frame(stream_sid)) # ------------------------------------------------------------------ server async def _run_server(self) -> None: """Thin lifecycle wrapper; the real work lives in TwilioWebhookServer.""" assert self._webhook_server is not None await self._webhook_server.run() def _build_app(self) -> Any: """Test seam: build the FastAPI app for in-process exercise. Production code never calls this — the server's ``run()`` builds the app itself when uvicorn starts. Existing unit tests use ``TestClient(_build_app())`` to exercise the routes without binding a port; the delegation keeps that test surface stable. """ assert self._webhook_server is not None return self._webhook_server.build_app() async def _media_stream_loop(self, ws: Any) -> None: """Test seam: kick off the Media Streams WS loop directly. The two-adapter-bridge test in ``tests/voice/test_twilio_two_adapter_bridge.py`` uses this to drive a loopback WS without going through the FastAPI route wrapper. Production code reaches the loop via the ``/twilio/stream`` WebSocket handler defined in ``_twilio_server.build_app``. """ assert self._webhook_server is not None await self._webhook_server.media_stream_loop(ws) # ------------------------------------------------------------------ assertions def _assert_connected(self) -> None: if self._rest is None: raise RuntimeError("TwilioAgentAdapter: not connected; call connect() or use `async with`.") def _assert_stream_live(self) -> None: self._assert_connected() if self._stream_ws is None or self._stream_sid is None: raise RuntimeError( "TwilioAgentAdapter: no live media stream. Call place_call() or " "wait_for_call() first." )Ancestors
- VoiceAgentAdapter
- AgentAdapter
- abc.ABC
Class variables
var capabilities : ClassVar[AdapterCapabilities]
Methods
async def connect(self) ‑> None-
Resolve number SID and start the FastAPI webhook + WS server.
Does NOT modify the Twilio account's
voice_url. That side-effect only happens whenwait_for_call()is invoked — callers (who will useplace_call()) never overwrite their number's inbound webhook, which makes caller-mode adapters safe to run against a shared pool of Twilio numbers without clobbering anyone's prod webhook.Idempotent: calling connect() on an already-connected adapter is a no-op. This lets the scenario executor's auto-connect step (
_voice_connect_all) coexist with explicit harness-driven connects (TwilioHarness__aenter__) that already brought the adapter up before scenario.run() was called.Expand source code
async def connect(self) -> None: """Resolve number SID and start the FastAPI webhook + WS server. Does NOT modify the Twilio account's ``voice_url``. That side-effect only happens when ``wait_for_call()`` is invoked — callers (who will use ``place_call()``) never overwrite their number's inbound webhook, which makes caller-mode adapters safe to run against a shared pool of Twilio numbers without clobbering anyone's prod webhook. Idempotent: calling connect() on an already-connected adapter is a no-op. This lets the scenario executor's auto-connect step (``_voice_connect_all``) coexist with explicit harness-driven connects (``TwilioHarness`` ``__aenter__``) that already brought the adapter up before scenario.run() was called. """ if self._rest is not None: return if self.public_base_url is None: raise RuntimeError( "TwilioAgentAdapter: public_base_url is required. Wrap the " "adapter in scenario.voice.testing.TwilioHarness, or supply " "a stable public HTTPS URL that routes to this machine." ) self._rest = TwilioRESTHelper(self.account_sid, self.auth_token) self._phone_number_sid = self._rest.resolve_phone_number_sid(self.phone_number) # Stamp Twilio-specific attrs onto the active ``voice.adapter.connect`` # span (opened by the executor connect loop). Base spans are name-owned; # the adapter contributes attributes, never a parallel span name — mirror # ElevenLabs' ``voice.elevenlabs.agent_id`` seam (``adapters/elevenlabs.py``). # NOTE: no ``voice.twilio.direction`` here — connect() is direction- # agnostic (``_mode`` stays "idle" until place_call()/wait_for_call(), # after this span has already closed); direction is stamped on the NEW # ``voice.adapter.dial`` span instead (see place_call/wait_for_call). set_span_attributes( _otel_trace.get_current_span(), { "voice.twilio.phone_number_sid": self._phone_number_sid, "voice.twilio.validate_signature": self.validate_signature, "voice.twilio.webhook_port": self.http_port, }, ) self._stream_connected = asyncio.Event() self._inbound_queue = asyncio.Queue() self._server_shutdown = asyncio.Event() self._stream_ended = False self._frames_received = 0 self._dtmf_received = 0 self._stream_ended_reason = "none" self._webhook_invocations = 0 self._webhook_rejected = 0 self._mode = "idle" # Webhook server is its own unit — see _twilio_server.py. The # adapter only orchestrates lifecycle; the routes, signature # validation, and WS framing live in TwilioWebhookServer. from ._twilio_server import TwilioWebhookServer self._webhook_server: Optional[TwilioWebhookServer] = TwilioWebhookServer(self) self._server_task = asyncio.create_task(self._run_server()) # Give uvicorn a beat to bind the port before Twilio hits it. await asyncio.sleep(0.2) async def disconnect(self) ‑> None-
Restore prior voice_url (answer mode only), tear down server.
Best-effort on errors. In caller mode we never touched the Twilio number's voice_url, so there's nothing to restore.
Expand source code
async def disconnect(self) -> None: """Restore prior voice_url (answer mode only), tear down server. Best-effort on errors. In caller mode we never touched the Twilio number's voice_url, so there's nothing to restore. """ if self._rest is None: return # 1. Restore webhook first so Twilio doesn't keep hitting a dead URL. # ``rest_restore_failed`` tracks whether either restore below raised — # previously fully swallowed via ``suppress(Exception)``; now surfaced # onto the disconnect span (#775 Tier 2b) so a Twilio REST outage during # teardown is no longer invisible. The swallow behavior itself (a failed # restore must never block disconnect()) is unchanged. rest_restore_failed = False if self._mode == "answer" and self._phone_number_sid is not None: try: prior = self._prior_voice_url or "" self._rest.write_voice_url(self._phone_number_sid, prior) logger.debug( "TwilioAgentAdapter: restored voice_url=%r on %s", prior, self._phone_number_sid, ) except Exception: rest_restore_failed = True # place_call() rewrites the CALLEE's voice_url to attach Media # Streams to B-leg. Restore that too. if self._mode == "call" and self._callee_phone_number_sid is not None: try: prior_b = self._prior_callee_voice_url or "" self._rest.write_voice_url(self._callee_phone_number_sid, prior_b) logger.debug( "TwilioAgentAdapter: restored callee voice_url=%r on %s", prior_b, self._callee_phone_number_sid, ) except Exception: rest_restore_failed = True # 2. Signal server to shut down, then wait for the task. This is what # ACTUALLY closes a still-live media stream (uvicorn's graceful # shutdown tears down the open WS), which is what lets the media # loop's ``finally``/except-handlers (``_twilio_server.py``) make # their FINAL writes to ``_frames_received`` / ``_stream_ended_reason`` # / etc. The counter stamp below MUST run after this await, not # before — stamping first (the #775 review-caught bug) reads the # pre-teardown snapshot and reports ``stream_ended_reason=="none"`` / # undercounted frames for a call that was still live when disconnect() # was invoked, even though the call did in fact end (via this very # shutdown) moments later. if self._server_shutdown is not None: self._server_shutdown.set() if self._server_task is not None: with suppress(Exception): await asyncio.wait_for(self._server_task, timeout=3.0) # Stamp call-lifetime counters onto the active ``voice.adapter.disconnect`` # span (opened by the executor disconnect loop) — mirrors ElevenLabs' # pump-counter stamp (``adapters/elevenlabs.py``). Sampled AFTER the # shutdown-await above (not before) so a still-live call's FINAL # counts/reason are captured, not a mid-call snapshot. The disconnect # span stays the current span across the await (contextvars survive # awaits within the same task), so ``get_current_span()`` here still # targets it. set_span_attributes( _otel_trace.get_current_span(), { "voice.twilio.frames_received": self._frames_received, "voice.twilio.dtmf_received": self._dtmf_received, "voice.twilio.stream_ended_reason": self._stream_ended_reason, "voice.twilio.webhook_invocations": self._webhook_invocations, "voice.twilio.webhook_rejected": self._webhook_rejected, "voice.twilio.rest_restore_failed": rest_restore_failed, }, ) # 3. Reset state. self._rest = None self._phone_number_sid = None self._prior_voice_url = None self._callee_phone_number_sid = None self._prior_callee_voice_url = None self._mode = "idle" self._server_task = None self._server_shutdown = None self._call_sid = None self._stream_sid = None self._stream_connected = None self._stream_ws = None if self._inbound_queue is not None: # A recv_audio() task may already be blocked in get() on this queue. # Wake that in-flight consumer before dropping our reference; setting # _inbound_queue to None alone leaves it parked until its receive # timeout expires (#757). An empty chunk is the adapter's existing # end-of-stream signal and lets the shared drain finish cleanly. self._inbound_queue.put_nowait(AudioChunk(data=b"")) self._inbound_queue = None self._stream_ended = False self._frames_received = 0 self._dtmf_received = 0 self._stream_ended_reason = "none" self._webhook_invocations = 0 self._webhook_rejected = 0 async def interrupt(self) ‑> None-
Drop any buffered outbound audio on Twilio's side (
clearevent).Expand source code
async def interrupt(self) -> None: """Drop any buffered outbound audio on Twilio's side (``clear`` event).""" self._assert_stream_live() ws = self._stream_ws stream_sid = self._stream_sid assert ws is not None and stream_sid is not None await ws.send_text(build_clear_frame(stream_sid)) async def place_call(self, to: str, *, timeout: float = 120.0, attach_stream_to_self: bool = True) ‑> None-
Originate an outbound call from this adapter's Twilio number to
to.Twilio's REST
Calls.createruns TwiML on TWO legs of the resulting call:- A-leg (the originator,
from_=self.phone_number): runs the inline TwiML passed viatwiml=toCalls.create. We use<Pause length=120>so the originator just holds the bridge open while the demo runs. - B-leg (the callee,
to=): when Twilio dials B and B picks up, B's number'svoice_urlfires — that's where the bridge's Media Streams attach. This is identical to the inbound demo's flow: B's voice_url returns<Connect><Stream>, the WS opens, audio flows.
So
place_callonly makes sense whentois another Twilio number on this account whosevoice_urlis set to OUR harness webhook. To make that wiring automatic,place_calltemporarily rewrites B'svoice_urlfor the duration of the call and restores it ondisconnect. The harness on this adapter's own number does NOT need to be answer-mode — the Stream attaches to B's leg, NOT this adapter's leg, but B's webhook is hosted on this adapter's local server, so the WS still lands here.The bidirectional audio model is unchanged:
send_audiowrites frames over the WS (B hears them and bridges to A),recv_audioreads inbound frames off the WS (whatever the bridge mixes from both legs).Limitation:
toMUST be a phone number on this same Twilio account. Calling an external PSTN endpoint (a real cell phone) requires a different topology (<Start><Stream>+<Dial>) which we don't implement here because the inline TwiML route on the A-leg can't capture B's audio when B is external.Default timeout 120s covers cloudflared cold-start latency.
Raises
RuntimeError- If called after
wait_for_call()(modes are exclusive per adapter instance), or iftois not a Twilio number on this account. ValueError- If
tois not in E.164 format. asyncio.TimeoutError- If the media stream doesn't open within
timeoutseconds.
Expand source code
async def place_call( self, to: str, *, timeout: float = 120.0, attach_stream_to_self: bool = True, ) -> None: """ Originate an outbound call from this adapter's Twilio number to ``to``. Twilio's REST ``Calls.create`` runs TwiML on TWO legs of the resulting call: - **A-leg** (the originator, ``from_=self.phone_number``): runs the inline TwiML passed via ``twiml=`` to ``Calls.create``. We use ``<Pause length=120>`` so the originator just holds the bridge open while the demo runs. - **B-leg** (the callee, ``to=``): when Twilio dials B and B picks up, B's number's ``voice_url`` fires — that's where the bridge's Media Streams attach. This is identical to the inbound demo's flow: B's voice_url returns ``<Connect><Stream>``, the WS opens, audio flows. So ``place_call`` only makes sense when ``to`` is another Twilio number on this account whose ``voice_url`` is set to OUR harness webhook. To make that wiring automatic, ``place_call`` temporarily rewrites B's ``voice_url`` for the duration of the call and restores it on ``disconnect``. The harness on this adapter's own number does NOT need to be answer-mode — the Stream attaches to B's leg, NOT this adapter's leg, but B's webhook is hosted on this adapter's local server, so the WS still lands here. The bidirectional audio model is unchanged: ``send_audio`` writes frames over the WS (B hears them and bridges to A), ``recv_audio`` reads inbound frames off the WS (whatever the bridge mixes from both legs). Limitation: ``to`` MUST be a phone number on this same Twilio account. Calling an external PSTN endpoint (a real cell phone) requires a different topology (``<Start><Stream>`` + ``<Dial>``) which we don't implement here because the inline TwiML route on the A-leg can't capture B's audio when B is external. Default timeout 120s covers cloudflared cold-start latency. Raises: RuntimeError: If called after ``wait_for_call()`` (modes are exclusive per adapter instance), or if ``to`` is not a Twilio number on this account. ValueError: If ``to`` is not in E.164 format. asyncio.TimeoutError: If the media stream doesn't open within ``timeout`` seconds. """ self._assert_connected() self._enter_mode("call") validate_e164(to) assert self.public_base_url is not None assert self._rest is not None assert self._stream_connected is not None # NEW ``voice.adapter.dial`` span (#775 Tier 2a): self-instrumented, # since no executor span is active by the time place_call()/ # wait_for_call() run (the ``voice.adapter.connect`` span already # closed). Wraps the REST dial + the ``_stream_connected`` wait — the # "I placed the call but no media ever streamed" failure surface. with voice_span( "voice.adapter.dial", { "voice.adapter.class": type(self).__name__, "voice.twilio.direction": "outbound", "voice.twilio.to": _redact_e164(to), "voice.twilio.from": _redact_e164(self.phone_number), }, ) as _dial: if attach_stream_to_self: # Resolve B-leg's number SID and snapshot+rewrite its voice_url so # B's leg attaches its Media Stream to our harness webhook. We own # this number (same Twilio account); disconnect() will restore. self._callee_phone_number_sid = self._rest.resolve_phone_number_sid(to) self._prior_callee_voice_url = self._rest.read_voice_url( self._callee_phone_number_sid ) webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice" self._rest.write_voice_url(self._callee_phone_number_sid, webhook_url) logger.info( "TwilioAgentAdapter: rewrote callee %s voice_url to %s", _redact_e164(to), webhook_url, ) # A-leg TwiML: play a short deterministic <Say> line, then hold # the bridge open. Twilio runs this on the originator side while # B's webhook attaches the Media Stream. # # The <Say> gives the recording a known-good utterance to # transcribe. A bare <Pause> alone produces 120s of line silence # that Whisper has been observed to hallucinate as non-English # text (issue #465 in this PR). The Say is a one-time anchor at # call setup; the Media Stream carries the real bidirectional # conversation that follows. inline_a_leg_twiml = ( '<?xml version="1.0" encoding="UTF-8"?>' "<Response>" f'<Say voice="Polly.Joanna">{PLACE_CALL_A_LEG_SAY_TEXT}</Say>' '<Pause length="120"/>' "</Response>" ) self._call_sid = self._rest.place_call( to=to, from_=self.phone_number, twiml=inline_a_leg_twiml ) logger.info( "TwilioAgentAdapter: placed call %s from %s to %s", self._call_sid, _redact_e164(self.phone_number), _redact_e164(to), ) set_span_attributes(_dial, {"voice.twilio.call_sid": self._call_sid}) if attach_stream_to_self: # Wait for OUR webhook to fire — only meaningful when we rewrote # the callee's voice_url to point at us. In originator-only mode # (attach_stream_to_self=False), there's no stream coming to us; # the callee has its own harness which owns the stream. _dial_wait_started = time.monotonic() try: await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout) except asyncio.TimeoutError: # The media stream never connected — the marquee "I placed # the call but no media ever streamed" failure. Tag the # outcome BEFORE re-raising; the raised TimeoutError still # marks the span ERROR (voice_span's own exception handling). _dial.set_attribute( "voice.twilio.dial_outcome", "stream_connect_timeout" ) raise set_span_attributes( _dial, { "voice.twilio.stream_connect_latency_ms": round( (time.monotonic() - _dial_wait_started) * 1000 ), }, ) - A-leg (the originator,
async def send_dtmf(self, tones: str) ‑> None-
Send DTMF digits on the live call (uses Twilio REST
<Play digits>).Expand source code
async def send_dtmf(self, tones: str) -> None: """Send DTMF digits on the live call (uses Twilio REST ``<Play digits>``).""" if self._rest is None or self._call_sid is None: raise RuntimeError("TwilioAgentAdapter: no active call; send_dtmf requires an in-progress call") # Run blocking REST call off-thread so we don't stall the event loop. await asyncio.to_thread(self._rest.send_dtmf_on_call, self._call_sid, tones) async def wait_for_call(self, timeout: float = 120.0) ‑> None-
Block until someone dials in and the media stream is live.
In default mode (no conference_room): the number's
voice_urlis overwritten to point at our webhook so inbound calls reach us. Caller (place_call) elsewhere will dial this number.In conference mode: there's no inbound call to wait for — the two-Twilio-number demo can't naturally have one side receive an inbound call when both legs need conference TwiML. Instead, we ORIGINATE an outbound call FROM this adapter's number TO itself with inline TwiML that opens the capture stream and dials into the shared conference room. This is the symmetric counterpart to
place_callin conference mode — both adapters end up as participants in the same room, exchanging audio via the bridge.Default timeout 120s covers cloudflared cold-start, conference-room formation, and Twilio's webhook ramp.
Raises
RuntimeError- If called after
place_call(). asyncio.TimeoutError- If nobody dials in within
timeout.
Expand source code
async def wait_for_call(self, timeout: float = 120.0) -> None: """ Block until someone dials in and the media stream is live. In **default mode** (no conference_room): the number's ``voice_url`` is overwritten to point at our webhook so inbound calls reach us. Caller (``place_call``) elsewhere will dial this number. In **conference mode**: there's no inbound call to wait for — the two-Twilio-number demo can't naturally have one side receive an inbound call when both legs need conference TwiML. Instead, we ORIGINATE an outbound call FROM this adapter's number TO itself with inline TwiML that opens the capture stream and dials into the shared conference room. This is the symmetric counterpart to ``place_call`` in conference mode — both adapters end up as participants in the same room, exchanging audio via the bridge. Default timeout 120s covers cloudflared cold-start, conference-room formation, and Twilio's webhook ramp. Raises: RuntimeError: If called after ``place_call()``. asyncio.TimeoutError: If nobody dials in within ``timeout``. """ self._assert_connected() self._enter_mode("answer") assert self.public_base_url is not None assert self._rest is not None assert self._phone_number_sid is not None assert self._stream_connected is not None # NEW ``voice.adapter.dial`` span (#775 Tier 2a) — see place_call()'s # comment for the rationale (self-instrumented, no executor span active # here). with voice_span( "voice.adapter.dial", { "voice.adapter.class": type(self).__name__, "voice.twilio.direction": "inbound", "voice.twilio.to": _redact_e164(self.phone_number), }, ) as _dial: # Snapshot the prior webhook so we can restore it on disconnect, then # point the number at our server. Only answer mode does this. self._prior_voice_url = self._rest.read_voice_url(self._phone_number_sid) webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice" self._rest.write_voice_url(self._phone_number_sid, webhook_url) logger.info("TwilioAgentAdapter: webhook set to %s", webhook_url) _dial_wait_started = time.monotonic() try: await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout) except asyncio.TimeoutError: # Nobody dialed in — tag the outcome BEFORE re-raising; the # raised TimeoutError still marks the span ERROR (voice_span's # own exception handling). _dial.set_attribute( "voice.twilio.dial_outcome", "stream_connect_timeout" ) raise set_span_attributes( _dial, { "voice.twilio.call_sid": self._call_sid, "voice.twilio.stream_connect_latency_ms": round( (time.monotonic() - _dial_wait_started) * 1000 ), }, )
Inherited members