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.
#: ``"max_duration"`` is the a-leg duration cap firing (#762 guardrail (b)) —
#: kept distinct from ``"close"`` so a scenario author can tell "we hung the
#: call up at the cap" from "the callee hung up".
TwilioStreamEndedReason = Literal["stop", "close", "error", "none", "max_duration"]
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,
TunnelNotReadyError,
TunnelReadiness,
TwilioRESTHelper,
_redact_e164,
build_clear_frame,
build_media_frame,
escape_xml_attr,
iter_mulaw_frames,
mint_stream_nonce,
normalize_e164,
pcm16_24k_to_mulaw8k,
resolve_max_call_duration,
stream_ws_url,
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."
)
#: Refusal for ``send_dtmf`` under a-leg (#762 AC10). Says WHY, because
#: "unsupported" alone reads as an oversight rather than a deliberate guard.
A_LEG_SEND_DTMF_UNSUPPORTED = (
"TwilioAgentAdapter: send_dtmf is unsupported in external (a-leg) mode. "
"Sending DTMF replaces the live call's TwiML, which redirects the call away "
"from the <Connect><Stream> verb and would end the media stream this "
"scenario is running on. In b-leg mode the stream rides the callee's leg, "
"so the redirect is harmless; in a-leg mode it is the call. Use b-leg mode "
"(an owned callee number) if the scenario needs DTMF."
)
#: Effective stream-attach mode resolved from ``place_call``'s two parameters.
StreamAttachMode = Literal["a-leg", "b-leg", "originator-only"]
def _resolve_stream_mode(
attach_stream: Optional[Literal["a-leg", "b-leg"]],
attach_stream_to_self: Optional[bool],
) -> StreamAttachMode:
"""Resolve the effective ``place_call`` stream-attach mode.
``attach_stream`` (typed) supersedes the legacy ``attach_stream_to_self``
bool, which names the same choice in a narrower vocabulary: ``True`` is
"b-leg", ``False`` is the originator-only third mode ``attach_stream``
cannot express. Passing both is only allowed when they AGREE — every other
combination is a genuine disagreement about where the stream attaches, so
it raises rather than silently discarding the bool.
"""
if attach_stream_to_self is None:
return attach_stream or "b-leg"
implied: StreamAttachMode = "b-leg" if attach_stream_to_self else "originator-only"
if attach_stream is not None and attach_stream != implied:
raise ValueError(
f"place_call: attach_stream={attach_stream!r} conflicts with "
f"attach_stream_to_self={attach_stream_to_self!r}; pass only one."
)
return implied
def _build_connect_stream_twiml(
ws_url: str,
*,
stream_parameters: dict[str, str],
) -> str:
"""Build inline ``<Connect><Stream>`` origination TwiML for a-leg mode.
``stream_parameters`` renders ``<Parameter name=.. value=../>`` children
inside ``<Stream>``; an empty dict renders the self-closing
``<Stream url=".."/>`` form byte-identically to the inbound webhook's TwiML.
Every attribute value goes through ``escape_xml_attr`` so this builder
cannot be broken out of, and so it stays byte-identical to the JS twin
(``buildConnectStreamTwiml``), which has always escaped.
"""
param_children = "".join(
f'<Parameter name="{escape_xml_attr(name)}" '
f'value="{escape_xml_attr(value)}"/>'
for name, value in stream_parameters.items()
)
escaped_url = escape_xml_attr(ws_url)
stream_el = (
f'<Stream url="{escaped_url}">{param_children}</Stream>'
if param_children
else f'<Stream url="{escaped_url}"/>'
)
return (
'<?xml version="1.0" encoding="UTF-8"?>'
"<Response>"
f"<Connect>{stream_el}</Connect>"
"</Response>"
)
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,
allowed_callees: Optional[list[str]] = None,
tunnel_readiness: Optional[TunnelReadiness] = 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
# Destination allowlist for a-leg mode (#762 guardrail (c)). ``None``
# denies every a-leg destination: a-leg dials numbers this account does
# NOT own, so an unguarded ``to`` is an unbounded dialer and the
# capability has to be opted into per destination. Entries are
# normalised (and so validated) here rather than at dial time, so a
# typo fails at setup instead of when a PSTN call is about to be
# billed. b-leg is deliberately not gated on this — it can only reach
# numbers on this account, which is its own guardrail.
self.allowed_callees = (
{normalize_e164(n) for n in allowed_callees} if allowed_callees else None
)
# Edge-readiness probe consulted before a-leg origination (#762
# guardrail (c)). ``CloudflareTunnel`` satisfies it structurally;
# ``TwilioHarness`` wires its own tunnel in. ``None`` means the caller
# owns a stable public URL and has nothing to wait for.
self.tunnel_readiness = tunnel_readiness
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
# Call generation (#762 P1). Bumped by connect()/place_call()/
# wait_for_call() every time we (re)dial. The media loop captures it at
# the moment a socket is ADOPTED and re-checks it on every later side
# effect (enqueue, signal-connected, mark-ended, cancel-watchdog, being
# ``_stream_ws``): a socket adopted under an older generation — one that
# connected and even adopted BEFORE this dial armed the nonce — is stale
# the instant the counter moves, and is closed on its next frame instead
# of injecting audio into, or ending, the current call.
self._call_generation: int = 0
# A-leg WS auth (#762 guardrail (a)). Set by place_call in "a-leg" mode
# only; its non-None-ness is what ARMS media-stream nonce enforcement.
# In b-leg mode the signed POST /twilio/voice precedes the socket, so
# the socket inherits that trust and this stays None — enforcement is
# keyed on the mode we originated in, never on whether the inbound frame
# happens to carry a nonce (which an attacker could simply omit).
self._stream_nonce: Optional[str] = None
# Adapter-side max-call-duration watchdog (#762 guardrail (b)). Armed by
# place_call in "a-leg" mode; the belt to Twilio's own ``TimeLimit``
# suspenders, which is the half that survives this process dying.
self._max_duration_task: Optional[asyncio.Task] = None
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._stream_nonce = None
self._cancel_max_duration_timer()
# Fresh connection lifecycle — bump the generation so any socket left
# over from a previous connect can never pass the media loop's identity
# check against this one.
self._call_generation += 1
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.
# Disarm the duration watchdog first: from here on the call is ours to
# end, and a timer that fired mid-teardown would hang up a SID this
# adapter may already have replaced with a later call's.
self._cancel_max_duration_timer()
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
self._stream_nonce = None
self._cancel_max_duration_timer()
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
def _reset_connection_state_for_dial(self) -> None:
"""Invalidate any socket adopted under a prior call generation (#762 P1).
Bumping ``_call_generation`` is what evicts a socket that connected — and
even ADOPTED — before this dial armed the nonce: the media loop captured
the old generation at adoption and closes itself on its next frame once
the counter moves. We also reset the per-dial connection state the loop
writes, so this dial starts from a clean slate and, critically, cannot
resolve on a stale socket's ``_stream_connected`` signal:
- clear ``_stream_connected`` — otherwise a socket that adopted un-gated
before arming has already fired it and ``place_call`` would return
before any socket authenticates;
- drop ``_stream_ws``/``_stream_sid`` — the stale socket must not remain
our outbound transport (``send_audio`` writes there);
- drain the inbound queue and zero ``_frames_received`` — the stale
socket's media must not surface as this call's audio;
- clear the terminal flag/reason — the stale session's ``stop`` must not
read as this call having ended.
Called by both ``place_call`` and ``wait_for_call``. Inbound/b-leg
behaviour is unchanged: no socket has adopted yet when the genuine one
arrives, so it adopts under the current generation and connects exactly
as before.
"""
self._call_generation += 1
self._stream_ws = None
self._stream_sid = None
if self._stream_connected is not None:
self._stream_connected.clear()
if self._inbound_queue is not None:
while not self._inbound_queue.empty():
self._inbound_queue.get_nowait()
self._stream_ended = False
self._stream_ended_reason = "none"
self._frames_received = 0
async def place_call(
self,
to: str,
*,
timeout: float = 120.0,
attach_stream_to_self: Optional[bool] = None,
attach_stream: Optional[Literal["a-leg", "b-leg"]] = None,
max_call_duration_seconds: Optional[int] = None,
) -> 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 of the default (**b-leg**) mode: ``to`` MUST be a phone
number on this same Twilio account. Calling an external PSTN endpoint
(a real cell phone) is instead served by **a-leg** mode.
Stream-attach mode is selected by ``attach_stream``:
- ``"b-leg"`` — today's default: rewrite the callee's ``voice_url``,
hold the A-leg open with ``<Say>…<Pause>``, restore on disconnect.
Only works for numbers this account owns.
- ``"a-leg"`` — originate with inline ``<Connect><Stream>`` so the
Media Stream rides OUR own leg. Touches nothing on the callee (no
``resolve_phone_number_sid``/``read_voice_url``/``write_voice_url``),
so ``to`` can be any external number. Because that reaches numbers
this account does not own, ``to`` must appear in the adapter's
``allowed_callees`` (deny-by-default) and the public base URL must be
reachable from the edge — both checked before origination.
The legacy ``attach_stream_to_self`` bool is still honoured but
superseded by ``attach_stream``: ``True``/unset → ``"b-leg"``,
``False`` → originator-only (a third mode ``attach_stream`` cannot
express). Passing both with disagreeing intent raises ``ValueError``.
Default timeout 120s covers cloudflared cold-start latency.
``timeout`` and ``max_call_duration_seconds`` are two different clocks
and are configured independently:
- ``timeout`` bounds how long we WAIT FOR THE MEDIA STREAM TO CONNECT
after origination. It says nothing about how long the call may run.
- ``max_call_duration_seconds`` bounds HOW LONG THE CALL MAY LIVE
(a-leg only, default ``DEFAULT_MAX_CALL_DURATION_SECONDS``, hard cap
``MAX_CALL_DURATION_CAP_SECONDS``; a larger request raises). Under
a-leg's ``<Connect>`` the call lives exactly as long as the WebSocket,
so without a ceiling a hung executor keeps a billing PSTN call open
forever. Two mechanisms enforce it: Twilio's own ``TimeLimit`` on
``Calls.create`` (the load-bearing half — it still fires if this
process hangs or is killed) and an adapter-side wall-clock timer that
hangs the call up via REST and closes the socket.
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, or — in a-leg mode —
is not on ``allowed_callees``.
TunnelNotReadyError: If a-leg mode's readiness probe says the
public base URL is not reachable from the edge yet.
asyncio.TimeoutError: If the media stream doesn't open within
``timeout`` seconds.
"""
self._assert_connected()
self._enter_mode("call")
validate_e164(to)
# Resolve the effective stream-attach mode BEFORE any REST call so a
# conflicting-parameter caller error surfaces before we dial. `attach_stream`
# (typed) supersedes the legacy `attach_stream_to_self` bool; the bool's
# third mode (originator-only) is unreachable via `attach_stream` by design.
mode = _resolve_stream_mode(attach_stream, attach_stream_to_self)
# Only a-leg loses <Pause>'s implicit ceiling, so only a-leg carries a
# duration cap. Naming one in another mode is rejected rather than
# ignored: silently dropping it would leave the caller believing the
# call is bounded when it is not.
if mode == "a-leg":
# Guardrail (c): destinations are deny-by-default in a-leg mode.
# Checked here, with the other local caller-fixable failures, so a
# misconfigured allowlist costs nothing and dials nothing.
self._assert_callee_allowed(to)
max_call_duration = resolve_max_call_duration(max_call_duration_seconds)
elif max_call_duration_seconds is not None:
raise ValueError(
"place_call: max_call_duration_seconds is only supported with "
'attach_stream="a-leg"; b-leg and originator-only modes hold the '
"originator leg with <Pause> and are bounded by it."
)
else:
max_call_duration = None
# Guardrail (c), second half: a-leg hands Twilio our public URL and
# Twilio opens the media socket against it seconds later. Probe the edge
# AFTER the free local checks above and BEFORE origination, so an
# unreachable tunnel is a named error instead of a billed call that
# dies in a stream-connect timeout.
if mode == "a-leg":
await self._assert_tunnel_ready()
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:
# #762 P1: bump the call generation and reset per-dial connection
# state BEFORE originating, so a socket that adopted un-gated before
# this dial is evicted and cannot resolve the wait below.
self._reset_connection_state_for_dial()
if mode == "b-leg":
# 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: the Media Stream rides OUR own leg via inline
# <Connect><Stream>, so `to` can be any external number and we
# touch nothing on the callee. Same TwiML shape the inbound webhook
# returns (`_twilio_server.py`), now on origination.
#
# Guardrail (a): the socket is the only inbound signal on this path,
# so it must authenticate itself. Mint a per-call CSPRNG nonce and
# ship it as a <Parameter> child; Twilio echoes it back in the
# `start` frame's customParameters, and the media loop
# (`_twilio_server.py`) closes any socket that cannot present it.
#
# Assigned UNCONDITIONALLY (mirrors the JS twin): the field also
# ARMS nonce enforcement and the a-leg send_dtmf refusal, so a b-leg
# dial that left an earlier a-leg call's nonce standing would gate a
# socket whose `start` frame carries no <Parameter> at all — the
# b-leg call would never connect.
nonce = mint_stream_nonce() if mode == "a-leg" else None
self._stream_nonce = nonce
if nonce is not None:
origination_twiml = _build_connect_stream_twiml(
stream_ws_url(self.public_base_url),
stream_parameters={"nonce": nonce},
)
else:
# A-leg (originator-side) TwiML for b-leg / originator-only modes:
# 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.
origination_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=origination_twiml,
time_limit=max_call_duration,
)
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 max_call_duration is not None:
self._arm_max_duration_timer(max_call_duration, self._call_sid, to)
if mode != "originator-only":
# Wait for the media stream to reach us. In b-leg mode it arrives
# via the callee's rewritten voice_url; in a-leg mode it rides our
# own <Connect><Stream> leg. In originator-only mode there's no
# stream coming to us; the callee has its own harness which owns it.
_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"
)
if mode == "a-leg" and self._call_sid is not None:
# The call is already originated and, under a-leg's
# <Connect>, lives to the duration cap even though
# nothing will ever stream on it. Hang it up before
# handing the caller their TimeoutError — otherwise the
# ordinary connect-failure path bills for the cap.
await self._abandon_originated_call(self._call_sid)
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:
# #762 P1: bump the call generation and reset per-dial connection
# state, symmetric with place_call. For inbound this only guards a
# reconnect/back-to-back dial; a fresh inbound socket still connects
# exactly as before.
self._reset_connection_state_for_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 _assert_callee_allowed(self, to: str) -> None:
"""Refuse an a-leg destination that was not explicitly allowlisted.
Default-deny: no ``allowed_callees`` means no a-leg call at all. The
message names the field to set, because "denied" without "here is how
to allow it" just sends the caller reading source.
Comparison is an exact set lookup on the normalised number, so a
near-miss — a prefix, a suffix, an extra space — is a refusal.
"""
if self.allowed_callees is None:
raise ValueError(
'place_call: attach_stream="a-leg" requires allowed_callees. '
"a-leg dials numbers this Twilio account does not own, so "
"destinations are deny-by-default: pass "
"TwilioAgentAdapter(allowed_callees=[...]) listing every "
f"number this adapter may dial, including {_redact_e164(to)}."
)
if normalize_e164(to) not in self.allowed_callees:
raise ValueError(
f"place_call: destination {_redact_e164(to)} is not in "
"allowed_callees. Add it to "
"TwilioAgentAdapter(allowed_callees=[...]) to permit this "
"destination."
)
def _assert_dtmf_supported(self) -> None:
"""Refuse DTMF on a call whose media stream rides our own leg (#762 AC10).
``send_dtmf`` works by ``calls(sid).update(twiml=...)``, which REPLACES
the TwiML the call is executing. Under b-leg that TwiML is a ``<Say>`` +
``<Pause>`` hold on our leg, so replacing it costs nothing — the Media
Stream lives on the callee's leg. Under a-leg the TwiML being replaced
IS the ``<Connect><Stream>`` verb carrying the scenario, so the same
REST call would redirect the call away from the socket and end the media
session mid-run.
Armed off ``_stream_nonce`` — set only by an a-leg ``place_call``, the
same signal the media loop uses to arm nonce enforcement — rather than
re-derived from the caller's arguments here.
"""
if self._stream_nonce is not None:
raise RuntimeError(A_LEG_SEND_DTMF_UNSUPPORTED)
async def _assert_tunnel_ready(self) -> None:
"""Refuse to originate until the public URL is reachable from the edge.
Delegates to whatever readiness probe was supplied — ``CloudflareTunnel``
already knows how to answer this. No probe means the caller owns a
stable public URL and there is nothing to wait for.
"""
if self.tunnel_readiness is None:
return
try:
await self.tunnel_readiness.wait_until_edge_reachable()
except TunnelNotReadyError:
raise
except Exception as exc:
raise TunnelNotReadyError(
f"place_call: public_base_url {self.public_base_url!r} is not "
"reachable from the edge yet, so Twilio's media stream would "
"connect to nothing. Not originating."
) from exc
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
# ------------------------------------------------------- max call duration
def _arm_max_duration_timer(self, seconds: int, call_sid: str, to: str) -> None:
"""Start the wall-clock watchdog that ends ``call_sid`` at ``seconds``.
Any previously-armed timer is cancelled first, so a second
``place_call`` on the same adapter can never leave an older call's timer
alive to hang up the newer call's SID.
"""
self._cancel_max_duration_timer()
# Capture the generation this watchdog is armed under (#762 P2). After
# the hang-up REST round-trip the handler re-checks it before closing the
# socket, so an expiry whose REST response resolves AFTER a newer call has
# taken over cannot close that newer call's authenticated socket.
generation = self._call_generation
self._max_duration_task = asyncio.create_task(
self._run_max_duration_timer(seconds, call_sid, to, generation)
)
async def _abandon_originated_call(self, call_sid: str) -> None:
"""Disarm the watchdog and hang up a call nothing will ever stream on.
Best-effort on the REST hangup: the caller is already raising, and
Twilio's own ``TimeLimit`` remains the backstop if this leg of the
teardown fails.
"""
self._cancel_max_duration_timer()
rest = self._rest
if rest is not None:
# Blocking REST call off-thread, as send_dtmf does.
with suppress(Exception):
await asyncio.to_thread(rest.end_call, call_sid)
def _cancel_max_duration_timer(self) -> None:
"""Disarm the watchdog. Idempotent; safe when none was ever armed."""
task = self._max_duration_task
self._max_duration_task = None
if task is not None and not task.done():
task.cancel()
async def _await_max_duration(self, seconds: int) -> None:
"""Sleep until the duration cap elapses.
Test seam (same role as ``_build_app``/``_media_stream_loop``): tests
replace this on the instance to drive expiry on controlled time instead
of sleeping for real.
"""
await asyncio.sleep(seconds)
async def _run_max_duration_timer(
self, seconds: int, call_sid: str, to: str, generation: int
) -> None:
"""Hang up the call and close the socket once the cap elapses.
The belt to Twilio's ``TimeLimit`` suspenders: it fires even if the
Twilio-side limit was misconfigured or silently dropped. Cancelled (and
therefore silent) on ``disconnect()``, at stream end, and on re-arm.
``generation`` is the call generation this watchdog was armed under
(#762 P2). Ending the ORIGINATED ``call_sid`` via REST is always correct
— it targets the expired call by SID. But the OTHER two side effects
target whatever call is live NOW, so both are gated on still owning the
generation:
- The ended-reason stamp (before the REST hang-up): ``"max_duration"``
wins ended-reason ties permanently, so a stale watchdog — one that
fires after a newer call (a b-leg / inbound / no-cap dial that bumps
the generation WITHOUT cancelling this task) has taken over — would
otherwise mislabel that newer live call for the rest of its life.
- The socket close (after the REST hang-up ``await``): a newer
``place_call`` can adopt a fresh, authenticated socket while our
hang-up response is still pending, and closing ``_stream_ws`` would
tear it down.
"""
await self._await_max_duration(seconds)
logger.warning(
"TwilioAgentAdapter: max call duration of %ss reached on call to "
"%s — ending call %s",
seconds,
_redact_e164(to),
call_sid,
)
# Only stamp the ended-reason while we still own the call generation; a
# stale watchdog must not relabel a newer live call.
if generation == self._call_generation:
self._set_stream_ended_reason("max_duration")
rest = self._rest
if rest is not None:
# Blocking REST call off-thread, as send_dtmf does. Targets our own
# SID, so it is correct even once a newer call has taken over.
with suppress(Exception):
await asyncio.to_thread(rest.end_call, call_sid)
# A newer call took over while the hang-up was in flight — its socket is
# not ours to close.
if generation != self._call_generation:
return
ws = self._stream_ws
if ws is not None:
with suppress(Exception):
await ws.close()
def _set_stream_ended_reason(self, reason: TwilioStreamEndedReason) -> None:
"""Record how the media session ended, cap verdict winning ties.
The watchdog closes the socket itself, so the media loop's own "close"
verdict lands moments later and would otherwise mask WHY the call
ended. connect()/disconnect() reset the field directly and so still
clear it for the next session.
"""
if self._stream_ended_reason == "max_duration":
return
self._stream_ended_reason = reason
# ------------------------------------------------------------------ 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>``).
Refused in a-leg mode (#762 AC10) — see ``_assert_dtmf_supported``.
"""
self._assert_dtmf_supported()
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 A_LEG_SEND_DTMF_UNSUPPORTED-
Refusal for
send_dtmfunder a-leg (#762 AC10). Says WHY, because "unsupported" alone reads as an oversight rather than a deliberate guard. 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 StreamAttachMode-
Effective stream-attach mode resolved from
place_call's two parameters. 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."max_duration"is the a-leg duration cap firing (#762 guardrail (b)) — kept distinct from"close"so a scenario author can tell "we hung the call up at the cap" from "the callee hung up".
Classes
class TwilioAgentAdapter (*, account_sid: str, auth_token: str, phone_number: str, public_base_url: Optional[str] = None, allowed_callers: Optional[list[str]] = None, allowed_callees: Optional[list[str]] = None, tunnel_readiness: Optional[TunnelReadiness] = 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, allowed_callees: Optional[list[str]] = None, tunnel_readiness: Optional[TunnelReadiness] = 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 # Destination allowlist for a-leg mode (#762 guardrail (c)). ``None`` # denies every a-leg destination: a-leg dials numbers this account does # NOT own, so an unguarded ``to`` is an unbounded dialer and the # capability has to be opted into per destination. Entries are # normalised (and so validated) here rather than at dial time, so a # typo fails at setup instead of when a PSTN call is about to be # billed. b-leg is deliberately not gated on this — it can only reach # numbers on this account, which is its own guardrail. self.allowed_callees = ( {normalize_e164(n) for n in allowed_callees} if allowed_callees else None ) # Edge-readiness probe consulted before a-leg origination (#762 # guardrail (c)). ``CloudflareTunnel`` satisfies it structurally; # ``TwilioHarness`` wires its own tunnel in. ``None`` means the caller # owns a stable public URL and has nothing to wait for. self.tunnel_readiness = tunnel_readiness 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 # Call generation (#762 P1). Bumped by connect()/place_call()/ # wait_for_call() every time we (re)dial. The media loop captures it at # the moment a socket is ADOPTED and re-checks it on every later side # effect (enqueue, signal-connected, mark-ended, cancel-watchdog, being # ``_stream_ws``): a socket adopted under an older generation — one that # connected and even adopted BEFORE this dial armed the nonce — is stale # the instant the counter moves, and is closed on its next frame instead # of injecting audio into, or ending, the current call. self._call_generation: int = 0 # A-leg WS auth (#762 guardrail (a)). Set by place_call in "a-leg" mode # only; its non-None-ness is what ARMS media-stream nonce enforcement. # In b-leg mode the signed POST /twilio/voice precedes the socket, so # the socket inherits that trust and this stays None — enforcement is # keyed on the mode we originated in, never on whether the inbound frame # happens to carry a nonce (which an attacker could simply omit). self._stream_nonce: Optional[str] = None # Adapter-side max-call-duration watchdog (#762 guardrail (b)). Armed by # place_call in "a-leg" mode; the belt to Twilio's own ``TimeLimit`` # suspenders, which is the half that survives this process dying. self._max_duration_task: Optional[asyncio.Task] = None 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._stream_nonce = None self._cancel_max_duration_timer() # Fresh connection lifecycle — bump the generation so any socket left # over from a previous connect can never pass the media loop's identity # check against this one. self._call_generation += 1 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. # Disarm the duration watchdog first: from here on the call is ours to # end, and a timer that fired mid-teardown would hang up a SID this # adapter may already have replaced with a later call's. self._cancel_max_duration_timer() 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 self._stream_nonce = None self._cancel_max_duration_timer() 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 def _reset_connection_state_for_dial(self) -> None: """Invalidate any socket adopted under a prior call generation (#762 P1). Bumping ``_call_generation`` is what evicts a socket that connected — and even ADOPTED — before this dial armed the nonce: the media loop captured the old generation at adoption and closes itself on its next frame once the counter moves. We also reset the per-dial connection state the loop writes, so this dial starts from a clean slate and, critically, cannot resolve on a stale socket's ``_stream_connected`` signal: - clear ``_stream_connected`` — otherwise a socket that adopted un-gated before arming has already fired it and ``place_call`` would return before any socket authenticates; - drop ``_stream_ws``/``_stream_sid`` — the stale socket must not remain our outbound transport (``send_audio`` writes there); - drain the inbound queue and zero ``_frames_received`` — the stale socket's media must not surface as this call's audio; - clear the terminal flag/reason — the stale session's ``stop`` must not read as this call having ended. Called by both ``place_call`` and ``wait_for_call``. Inbound/b-leg behaviour is unchanged: no socket has adopted yet when the genuine one arrives, so it adopts under the current generation and connects exactly as before. """ self._call_generation += 1 self._stream_ws = None self._stream_sid = None if self._stream_connected is not None: self._stream_connected.clear() if self._inbound_queue is not None: while not self._inbound_queue.empty(): self._inbound_queue.get_nowait() self._stream_ended = False self._stream_ended_reason = "none" self._frames_received = 0 async def place_call( self, to: str, *, timeout: float = 120.0, attach_stream_to_self: Optional[bool] = None, attach_stream: Optional[Literal["a-leg", "b-leg"]] = None, max_call_duration_seconds: Optional[int] = None, ) -> 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 of the default (**b-leg**) mode: ``to`` MUST be a phone number on this same Twilio account. Calling an external PSTN endpoint (a real cell phone) is instead served by **a-leg** mode. Stream-attach mode is selected by ``attach_stream``: - ``"b-leg"`` — today's default: rewrite the callee's ``voice_url``, hold the A-leg open with ``<Say>…<Pause>``, restore on disconnect. Only works for numbers this account owns. - ``"a-leg"`` — originate with inline ``<Connect><Stream>`` so the Media Stream rides OUR own leg. Touches nothing on the callee (no ``resolve_phone_number_sid``/``read_voice_url``/``write_voice_url``), so ``to`` can be any external number. Because that reaches numbers this account does not own, ``to`` must appear in the adapter's ``allowed_callees`` (deny-by-default) and the public base URL must be reachable from the edge — both checked before origination. The legacy ``attach_stream_to_self`` bool is still honoured but superseded by ``attach_stream``: ``True``/unset → ``"b-leg"``, ``False`` → originator-only (a third mode ``attach_stream`` cannot express). Passing both with disagreeing intent raises ``ValueError``. Default timeout 120s covers cloudflared cold-start latency. ``timeout`` and ``max_call_duration_seconds`` are two different clocks and are configured independently: - ``timeout`` bounds how long we WAIT FOR THE MEDIA STREAM TO CONNECT after origination. It says nothing about how long the call may run. - ``max_call_duration_seconds`` bounds HOW LONG THE CALL MAY LIVE (a-leg only, default ``DEFAULT_MAX_CALL_DURATION_SECONDS``, hard cap ``MAX_CALL_DURATION_CAP_SECONDS``; a larger request raises). Under a-leg's ``<Connect>`` the call lives exactly as long as the WebSocket, so without a ceiling a hung executor keeps a billing PSTN call open forever. Two mechanisms enforce it: Twilio's own ``TimeLimit`` on ``Calls.create`` (the load-bearing half — it still fires if this process hangs or is killed) and an adapter-side wall-clock timer that hangs the call up via REST and closes the socket. 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, or — in a-leg mode — is not on ``allowed_callees``. TunnelNotReadyError: If a-leg mode's readiness probe says the public base URL is not reachable from the edge yet. asyncio.TimeoutError: If the media stream doesn't open within ``timeout`` seconds. """ self._assert_connected() self._enter_mode("call") validate_e164(to) # Resolve the effective stream-attach mode BEFORE any REST call so a # conflicting-parameter caller error surfaces before we dial. `attach_stream` # (typed) supersedes the legacy `attach_stream_to_self` bool; the bool's # third mode (originator-only) is unreachable via `attach_stream` by design. mode = _resolve_stream_mode(attach_stream, attach_stream_to_self) # Only a-leg loses <Pause>'s implicit ceiling, so only a-leg carries a # duration cap. Naming one in another mode is rejected rather than # ignored: silently dropping it would leave the caller believing the # call is bounded when it is not. if mode == "a-leg": # Guardrail (c): destinations are deny-by-default in a-leg mode. # Checked here, with the other local caller-fixable failures, so a # misconfigured allowlist costs nothing and dials nothing. self._assert_callee_allowed(to) max_call_duration = resolve_max_call_duration(max_call_duration_seconds) elif max_call_duration_seconds is not None: raise ValueError( "place_call: max_call_duration_seconds is only supported with " 'attach_stream="a-leg"; b-leg and originator-only modes hold the ' "originator leg with <Pause> and are bounded by it." ) else: max_call_duration = None # Guardrail (c), second half: a-leg hands Twilio our public URL and # Twilio opens the media socket against it seconds later. Probe the edge # AFTER the free local checks above and BEFORE origination, so an # unreachable tunnel is a named error instead of a billed call that # dies in a stream-connect timeout. if mode == "a-leg": await self._assert_tunnel_ready() 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: # #762 P1: bump the call generation and reset per-dial connection # state BEFORE originating, so a socket that adopted un-gated before # this dial is evicted and cannot resolve the wait below. self._reset_connection_state_for_dial() if mode == "b-leg": # 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: the Media Stream rides OUR own leg via inline # <Connect><Stream>, so `to` can be any external number and we # touch nothing on the callee. Same TwiML shape the inbound webhook # returns (`_twilio_server.py`), now on origination. # # Guardrail (a): the socket is the only inbound signal on this path, # so it must authenticate itself. Mint a per-call CSPRNG nonce and # ship it as a <Parameter> child; Twilio echoes it back in the # `start` frame's customParameters, and the media loop # (`_twilio_server.py`) closes any socket that cannot present it. # # Assigned UNCONDITIONALLY (mirrors the JS twin): the field also # ARMS nonce enforcement and the a-leg send_dtmf refusal, so a b-leg # dial that left an earlier a-leg call's nonce standing would gate a # socket whose `start` frame carries no <Parameter> at all — the # b-leg call would never connect. nonce = mint_stream_nonce() if mode == "a-leg" else None self._stream_nonce = nonce if nonce is not None: origination_twiml = _build_connect_stream_twiml( stream_ws_url(self.public_base_url), stream_parameters={"nonce": nonce}, ) else: # A-leg (originator-side) TwiML for b-leg / originator-only modes: # 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. origination_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=origination_twiml, time_limit=max_call_duration, ) 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 max_call_duration is not None: self._arm_max_duration_timer(max_call_duration, self._call_sid, to) if mode != "originator-only": # Wait for the media stream to reach us. In b-leg mode it arrives # via the callee's rewritten voice_url; in a-leg mode it rides our # own <Connect><Stream> leg. In originator-only mode there's no # stream coming to us; the callee has its own harness which owns it. _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" ) if mode == "a-leg" and self._call_sid is not None: # The call is already originated and, under a-leg's # <Connect>, lives to the duration cap even though # nothing will ever stream on it. Hang it up before # handing the caller their TimeoutError — otherwise the # ordinary connect-failure path bills for the cap. await self._abandon_originated_call(self._call_sid) 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: # #762 P1: bump the call generation and reset per-dial connection # state, symmetric with place_call. For inbound this only guards a # reconnect/back-to-back dial; a fresh inbound socket still connects # exactly as before. self._reset_connection_state_for_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 _assert_callee_allowed(self, to: str) -> None: """Refuse an a-leg destination that was not explicitly allowlisted. Default-deny: no ``allowed_callees`` means no a-leg call at all. The message names the field to set, because "denied" without "here is how to allow it" just sends the caller reading source. Comparison is an exact set lookup on the normalised number, so a near-miss — a prefix, a suffix, an extra space — is a refusal. """ if self.allowed_callees is None: raise ValueError( 'place_call: attach_stream="a-leg" requires allowed_callees. ' "a-leg dials numbers this Twilio account does not own, so " "destinations are deny-by-default: pass " "TwilioAgentAdapter(allowed_callees=[...]) listing every " f"number this adapter may dial, including {_redact_e164(to)}." ) if normalize_e164(to) not in self.allowed_callees: raise ValueError( f"place_call: destination {_redact_e164(to)} is not in " "allowed_callees. Add it to " "TwilioAgentAdapter(allowed_callees=[...]) to permit this " "destination." ) def _assert_dtmf_supported(self) -> None: """Refuse DTMF on a call whose media stream rides our own leg (#762 AC10). ``send_dtmf`` works by ``calls(sid).update(twiml=...)``, which REPLACES the TwiML the call is executing. Under b-leg that TwiML is a ``<Say>`` + ``<Pause>`` hold on our leg, so replacing it costs nothing — the Media Stream lives on the callee's leg. Under a-leg the TwiML being replaced IS the ``<Connect><Stream>`` verb carrying the scenario, so the same REST call would redirect the call away from the socket and end the media session mid-run. Armed off ``_stream_nonce`` — set only by an a-leg ``place_call``, the same signal the media loop uses to arm nonce enforcement — rather than re-derived from the caller's arguments here. """ if self._stream_nonce is not None: raise RuntimeError(A_LEG_SEND_DTMF_UNSUPPORTED) async def _assert_tunnel_ready(self) -> None: """Refuse to originate until the public URL is reachable from the edge. Delegates to whatever readiness probe was supplied — ``CloudflareTunnel`` already knows how to answer this. No probe means the caller owns a stable public URL and there is nothing to wait for. """ if self.tunnel_readiness is None: return try: await self.tunnel_readiness.wait_until_edge_reachable() except TunnelNotReadyError: raise except Exception as exc: raise TunnelNotReadyError( f"place_call: public_base_url {self.public_base_url!r} is not " "reachable from the edge yet, so Twilio's media stream would " "connect to nothing. Not originating." ) from exc 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 # ------------------------------------------------------- max call duration def _arm_max_duration_timer(self, seconds: int, call_sid: str, to: str) -> None: """Start the wall-clock watchdog that ends ``call_sid`` at ``seconds``. Any previously-armed timer is cancelled first, so a second ``place_call`` on the same adapter can never leave an older call's timer alive to hang up the newer call's SID. """ self._cancel_max_duration_timer() # Capture the generation this watchdog is armed under (#762 P2). After # the hang-up REST round-trip the handler re-checks it before closing the # socket, so an expiry whose REST response resolves AFTER a newer call has # taken over cannot close that newer call's authenticated socket. generation = self._call_generation self._max_duration_task = asyncio.create_task( self._run_max_duration_timer(seconds, call_sid, to, generation) ) async def _abandon_originated_call(self, call_sid: str) -> None: """Disarm the watchdog and hang up a call nothing will ever stream on. Best-effort on the REST hangup: the caller is already raising, and Twilio's own ``TimeLimit`` remains the backstop if this leg of the teardown fails. """ self._cancel_max_duration_timer() rest = self._rest if rest is not None: # Blocking REST call off-thread, as send_dtmf does. with suppress(Exception): await asyncio.to_thread(rest.end_call, call_sid) def _cancel_max_duration_timer(self) -> None: """Disarm the watchdog. Idempotent; safe when none was ever armed.""" task = self._max_duration_task self._max_duration_task = None if task is not None and not task.done(): task.cancel() async def _await_max_duration(self, seconds: int) -> None: """Sleep until the duration cap elapses. Test seam (same role as ``_build_app``/``_media_stream_loop``): tests replace this on the instance to drive expiry on controlled time instead of sleeping for real. """ await asyncio.sleep(seconds) async def _run_max_duration_timer( self, seconds: int, call_sid: str, to: str, generation: int ) -> None: """Hang up the call and close the socket once the cap elapses. The belt to Twilio's ``TimeLimit`` suspenders: it fires even if the Twilio-side limit was misconfigured or silently dropped. Cancelled (and therefore silent) on ``disconnect()``, at stream end, and on re-arm. ``generation`` is the call generation this watchdog was armed under (#762 P2). Ending the ORIGINATED ``call_sid`` via REST is always correct — it targets the expired call by SID. But the OTHER two side effects target whatever call is live NOW, so both are gated on still owning the generation: - The ended-reason stamp (before the REST hang-up): ``"max_duration"`` wins ended-reason ties permanently, so a stale watchdog — one that fires after a newer call (a b-leg / inbound / no-cap dial that bumps the generation WITHOUT cancelling this task) has taken over — would otherwise mislabel that newer live call for the rest of its life. - The socket close (after the REST hang-up ``await``): a newer ``place_call`` can adopt a fresh, authenticated socket while our hang-up response is still pending, and closing ``_stream_ws`` would tear it down. """ await self._await_max_duration(seconds) logger.warning( "TwilioAgentAdapter: max call duration of %ss reached on call to " "%s — ending call %s", seconds, _redact_e164(to), call_sid, ) # Only stamp the ended-reason while we still own the call generation; a # stale watchdog must not relabel a newer live call. if generation == self._call_generation: self._set_stream_ended_reason("max_duration") rest = self._rest if rest is not None: # Blocking REST call off-thread, as send_dtmf does. Targets our own # SID, so it is correct even once a newer call has taken over. with suppress(Exception): await asyncio.to_thread(rest.end_call, call_sid) # A newer call took over while the hang-up was in flight — its socket is # not ours to close. if generation != self._call_generation: return ws = self._stream_ws if ws is not None: with suppress(Exception): await ws.close() def _set_stream_ended_reason(self, reason: TwilioStreamEndedReason) -> None: """Record how the media session ended, cap verdict winning ties. The watchdog closes the socket itself, so the media loop's own "close" verdict lands moments later and would otherwise mask WHY the call ended. connect()/disconnect() reset the field directly and so still clear it for the next session. """ if self._stream_ended_reason == "max_duration": return self._stream_ended_reason = reason # ------------------------------------------------------------------ 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>``). Refused in a-leg mode (#762 AC10) — see ``_assert_dtmf_supported``. """ self._assert_dtmf_supported() 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._stream_nonce = None self._cancel_max_duration_timer() # Fresh connection lifecycle — bump the generation so any socket left # over from a previous connect can never pass the media loop's identity # check against this one. self._call_generation += 1 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. # Disarm the duration watchdog first: from here on the call is ours to # end, and a timer that fired mid-teardown would hang up a SID this # adapter may already have replaced with a later call's. self._cancel_max_duration_timer() 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 self._stream_nonce = None self._cancel_max_duration_timer() 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: Optional[bool] = None, attach_stream: "Optional[Literal['a-leg', 'b-leg']]" = None, max_call_duration_seconds: Optional[int] = None) ‑> 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 of the default (b-leg) mode:
toMUST be a phone number on this same Twilio account. Calling an external PSTN endpoint (a real cell phone) is instead served by a-leg mode.Stream-attach mode is selected by
attach_stream:"b-leg"— today's default: rewrite the callee'svoice_url, hold the A-leg open with<Say>…<Pause>, restore on disconnect. Only works for numbers this account owns."a-leg"— originate with inline<Connect><Stream>so the Media Stream rides OUR own leg. Touches nothing on the callee (noresolve_phone_number_sid/read_voice_url/write_voice_url), sotocan be any external number. Because that reaches numbers this account does not own,tomust appear in the adapter'sallowed_callees(deny-by-default) and the public base URL must be reachable from the edge — both checked before origination.
The legacy
attach_stream_to_selfbool is still honoured but superseded byattach_stream:True/unset →"b-leg",False→ originator-only (a third modeattach_streamcannot express). Passing both with disagreeing intent raisesValueError.Default timeout 120s covers cloudflared cold-start latency.
timeoutandmax_call_duration_secondsare two different clocks and are configured independently:timeoutbounds how long we WAIT FOR THE MEDIA STREAM TO CONNECT after origination. It says nothing about how long the call may run.max_call_duration_secondsbounds HOW LONG THE CALL MAY LIVE (a-leg only, defaultDEFAULT_MAX_CALL_DURATION_SECONDS, hard capMAX_CALL_DURATION_CAP_SECONDS; a larger request raises). Under a-leg's<Connect>the call lives exactly as long as the WebSocket, so without a ceiling a hung executor keeps a billing PSTN call open forever. Two mechanisms enforce it: Twilio's ownTimeLimitonCalls.create(the load-bearing half — it still fires if this process hangs or is killed) and an adapter-side wall-clock timer that hangs the call up via REST and closes the socket.
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, or — in a-leg mode — is not onallowed_callees. TunnelNotReadyError- If a-leg mode's readiness probe says the public base URL is not reachable from the edge yet.
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: Optional[bool] = None, attach_stream: Optional[Literal["a-leg", "b-leg"]] = None, max_call_duration_seconds: Optional[int] = None, ) -> 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 of the default (**b-leg**) mode: ``to`` MUST be a phone number on this same Twilio account. Calling an external PSTN endpoint (a real cell phone) is instead served by **a-leg** mode. Stream-attach mode is selected by ``attach_stream``: - ``"b-leg"`` — today's default: rewrite the callee's ``voice_url``, hold the A-leg open with ``<Say>…<Pause>``, restore on disconnect. Only works for numbers this account owns. - ``"a-leg"`` — originate with inline ``<Connect><Stream>`` so the Media Stream rides OUR own leg. Touches nothing on the callee (no ``resolve_phone_number_sid``/``read_voice_url``/``write_voice_url``), so ``to`` can be any external number. Because that reaches numbers this account does not own, ``to`` must appear in the adapter's ``allowed_callees`` (deny-by-default) and the public base URL must be reachable from the edge — both checked before origination. The legacy ``attach_stream_to_self`` bool is still honoured but superseded by ``attach_stream``: ``True``/unset → ``"b-leg"``, ``False`` → originator-only (a third mode ``attach_stream`` cannot express). Passing both with disagreeing intent raises ``ValueError``. Default timeout 120s covers cloudflared cold-start latency. ``timeout`` and ``max_call_duration_seconds`` are two different clocks and are configured independently: - ``timeout`` bounds how long we WAIT FOR THE MEDIA STREAM TO CONNECT after origination. It says nothing about how long the call may run. - ``max_call_duration_seconds`` bounds HOW LONG THE CALL MAY LIVE (a-leg only, default ``DEFAULT_MAX_CALL_DURATION_SECONDS``, hard cap ``MAX_CALL_DURATION_CAP_SECONDS``; a larger request raises). Under a-leg's ``<Connect>`` the call lives exactly as long as the WebSocket, so without a ceiling a hung executor keeps a billing PSTN call open forever. Two mechanisms enforce it: Twilio's own ``TimeLimit`` on ``Calls.create`` (the load-bearing half — it still fires if this process hangs or is killed) and an adapter-side wall-clock timer that hangs the call up via REST and closes the socket. 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, or — in a-leg mode — is not on ``allowed_callees``. TunnelNotReadyError: If a-leg mode's readiness probe says the public base URL is not reachable from the edge yet. asyncio.TimeoutError: If the media stream doesn't open within ``timeout`` seconds. """ self._assert_connected() self._enter_mode("call") validate_e164(to) # Resolve the effective stream-attach mode BEFORE any REST call so a # conflicting-parameter caller error surfaces before we dial. `attach_stream` # (typed) supersedes the legacy `attach_stream_to_self` bool; the bool's # third mode (originator-only) is unreachable via `attach_stream` by design. mode = _resolve_stream_mode(attach_stream, attach_stream_to_self) # Only a-leg loses <Pause>'s implicit ceiling, so only a-leg carries a # duration cap. Naming one in another mode is rejected rather than # ignored: silently dropping it would leave the caller believing the # call is bounded when it is not. if mode == "a-leg": # Guardrail (c): destinations are deny-by-default in a-leg mode. # Checked here, with the other local caller-fixable failures, so a # misconfigured allowlist costs nothing and dials nothing. self._assert_callee_allowed(to) max_call_duration = resolve_max_call_duration(max_call_duration_seconds) elif max_call_duration_seconds is not None: raise ValueError( "place_call: max_call_duration_seconds is only supported with " 'attach_stream="a-leg"; b-leg and originator-only modes hold the ' "originator leg with <Pause> and are bounded by it." ) else: max_call_duration = None # Guardrail (c), second half: a-leg hands Twilio our public URL and # Twilio opens the media socket against it seconds later. Probe the edge # AFTER the free local checks above and BEFORE origination, so an # unreachable tunnel is a named error instead of a billed call that # dies in a stream-connect timeout. if mode == "a-leg": await self._assert_tunnel_ready() 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: # #762 P1: bump the call generation and reset per-dial connection # state BEFORE originating, so a socket that adopted un-gated before # this dial is evicted and cannot resolve the wait below. self._reset_connection_state_for_dial() if mode == "b-leg": # 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: the Media Stream rides OUR own leg via inline # <Connect><Stream>, so `to` can be any external number and we # touch nothing on the callee. Same TwiML shape the inbound webhook # returns (`_twilio_server.py`), now on origination. # # Guardrail (a): the socket is the only inbound signal on this path, # so it must authenticate itself. Mint a per-call CSPRNG nonce and # ship it as a <Parameter> child; Twilio echoes it back in the # `start` frame's customParameters, and the media loop # (`_twilio_server.py`) closes any socket that cannot present it. # # Assigned UNCONDITIONALLY (mirrors the JS twin): the field also # ARMS nonce enforcement and the a-leg send_dtmf refusal, so a b-leg # dial that left an earlier a-leg call's nonce standing would gate a # socket whose `start` frame carries no <Parameter> at all — the # b-leg call would never connect. nonce = mint_stream_nonce() if mode == "a-leg" else None self._stream_nonce = nonce if nonce is not None: origination_twiml = _build_connect_stream_twiml( stream_ws_url(self.public_base_url), stream_parameters={"nonce": nonce}, ) else: # A-leg (originator-side) TwiML for b-leg / originator-only modes: # 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. origination_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=origination_twiml, time_limit=max_call_duration, ) 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 max_call_duration is not None: self._arm_max_duration_timer(max_call_duration, self._call_sid, to) if mode != "originator-only": # Wait for the media stream to reach us. In b-leg mode it arrives # via the callee's rewritten voice_url; in a-leg mode it rides our # own <Connect><Stream> leg. In originator-only mode there's no # stream coming to us; the callee has its own harness which owns it. _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" ) if mode == "a-leg" and self._call_sid is not None: # The call is already originated and, under a-leg's # <Connect>, lives to the duration cap even though # nothing will ever stream on it. Hang it up before # handing the caller their TimeoutError — otherwise the # ordinary connect-failure path bills for the cap. await self._abandon_originated_call(self._call_sid) 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>).Refused in a-leg mode (#762 AC10) — see
_assert_dtmf_supported.Expand source code
async def send_dtmf(self, tones: str) -> None: """Send DTMF digits on the live call (uses Twilio REST ``<Play digits>``). Refused in a-leg mode (#762 AC10) — see ``_assert_dtmf_supported``. """ self._assert_dtmf_supported() 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: # #762 P1: bump the call generation and reset per-dial connection # state, symmetric with place_call. For inbound this only guards a # reconnect/back-to-back dial; a fresh inbound socket still connects # exactly as before. self._reset_connection_state_for_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