feat(voice): per-call TwiML overrides for custom hosts (host_twiml_options + websocket_url)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "twilio-agent-connect"
-version = "2.0.0"
+version = "2.1.0"
 description = "A Python framework for building agentic applications with Twilio"
 authors = [{ name = "Twilio Conversation AI", email = "team_conversational-ai@twilio.com" }]
 readme = "README.md"

src/tac/channels/voice/channel.py

@@ -156,6 +156,8 @@ def _get_twilio_client(self) -> Client:
     async def handle_incoming_call(
         self,
         twiml_request: TwiMLRequest | None = None,
+        *,
+        host_twiml_options: TwiMLOptions | None = None,
     ) -> str:
         """
         Generate TwiML response for incoming voice calls.
@@ -170,44 +172,75 @@ async def handle_incoming_call(
         TwiML fields are merged per-field, highest precedence first:
           1. Output of the customizer registered via
              ``VoiceChannel.on_inbound_call_twiml(...)`` if configured
-             and ``twiml_request`` is given.
+             and ``twiml_request`` is given. (Application-owned.)
           2. ``VoiceChannelConfig.default_twiml_options`` — per-channel defaults.
-          3. TAC defaults: a fixed default ``welcome_greeting``,
-             ``conversation_configuration`` from ``TACConfig``, and
+          3. ``host_twiml_options`` — per-call transport facts supplied by the
+             host (the code owning the route), e.g. a per-call ``websocket_url``
+             with an affinity token.
+          4. TAC defaults: a fixed default ``welcome_greeting``,
+             ``conversation_configuration`` from ``TACConfig``,
              ``action_url`` resolved via Studio handoff (when
              ``studio_handoff_flow_sid`` is configured), else derived from
-             ``TACConfig.voice_public_domain`` + ``voice_action_path``.
+             ``TACConfig.voice_public_domain`` + ``voice_action_path``, and the
+             ``websocket_url`` derived from ``TACConfig.voice_public_domain`` +
+             ``voice_websocket_path``.
 
         Fields not set at a layer fall through to lower layers. Lists
         (``languages``) and nested models (``custom_parameters``) replace
-        wholesale when set at a higher-priority layer.
+        wholesale when set at a higher-priority layer. ``websocket_url`` falls
+        back to the ``TACConfig``-derived URL if unset at every layer.
+
+        The two arguments are complementary, not alternatives — a custom host
+        typically passes both on the same call: ``twiml_request`` carries the
+        inbound call's data (so the application's customizer can run), and
+        ``host_twiml_options`` carries the host's own per-call overrides.
 
         Args:
-            twiml_request: Parsed Twilio webhook fields. Passed to the
-                customizer if one is configured on the channel.
+            twiml_request: The incoming Twilio voice webhook, parsed into a
+                framework-neutral form (From, To, CallSid, CallerCountry, …).
+                Supplied by Twilio; forwarded to the ``on_inbound_call_twiml``
+                customizer so the application can produce per-call overrides.
+            host_twiml_options: Per-call TwiML overrides supplied by the *host*
+                (the code owning the route — e.g. a custom server), for
+                transport facts the SDK can't derive, such as a per-call
+                ``websocket_url`` with an affinity token. Layered below
+                ``default_twiml_options`` and the application customizer, so a
+                developer's explicit settings still win.
 
         Returns:
             TwiML XML string for call connection.
         """
-        websocket_url = self._resolve_websocket_url("handle_incoming_call")
-
         customized: TwiMLOptions | None = None
         if self._on_inbound_call_twiml is not None and twiml_request is not None:
             customized = await self._on_inbound_call_twiml(twiml_request)
 
-        merged = self._build_twiml_options(customized)
+        merged = self._build_twiml_options(host_twiml_options, customized)
+        # merged.websocket_url is either a validated non-empty URL (set by some
+        # layer) or None; fall back to the TACConfig-derived URL only when None.
+        websocket_url = (
+            merged.websocket_url
+            if merged.websocket_url is not None
+            else self._resolve_websocket_url("handle_incoming_call")
+        )
         return twiml.generate_twiml(websocket_url, merged)
 
-    def _build_twiml_options(self, per_call: TwiMLOptions | None) -> TwiMLOptions:
-        """Layer TwiML options: TAC defaults → channel ``default_twiml_options``
-        → ``per_call`` (customizer output for inbound, or
+    def _build_twiml_options(
+        self,
+        host: TwiMLOptions | None,
+        per_call: TwiMLOptions | None,
+    ) -> TwiMLOptions:
+        """Layer TwiML options, lowest precedence first: TAC defaults →
+        ``host`` (calling host's per-call values) → ``default_twiml_options`` →
+        ``per_call`` (application customizer output for inbound, or
         ``InitiateVoiceConversationOptions.twiml_options`` for outbound).
         """
         merged = TwiMLOptions(
             welcome_greeting=DEFAULT_WELCOME_GREETING,
             conversation_configuration=self.tac.config.conversation_configuration_id,
-            action_url=self._resolve_action_url(per_call),
+            action_url=self._resolve_action_url(host, per_call),
         )
+        if host is not None:
+            self._overlay_fields(merged, host)
         if self.config.default_twiml_options is not None:
             self._overlay_fields(merged, self.config.default_twiml_options)
         if per_call is not None:
@@ -234,14 +267,19 @@ def _overlay_fields(target: TwiMLOptions, source: TwiMLOptions) -> None:
                 continue
             setattr(target, field, getattr(source, field))
 
-    def _resolve_action_url(self, customized: TwiMLOptions | None) -> str | None:
+    def _resolve_action_url(
+        self,
+        host: TwiMLOptions | None,
+        customized: TwiMLOptions | None,
+    ) -> str | None:
         """Resolve the TwiML ``<Connect action=...>`` URL.
 
         Precedence (highest to lowest):
-          1. customizer
+          1. application customizer
           2. channel ``default_twiml_options``
-          3. Studio handoff (when ``studio_handoff_flow_sid`` is configured)
-          4. Channel default — derived from ``TACConfig.voice_public_domain``
+          3. ``host`` (calling host's per-call options)
+          4. Studio handoff (when ``studio_handoff_flow_sid`` is configured)
+          5. Channel default — derived from ``TACConfig.voice_public_domain``
              + ``TACConfig.voice_action_path``.
 
         User-expressed intent (Studio handoff is configured explicitly on
@@ -263,6 +301,8 @@ def _resolve_action_url(self, customized: TwiMLOptions | None) -> str | None:
             and "action_url" in self.config.default_twiml_options.model_fields_set
         ):
             return self.config.default_twiml_options.action_url
+        if host is not None and "action_url" in host.model_fields_set:
+            return host.action_url
         if self.tac.config.studio_handoff_flow_sid:
             return studio_voice_handoff_url(
                 self.tac.config.account_sid,
@@ -515,9 +555,6 @@ async def initiate_outbound_conversation(
         ``TACConfig.voice_websocket_path``, unless overridden per-call via
         ``options.websocket_url``.
         """
-        websocket_url = options.websocket_url or self._resolve_websocket_url(
-            "initiate_outbound_conversation"
-        )
         from_number = self.tac.config.phone_number
 
         self.logger.info(
@@ -526,10 +563,19 @@ async def initiate_outbound_conversation(
             from_number=mask_phone(from_number),
         )
 
-        # Same layering as handle_incoming_call, minus the customizer
-        # (customizers receive a TwiMLRequest from an inbound webhook; there
-        # is no equivalent for outbound).
-        merged = self._build_twiml_options(options.twiml_options)
+        # Outbound has no inbound customizer and no server layer; the per-call
+        # override is options.twiml_options.
+        merged = self._build_twiml_options(None, options.twiml_options)
+
+        # ``options.websocket_url`` is the dedicated per-call outbound override
+        # and wins over any websocket_url that came through the layered
+        # ``twiml_options`` merge; both fall back to the TACConfig-derived URL.
+        if options.websocket_url is not None:
+            websocket_url = options.websocket_url
+        elif merged.websocket_url is not None:
+            websocket_url = merged.websocket_url
+        else:
+            websocket_url = self._resolve_websocket_url("initiate_outbound_conversation")
 
         try:
             twiml_xml = twiml.generate_twiml(websocket_url, merged)

src/tac/channels/voice/config.py

@@ -21,7 +21,8 @@ class VoiceChannelConfig(BaseModel):
         1. Output of the customizer registered via
            ``VoiceChannel.on_inbound_call_twiml(...)`` [optional]
         2. ``default_twiml_options``                     [optional]
-        3. TAC defaults
+        3. ``handle_incoming_call(host_twiml_options=...)`` [optional]
+        4. TAC defaults
 
       Outbound calls (``initiate_outbound_conversation``):
         1. ``InitiateVoiceConversationOptions.twiml_options`` [optional]

src/tac/channels/voice/twiml.py

@@ -42,9 +42,13 @@
 )
 
 # Fields on TwiMLOptions that this module handles specially (not via the
-# generic _OPTIONAL_RELAY_ATTRS loop) — the action_url, the <Language>
-# children list, the <Parameter> children dict, and the extra escape hatch.
+# generic _OPTIONAL_RELAY_ATTRS loop) — the websocket_url (emitted as the
+# ``<ConversationRelay url=...>`` attribute, resolved from the positional
+# ``websocket_url`` arg or ``options.websocket_url``), the action_url, the
+# <Language> children list, the <Parameter> children dict, and the extra
+# escape hatch.
 _HANDLED_OUTSIDE_LOOP = {
+    "websocket_url",
     "action_url",
     "languages",
     "custom_parameters",
@@ -78,7 +82,7 @@ def _verify_attrs_in_sync() -> None:
 
 
 def generate_twiml(
-    websocket_url: str,
+    websocket_url: str | None = None,
     options: TwiMLOptions | dict[str, Any] | None = None,
 ) -> str:
     """
@@ -89,16 +93,24 @@ def generate_twiml(
     static ``twiml_options`` from ``VoiceChannelConfig``, and any per-call
     customizer output.
 
+    The WebSocket URL may be passed positionally or as ``options.websocket_url``
+    (positional wins when both are given), so a caller can pass everything in one
+    object: ``generate_twiml(options=TwiMLOptions(websocket_url=...))``.
+
     Args:
         websocket_url: Public WebSocket URL for ConversationRelay
-            (e.g. ``'wss://example.ngrok.app/ws'``).
+            (e.g. ``'wss://example.ngrok.app/ws'``). Optional if
+            ``options.websocket_url`` is set.
         options: Optional ``TwiMLOptions`` (or dict). See ``TwiMLOptions``
             for supported fields. Newly-added ConversationRelay attributes
             not yet typed on the model can be passed via ``extra``.
 
     Returns:
         TwiML XML string ready to return to Twilio.
 
+    Raises:
+        ValueError: If no WebSocket URL is provided via either source.
+
     Example:
         >>> twiml = generate_twiml(
         ...     "wss://example.com/voice",
@@ -113,6 +125,15 @@ def generate_twiml(
     elif isinstance(options, dict):
         options = TwiMLOptions(**options)
 
+    # Positional arg wins when both are set; fall back to options.websocket_url.
+    # (TwiMLOptions rejects an empty websocket_url, so any value here is real.)
+    resolved_websocket_url = websocket_url if websocket_url is not None else options.websocket_url
+    if not resolved_websocket_url:
+        raise ValueError(
+            "generate_twiml requires a WebSocket URL — pass it positionally or "
+            "set options.websocket_url."
+        )
+
     response = VoiceResponse()
 
     # Create Connect verb with optional action
@@ -123,7 +144,7 @@ def generate_twiml(
 
     # Build ConversationRelay kwargs. The twilio SDK converts snake_case to
     # camelCase automatically, and serializes bool/str as TwiML attribute values.
-    relay_kwargs: dict[str, Any] = {"url": websocket_url}
+    relay_kwargs: dict[str, Any] = {"url": resolved_websocket_url}
     for attr in _OPTIONAL_RELAY_ATTRS:
         value = getattr(options, attr)
         if value is None:

src/tac/models/voice.py

@@ -2,7 +2,7 @@
 
 from typing import Any, Literal
 
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 # Twilio uses the same four-value enum for several attributes that control
 # what caller input (DTMF, speech, both, neither) triggers a given behavior.
@@ -193,6 +193,16 @@ class TwiMLOptions(BaseModel):
         description="Conversation Service SID for ConversationRelay to automatically "
         "manage conversation creation and participants.",
     )
+    websocket_url: str | None = Field(
+        None,
+        description="ConversationRelay WebSocket URL (the <ConversationRelay url=...> "
+        "attribute). Leave None (the default) to use the URL the channel derives "
+        "from TACConfig.voice_public_domain + voice_websocket_path. Set it only for a "
+        "per-call URL — e.g. an affinity-routed host that appends a token to the "
+        "upgrade URL, typically via handle_incoming_call's host_twiml_options. Like "
+        "every other field, it layers: on_inbound_call_twiml customizer > "
+        "default_twiml_options > host_twiml_options > TAC default.",
+    )
 
     # Language, TTS, STT
     language: str | None = Field(
@@ -319,6 +329,20 @@ class TwiMLOptions(BaseModel):
 
     model_config = {"populate_by_name": True}
 
+    @field_validator("websocket_url")
+    @classmethod
+    def _reject_empty_websocket_url(cls, v: str | None) -> str | None:
+        """Reject an empty/whitespace-only ``websocket_url``. Leave it ``None``
+        (the default) to fall through to the derived URL; an empty string is a
+        misconfiguration that would otherwise emit ``<ConversationRelay url="">``.
+        """
+        if v is not None and not v.strip():
+            raise ValueError(
+                "websocket_url cannot be empty — omit it (None) to use the "
+                "derived URL, or provide a non-empty URL."
+            )
+        return v
+
     @model_validator(mode="after")
     def _reject_extra_shadowing_typed_fields(self) -> "TwiMLOptions":
         """Fail fast when ``extra`` includes a key that has a typed field on