diff --git a/getting_started/examples/features/outbound.py b/getting_started/examples/features/outbound.py index a693249..12ca8c3 100644 --- a/getting_started/examples/features/outbound.py +++ b/getting_started/examples/features/outbound.py @@ -41,8 +41,8 @@ ) from tac.models.session import ConversationSession from tac.models.tac import TACMemoryResponse +from tac.models.voice import TwiMLOptions from tac.server import TACFastAPIServer -from tac.server.config import TACServerConfig load_dotenv() set_tracing_disabled(True) @@ -91,7 +91,17 @@ def parse_args() -> argparse.Namespace: if tac.config.whatsapp_number else None ) -voice_channel = VoiceChannel(tac, config=VoiceChannelConfig(memory_mode="always")) +voice_channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + memory_mode="once", + # Channel-wide TwiML applied to every call (inbound + outbound). + default_twiml_options=TwiMLOptions( + voice="en-US-Journey-D", + welcome_greeting="Hi! How can I help?", + ), + ), +) async def handle_message_ready( @@ -162,18 +172,11 @@ async def initiate_outbound(args: argparse.Namespace) -> None: print("\nWaiting for replies... (Ctrl+C to exit)\n") elif args.channel == "voice": - server_config = TACServerConfig.from_env() - public_domain = server_config.public_domain - if not public_domain: - print("TWILIO_VOICE_PUBLIC_DOMAIN is required for voice calls.") - sys.exit(1) - voice_result = await voice_channel.initiate_outbound_conversation( InitiateVoiceConversationOptions( to=args.to, - websocket_url=f"wss://{public_domain}/ws", - welcome_greeting=args.welcome_greeting, - action_url=f"https://{public_domain}/conversation-relay-callback", + # Per-call TwiML overrides for this outbound call. Overrides channel defaults + twiml_options=TwiMLOptions(welcome_greeting=args.welcome_greeting), ) ) print(f"Call placed to {args.to} (SID: {voice_result.call_sid})") diff --git a/getting_started/examples/features/voice_twiml_customization.py b/getting_started/examples/features/voice_twiml_customization.py new file mode 100644 index 0000000..5194ae6 --- /dev/null +++ b/getting_started/examples/features/voice_twiml_customization.py @@ -0,0 +1,82 @@ +""" +Feature: ConversationRelay TwiML customization + +Two layers (highest precedence first): + +1. Per-call customizer — register on the channel via + ``voice_channel.on_inbound_call_twiml(...)``. Async callable that + receives a TwiMLRequest (parsed Twilio webhook fields: From, To, + CallerCountry, …). Inbound only. For outbound, pass per-call + TwiMLOptions on InitiateVoiceConversationOptions. +2. ``VoiceChannelConfig.default_twiml_options`` — static TwiMLOptions + applied to every call (inbound and outbound). + +Layers merge per-field: the customizer overrides only the fields it +explicitly sets; everything else falls through to ``default_twiml_options`` +and then to TAC defaults (websocket URL, action URL, conversation_configuration). + +This example shows both layers together — channel-wide defaults plus a +country-based customizer that overrides language/voice/greeting for +specific callers. +""" + +from dotenv import load_dotenv + +from tac import TAC, TACConfig +from tac.channels.voice import VoiceChannel, VoiceChannelConfig +from tac.models.session import ConversationSession +from tac.models.tac import TACMemoryResponse +from tac.models.voice import TwiMLOptions, TwiMLRequest +from tac.server import TACFastAPIServer + +load_dotenv() + +tac = TAC(config=TACConfig.from_env()) + + +async def handle_message_ready( + user_message: str, + context: ConversationSession, + memory_response: TACMemoryResponse | None, +) -> str: + return f"You said: {user_message}" + + +tac.on_message_ready(handle_message_ready) + + +async def customize_twiml(req: TwiMLRequest) -> TwiMLOptions: + """Per-call overrides for inbound calls. Only the fields you set here + override the channel default; the rest fall through.""" + if req.caller_country == "MX": + return TwiMLOptions( + language="es-MX", + welcome_greeting="¡Hola! ¿En qué puedo ayudarte?", + ) + if req.caller_country == "FR": + return TwiMLOptions( + language="fr-FR", + welcome_greeting="Bonjour ! Comment puis-je vous aider ?", + ) + return TwiMLOptions() # fall through to default_twiml_options + + +voice_channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + # Channel-wide defaults — apply to every call (inbound + outbound). + default_twiml_options=TwiMLOptions( + welcome_greeting="Hello! This is a default greeting.", + interruptible="speech", + # Escape hatch for ConversationRelay attributes TAC hasn't typed yet: + # extra={"newRelayAttribute": "value"}, + ), + ), +) +# Register the per-call inbound customizer. +voice_channel.on_inbound_call_twiml(customize_twiml) + + +if __name__ == "__main__": + server = TACFastAPIServer(tac=tac, voice_channel=voice_channel) + server.start() diff --git a/src/tac/channels/voice/__init__.py b/src/tac/channels/voice/__init__.py index 5d26eb0..3df4f1d 100644 --- a/src/tac/channels/voice/__init__.py +++ b/src/tac/channels/voice/__init__.py @@ -1,7 +1,22 @@ """Voice channel for handling voice-based conversations.""" from tac.channels.voice.channel import VoiceChannel -from tac.channels.voice.config import VoiceChannelConfig +from tac.channels.voice.config import InboundCallTwiMLHandler, VoiceChannelConfig from tac.channels.voice.twiml import generate_twiml +from tac.models.voice import ( + InterruptMode, + LanguageConfig, + TwiMLOptions, + TwiMLRequest, +) -__all__ = ["VoiceChannel", "VoiceChannelConfig", "generate_twiml"] +__all__ = [ + "InboundCallTwiMLHandler", + "InterruptMode", + "LanguageConfig", + "TwiMLOptions", + "TwiMLRequest", + "VoiceChannel", + "VoiceChannelConfig", + "generate_twiml", +] diff --git a/src/tac/channels/voice/channel.py b/src/tac/channels/voice/channel.py index f0160d1..dec5f72 100644 --- a/src/tac/channels/voice/channel.py +++ b/src/tac/channels/voice/channel.py @@ -22,16 +22,20 @@ PromptMessage, SetupMessage, TwiMLOptions, + TwiMLRequest, ) from tac.session import SessionState +from tac.tools.handoff import studio_voice_handoff_url from tac.utils.redaction import mask_phone from . import twiml -from .config import VoiceChannelConfig +from .config import InboundCallTwiMLHandler, VoiceChannelConfig _POLL_ATTEMPTS = 5 _POLL_BASE_DELAY = 0.25 +DEFAULT_WELCOME_GREETING = "Hello! How can I assist you today?" + class VoiceChannel(BaseChannel): """ @@ -73,10 +77,64 @@ def __init__( config = VoiceChannelConfig() super().__init__(tac, memory_mode=config.memory_mode) + self.config = config self.session_manager = config.session_manager + self._on_inbound_call_twiml: InboundCallTwiMLHandler | None = None self._websocket_manager = WebSocketManager() self._twilio_client: Client | None = None + def on_inbound_call_twiml(self, callback: InboundCallTwiMLHandler) -> None: + """Register a callback that produces per-call overrides for the + TwiML inside ```` on inbound calls. + + The callback receives a framework-neutral ``TwiMLRequest`` (parsed + from the Twilio webhook form) and returns a ``TwiMLOptions``. Fields + the callback explicitly sets override ``default_twiml_options`` and + TAC defaults; unset fields fall through. + + Example: + ```python + async def by_country(req: TwiMLRequest) -> TwiMLOptions: + if req.caller_country == "MX": + return TwiMLOptions(language="es-MX", welcome_greeting="¡Hola!") + return TwiMLOptions() + + + voice_channel.on_inbound_call_twiml(by_country) + ``` + + Outbound calls don't use this — pass per-call TwiML via + ``InitiateVoiceConversationOptions.twiml_options`` directly. + """ + self._on_inbound_call_twiml = callback + + def _resolve_websocket_url(self, action: str) -> str: + """Resolve the public WebSocket URL from + ``TACConfig.voice_public_domain`` + ``TACConfig.voice_websocket_path``. + Raises if ``voice_public_domain`` isn't set. + """ + if self.tac.config.voice_public_domain: + return ( + f"wss://{self.tac.config.voice_public_domain}{self.tac.config.voice_websocket_path}" + ) + raise ValueError( + f"{action} needs a WebSocket URL. Set TWILIO_VOICE_PUBLIC_DOMAIN " + "(or TACConfig.voice_public_domain)." + ) + + def _resolve_default_action_url(self) -> str | None: + """Resolve the default ```` cleanup URL. + + Returns None if ``voice_public_domain`` isn't set; that's fine because + action_url has higher-priority layers (customizer, twiml_options, + Studio handoff) above this fallback. + """ + if self.tac.config.voice_public_domain: + return ( + f"https://{self.tac.config.voice_public_domain}{self.tac.config.voice_action_path}" + ) + return None + @staticmethod def _caller_address(setup_msg: SetupMessage) -> str | None: """Return the phone number of the remote caller/callee from the setup message.""" @@ -97,52 +155,120 @@ def _get_twilio_client(self) -> Client: async def handle_incoming_call( self, - options: TwiMLOptions | dict[str, Any], + twiml_request: TwiMLRequest | None = None, ) -> str: """ Generate TwiML response for incoming voice calls. ConversationRelay automatically handles conversation creation and participant - management via the conversation_configuration parameter. + management via the ``conversation_configuration`` parameter. + + The WebSocket URL and default session-cleanup action URL are derived + from ``TACConfig.voice_public_domain`` + ``TACConfig.voice_websocket_path`` + / ``voice_action_path``. + + 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. + 2. ``VoiceChannelConfig.default_twiml_options`` — per-channel defaults. + 3. TAC defaults: a fixed default ``welcome_greeting``, + ``conversation_configuration`` from ``TACConfig``, and + ``action_url`` resolved via Studio handoff (when + ``studio_handoff_flow_sid`` is configured), else derived from + ``TACConfig.voice_public_domain`` + ``voice_action_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. Args: - options: TwiML generation options (TwiMLOptions or dict) containing: - - websocket_url (required): WebSocket URL for ConversationRelay - - custom_parameters (optional): Additional custom parameters - - welcome_greeting (optional): Initial greeting message - - action_url (optional): URL for call completion webhook + twiml_request: Parsed Twilio webhook fields. Passed to the + customizer if one is configured on the channel. Returns: - TwiML XML string for call connection + TwiML XML string for call connection. + """ + websocket_url = self._resolve_websocket_url("handle_incoming_call") - Example: - >>> twiml = await voice_channel.handle_incoming_call( - ... options={ - ... "websocket_url": "wss://example.com/ws", - ... "custom_parameters": {"session_id": "sess_123"}, - ... "welcome_greeting": "Hello!", - ... "action_url": "https://example.com/callback", - ... }, - ... ) + 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) + 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 + ``InitiateVoiceConversationOptions.twiml_options`` for outbound). """ - # Handle dict input (convert to TwiMLOptions) - if isinstance(options, dict): - options = TwiMLOptions(**options) - - # Set default welcome greeting if not provided - if options.welcome_greeting is None: - options.welcome_greeting = "Hello! How can I assist you today?" - - # ConversationRelay automatically creates conversation and participants - return twiml.generate_twiml( - TwiMLOptions( - websocket_url=options.websocket_url, - custom_parameters=options.custom_parameters, - welcome_greeting=options.welcome_greeting, - action_url=options.action_url, - conversation_configuration=self.tac.config.conversation_configuration_id, - ) + merged = TwiMLOptions( + welcome_greeting=DEFAULT_WELCOME_GREETING, + conversation_configuration=self.tac.config.conversation_configuration_id, + action_url=self._resolve_action_url(per_call), ) + if self.config.default_twiml_options is not None: + self._overlay_fields(merged, self.config.default_twiml_options) + if per_call is not None: + self._overlay_fields(merged, per_call) + return merged + + @staticmethod + def _overlay_fields(target: TwiMLOptions, source: TwiMLOptions) -> None: + """Apply fields explicitly set on ``source`` onto ``target``. + + Nested models (``custom_parameters``), lists (``languages``), and + dicts (``extra``) replace wholesale — there's no per-key merging. + If you add a field that should merge (e.g. a dict of headers), + special-case it here instead of getting the default overwrite behavior. + + ``action_url`` is skipped here on purpose — it's resolved once via + ``_resolve_action_url`` looking at every layer at once, and that + resolved value is written into ``target`` before this overlay runs. + Letting it through here would let a higher-priority layer that didn't + set action_url silently clobber a lower layer that did. + """ + for field in source.model_fields_set: + if field == "action_url": + continue + setattr(target, field, getattr(source, field)) + + def _resolve_action_url(self, customized: TwiMLOptions | None) -> str | None: + """Resolve the TwiML ```` URL. + + Precedence (highest to lowest): + 1. 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`` + + ``TACConfig.voice_action_path``. + + User-expressed intent (Studio handoff is configured explicitly on + ``TACConfig``) beats the SDK's generated cleanup default. If a user + sets both Studio handoff and runs in relay-only mode, Studio wins + for that call — the session-cleanup URL is skipped, same as if they + had set any other action_url via customizer or static options. + + Explicit ``action_url=None`` on a layer suppresses + ```` entirely — all lower layers are skipped. + Use this to disable the cleanup callback for a specific call (e.g. + from a customizer) or channel-wide. ``action_url`` left unset (not + in ``model_fields_set``) falls through to the next layer. + """ + if customized is not None and "action_url" in customized.model_fields_set: + return customized.action_url + if ( + self.config.default_twiml_options is not None + and "action_url" in self.config.default_twiml_options.model_fields_set + ): + return self.config.default_twiml_options.action_url + if self.tac.config.studio_handoff_flow_sid: + return studio_voice_handoff_url( + self.tac.config.account_sid, + self.tac.config.studio_handoff_flow_sid, + ) + return self._resolve_default_action_url() async def handle_conversation_relay_callback( self, @@ -347,11 +473,25 @@ async def initiate_outbound_conversation( conversation during passive hydration. The session is initialized lazily on the first prompt when the conversation is discovered by callSid. - ``options.websocket_url`` must be the publicly accessible WebSocket - endpoint (e.g., ``wss://your-domain.ngrok.app/ws``). Unlike inbound calls - where TACServer sets this automatically, outbound calls require it - explicitly since there is no incoming HTTP request to derive the host from. + TwiML fields are merged per-field, highest precedence first: + 1. ``options.twiml_options`` — per-call overrides + 2. ``VoiceChannelConfig.default_twiml_options`` — channel-wide defaults + 3. TAC defaults: welcome greeting, ``conversation_configuration`` + from ``TACConfig``, and ``action_url`` from Studio handoff (if + configured), else derived from ``TACConfig.voice_public_domain`` + + ``voice_action_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. + + The WebSocket URL is derived from ``TACConfig.voice_public_domain`` + + ``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( @@ -360,16 +500,13 @@ 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) + try: - twiml_xml = twiml.generate_twiml( - TwiMLOptions( - websocket_url=options.websocket_url, - welcome_greeting=options.welcome_greeting, - action_url=options.action_url, - conversation_configuration=self.tac.config.conversation_configuration_id, - custom_parameters=options.custom_parameters, - ) - ) + twiml_xml = twiml.generate_twiml(websocket_url, merged) client = self._get_twilio_client() call = await asyncio.to_thread( diff --git a/src/tac/channels/voice/config.py b/src/tac/channels/voice/config.py index 446a789..7443058 100644 --- a/src/tac/channels/voice/config.py +++ b/src/tac/channels/voice/config.py @@ -1,15 +1,37 @@ """Voice channel configuration.""" +from collections.abc import Awaitable, Callable + from pydantic import BaseModel, Field from tac.models.memory import MemoryMode +from tac.models.voice import TwiMLOptions, TwiMLRequest from tac.session import SessionManager, ThreadSafeSessionManager +InboundCallTwiMLHandler = Callable[[TwiMLRequest], Awaitable[TwiMLOptions]] + class VoiceChannelConfig(BaseModel): """ Configuration for Voice channel. + TwiML configuration layers (highest precedence first): + + Inbound calls (``handle_incoming_call``): + 1. Output of the customizer registered via + ``VoiceChannel.on_inbound_call_twiml(...)`` [optional] + 2. ``default_twiml_options`` [optional] + 3. TAC defaults + + Outbound calls (``initiate_outbound_conversation``): + 1. ``InitiateVoiceConversationOptions.twiml_options`` [optional] + 2. ``default_twiml_options`` [optional] + 3. TAC defaults + + All layers merge per-field via Pydantic's ``model_fields_set`` — only + fields a layer explicitly sets override lower layers. Lists (``languages``) + and nested models (``custom_parameters``) replace wholesale when set. + Attributes: session_manager: SessionManager for tracking and canceling in-flight tasks. Defaults to ThreadSafeSessionManager for automatic task cancellation on @@ -19,9 +41,17 @@ class VoiceChannelConfig(BaseModel): - "once": Retrieve memory once at conversation start with empty query and cache it. Cache is invalidated when conversation becomes INACTIVE. - "never": Skip memory retrieval + default_twiml_options: Static ``TwiMLOptions`` applied to every call + (inbound and outbound). Controls the TwiML inside + ```` — voice, language, transcription provider, + welcome_greeting, ```` children, etc. Use this when the + same ConversationRelay configuration is correct for every call. + + Per-call inbound customization is registered via + ``VoiceChannel.on_inbound_call_twiml(...)`` (not on this config). """ - model_config = {"arbitrary_types_allowed": True} + model_config = {"arbitrary_types_allowed": True, "extra": "forbid"} session_manager: SessionManager | None = Field( default_factory=ThreadSafeSessionManager, @@ -34,3 +64,11 @@ class VoiceChannelConfig(BaseModel): default="never", description="Memory retrieval mode for this channel", ) + default_twiml_options: TwiMLOptions | None = Field( + default=None, + description="Static TwiMLOptions for the TwiML inside , " + "applied to every call (inbound and outbound). Per-call inbound " + "customization is registered via VoiceChannel.on_inbound_call_twiml(...). " + "Note: ``custom_parameters`` and ``languages`` replace wholesale when a " + "higher-priority layer sets them — see VoiceChannel._overlay_fields.", + ) diff --git a/src/tac/channels/voice/twiml.py b/src/tac/channels/voice/twiml.py index 256a6a5..1eab08c 100644 --- a/src/tac/channels/voice/twiml.py +++ b/src/tac/channels/voice/twiml.py @@ -7,81 +7,158 @@ from tac.models.voice import TwiMLOptions +# Fields on TwiMLOptions that map to attributes and are +# emitted via the snake_case → camelCase conversion done by twilio's SDK. +# Must stay in sync with TwiMLOptions field declarations; see _verify_attrs_in_sync. +_OPTIONAL_RELAY_ATTRS = ( + "welcome_greeting", + "welcome_greeting_interruptible", + "conversation_configuration", + # language / TTS / STT + "language", + "tts_language", + "transcription_language", + "voice", + "tts_provider", + "transcription_provider", + "speech_model", + "elevenlabs_text_normalization", + # turn detection / interruption + "eot_threshold", + "partial_prompts", + "deepgram_smart_format", + "speech_timeout", + "interruptible", + "interrupt_sensitivity", + "report_input_during_agent_speech", + "ignore_backchannel", + "preemptible", + "dtmf_detection", + # hints / events / debug / intelligence + "hints", + "events", + "debug", + "intelligence_service", +) + +# Fields on TwiMLOptions that this module handles specially (not via the +# generic _OPTIONAL_RELAY_ATTRS loop) — the action_url, the +# children list, the children dict, and the extra escape hatch. +_HANDLED_OUTSIDE_LOOP = { + "action_url", + "languages", + "custom_parameters", + "extra", +} + + +def _verify_attrs_in_sync() -> None: + """Fail fast at import time if TwiMLOptions grows a field that isn't + accounted for here — either it's a new ConversationRelay attribute that + needs to go in _OPTIONAL_RELAY_ATTRS, or it's special-cased and should + be added to _HANDLED_OUTSIDE_LOOP. + """ + declared = set(TwiMLOptions.model_fields) + accounted = set(_OPTIONAL_RELAY_ATTRS) | _HANDLED_OUTSIDE_LOOP + missing = declared - accounted + extra = accounted - declared + if missing: + raise RuntimeError( + f"TwiMLOptions field(s) {sorted(missing)} not handled by twiml.py — " + "add to _OPTIONAL_RELAY_ATTRS or _HANDLED_OUTSIDE_LOOP." + ) + if extra: + raise RuntimeError( + f"twiml.py references TwiMLOptions field(s) {sorted(extra)} that " + "no longer exist on the model." + ) + + +_verify_attrs_in_sync() + def generate_twiml( - options: TwiMLOptions | dict[str, Any], + websocket_url: str, + options: TwiMLOptions | dict[str, Any] | None = None, ) -> str: """ - Generate TwiML XML for ConversationRelay with custom parameters. + Generate TwiML XML for ConversationRelay. - This is a low-level function for generating TwiML with arbitrary custom - parameters. For automatic conversation creation and participant management, - use VoiceChannel.handle_incoming_call() instead. + This is a low-level function. Most users should call + ``VoiceChannel.handle_incoming_call`` instead — it layers in TAC defaults, + static ``twiml_options`` from ``VoiceChannelConfig``, and any per-call + customizer output. Args: - options: TwiML generation options (TwiMLOptions model or dict with: - - websocket_url (required): WebSocket URL for ConversationRelay - - custom_parameters (optional): Dict of custom parameters - - welcome_greeting (optional): Initial greeting message - - action_url (optional): URL for call end webhook - - conversation_configuration (optional): Conversation Service SID for - automatic conversation creation + websocket_url: Public WebSocket URL for ConversationRelay + (e.g. ``'wss://example.ngrok.app/ws'``). + 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 + TwiML XML string ready to return to Twilio. Example: >>> twiml = generate_twiml( - ... { - ... "websocket_url": "wss://example.com/voice", - ... "custom_parameters": { - ... "session_id": "sess_abc123", - ... "user_language": "es", - ... }, - ... "welcome_greeting": "Hello!", - ... "conversation_configuration": "conv_configuration_xxxx", - ... } + ... "wss://example.com/voice", + ... TwiMLOptions( + ... welcome_greeting="Hello!", + ... conversation_configuration="conv_configuration_xxxx", + ... ), ... ) """ - # Handle dict input (convert to TwiMLOptions) - if isinstance(options, dict): + if options is None: + options = TwiMLOptions() + elif isinstance(options, dict): options = TwiMLOptions(**options) - websocket_url = options.websocket_url - custom_parameters = options.custom_parameters - welcome_greeting = options.welcome_greeting - action_url = options.action_url - conversation_configuration = options.conversation_configuration - - # Create VoiceResponse response = VoiceResponse() # Create Connect verb with optional action connect_kwargs: dict[str, str] = {} - if action_url: - connect_kwargs["action"] = action_url + if options.action_url: + connect_kwargs["action"] = options.action_url connect = response.connect(**connect_kwargs) - # Build ConversationRelay kwargs - relay_kwargs: dict[str, str] = {"url": websocket_url} - if welcome_greeting: - relay_kwargs["welcome_greeting"] = welcome_greeting - if conversation_configuration: - relay_kwargs["conversation_configuration"] = conversation_configuration + # 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} + for attr in _OPTIONAL_RELAY_ATTRS: + value = getattr(options, attr) + if value is None: + continue + # Twilio accepts True/False on `interruptible` for backward-compat + # but the documented enum is none|dtmf|speech|any. Normalize so we + # emit canonical values regardless of Twilio SDK's bool serialization. + if attr == "interruptible" and isinstance(value, bool): + value = "any" if value else "none" + relay_kwargs[attr] = value + + # TwiMLOptions' validator already rejects extra keys that shadow typed + # fields, so we can pass everything through here as-is. + if options.extra: + relay_kwargs.update(options.extra) - # Create ConversationRelay relay = connect.conversation_relay(**relay_kwargs) - # Add custom parameters - if custom_parameters: - # Handle both Pydantic model and dict + # Emit children, if any + if options.languages: + for lang in options.languages: + lang_kwargs: dict[str, Any] = {"code": lang.code} + for attr in ("voice", "tts_provider", "transcription_provider", "speech_model"): + value = getattr(lang, attr) + if value is not None: + lang_kwargs[attr] = value + relay.language(**lang_kwargs) + + # Add custom parameters as children + if options.custom_parameters: params_dict: dict[str, Any] = ( - custom_parameters.model_dump(by_alias=True, exclude_none=True) - if isinstance(custom_parameters, BaseModel) - else custom_parameters + options.custom_parameters.model_dump(by_alias=True, exclude_none=True) + if isinstance(options.custom_parameters, BaseModel) + else options.custom_parameters ) - - # Add each parameter as a child element for name, value in params_dict.items(): if value is not None: relay.parameter(name=name, value=str(value)) diff --git a/src/tac/core/config.py b/src/tac/core/config.py index 7653d34..c9ab8b8 100644 --- a/src/tac/core/config.py +++ b/src/tac/core/config.py @@ -277,6 +277,51 @@ def _normalize_and_validate_region(cls, v: object) -> str | None: "from this SID.", ) + voice_public_domain: str | None = Field( + default=None, + description="Public domain where voice routes are reachable (e.g. " + "'example.ngrok.app'). Used by VoiceChannel to construct the public " + "WebSocket URL and ConversationRelay action URL. Required when using " + "the Voice channel. Schemes (https://, wss://) and trailing slashes " + "are stripped automatically.", + ) + + voice_websocket_path: str = Field( + default="/ws", + description="Path the voice WebSocket is served at. Combined with " + "voice_public_domain to build the public WebSocket URL the voice " + "channel hands to Twilio in TwiML; TACFastAPIServer also registers " + "its WebSocket route at this path. Override only if you mount the " + "route at a non-default path.", + ) + + voice_action_path: str = Field( + default="/conversation-relay-callback", + description="Path the ConversationRelay action callback is served at. " + "Same role as voice_websocket_path but for the " + "cleanup callback.", + ) + + @field_validator("voice_public_domain", mode="before") + @classmethod + def _normalize_voice_public_domain(cls, v: str | None) -> str | None: + """Strip whitespace, schemes, and trailing slashes from voice_public_domain. + + A naive copy-paste from a browser address bar produces values like + ``https://example.ngrok.app/`` which would otherwise concatenate into + ``wss://https://example.ngrok.app//ws`` — clean them up at parse time. + """ + if v is None: + return v + v = v.strip() + if not v: + return None + for scheme in ("https://", "http://", "wss://", "ws://"): + if v.lower().startswith(scheme): + v = v[len(scheme) :] + break + return v.rstrip("/") or None + conversation_intelligence_config: ConversationIntelligenceConfig | None = Field( default=None, description="Optional Conversation Intelligence configuration for filtering webhook " @@ -330,6 +375,10 @@ def from_env(cls) -> "TACConfig": Default: INFO - TWILIO_REGION: Twilio region for data residency (e.g., 'au1', 'ie1') - TWILIO_STUDIO_HANDOFF_FLOW_SID: Studio Flow SID (FWxxx...) for handoff tool + - TWILIO_VOICE_PUBLIC_DOMAIN: Public domain for voice routes (required for voice) + - TWILIO_VOICE_WEBSOCKET_PATH: Path for voice WebSocket (default: /ws) + - TWILIO_VOICE_ACTION_PATH: Path for ConversationRelay action callback + (default: /conversation-relay-callback) Memory Configuration: - TWILIO_MEMORY_PROFILE_TRAIT_GROUPS: Trait groups to include @@ -368,6 +417,11 @@ def from_env(cls) -> "TACConfig": log_level=os.environ.get("TWILIO_LOG_LEVEL", "INFO"), region=os.environ.get("TWILIO_REGION"), studio_handoff_flow_sid=os.environ.get("TWILIO_STUDIO_HANDOFF_FLOW_SID"), + voice_public_domain=os.environ.get("TWILIO_VOICE_PUBLIC_DOMAIN"), + voice_websocket_path=os.environ.get("TWILIO_VOICE_WEBSOCKET_PATH", "/ws"), + voice_action_path=os.environ.get( + "TWILIO_VOICE_ACTION_PATH", "/conversation-relay-callback" + ), memory_config=memory_config, conversation_intelligence_config=conversation_intelligence_config, ) diff --git a/src/tac/models/outbound.py b/src/tac/models/outbound.py index a131d0a..7041c7b 100644 --- a/src/tac/models/outbound.py +++ b/src/tac/models/outbound.py @@ -5,6 +5,7 @@ from pydantic import BaseModel, Field from tac.models.session import ConversationSession +from tac.models.voice import TwiMLOptions class InitiateMessagingConversationOptions(BaseModel): @@ -52,15 +53,41 @@ class InitiateVoiceConversationOptions(BaseModel): The caller identity is always TAC's configured ``config.phone_number``. Multi-number deployments should use one TAC instance per line. + + TwiML for the outbound call is built by merging per-field, highest + precedence first: + 1. This call's ``twiml_options`` (per-call overrides) + 2. ``VoiceChannelConfig.default_twiml_options`` (channel-wide defaults) + 3. TAC defaults (welcome greeting, conversation_configuration, + action_url resolved via Studio handoff if configured) + + Fields you don't set at a layer fall through to lower layers — so + ``twiml_options=TwiMLOptions(voice="es-MX-Neural2-A")`` on this call + overrides only ``voice``; ``language``, ``interruptible``, etc. from the + channel config still apply. + + Set ``voice``, ``language``, ``interruptible``, etc. on the channel's + ``VoiceChannelConfig.default_twiml_options`` to apply them to every call + (both inbound and outbound). Use this model's ``twiml_options`` for + per-call overrides (e.g. campaign-specific ``custom_parameters``). """ to: str = Field(..., min_length=1) - websocket_url: str = Field(...) - welcome_greeting: str | None = Field(default=None) - action_url: str | None = Field(default=None) - custom_parameters: dict[str, str | int | bool] | None = Field(default=None) - - model_config = {"populate_by_name": True} + websocket_url: str | None = Field( + default=None, + description="Public WebSocket URL for ConversationRelay (e.g. " + "'wss://your-domain.ngrok.app/ws'). Optional — defaults to the URL " + "derived from ``TACConfig.voice_public_domain`` + " + "``voice_websocket_path``. Pass it here only to override the URL " + "for a specific call.", + ) + twiml_options: TwiMLOptions | None = Field( + default=None, + description="Per-call overrides for the TwiML inside . " + "Merged over VoiceChannelConfig.default_twiml_options and TAC defaults.", + ) + + model_config = {"populate_by_name": True, "extra": "forbid"} class InitiateVoiceConversationResult(BaseModel): diff --git a/src/tac/models/voice.py b/src/tac/models/voice.py index ee4d44f..f3de5e1 100644 --- a/src/tac/models/voice.py +++ b/src/tac/models/voice.py @@ -2,7 +2,11 @@ from typing import Any, Literal -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, 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. +InterruptMode = Literal["none", "dtmf", "speech", "any"] class CustomParameters(BaseModel): @@ -95,10 +99,78 @@ class InterruptMessage(BaseModel): VoiceMessage = SetupMessage | PromptMessage | InterruptMessage +class ConversationRelayCallbackPayload(BaseModel): + """ + Payload received from Twilio ConversationRelay callback webhook. + + Sent when a ConversationRelay session ends or transitions state. + """ + + account_sid: str = Field(..., alias="AccountSid", description="Twilio Account SID") + call_sid: str = Field(..., alias="CallSid", description="Twilio Call SID") + call_status: str = Field( + ..., + alias="CallStatus", + description="Call status (e.g., 'in-progress', 'completed', 'busy', 'no-answer')", + ) + from_number: str = Field(..., alias="From", description="Caller's identifier") + to_number: str = Field(..., alias="To", description="Recipient's identifier") + direction: str = Field(..., alias="Direction", description="Call direction (inbound/outbound)") + application_sid: str | None = Field( + None, alias="ApplicationSid", description="Twilio Application SID" + ) + session_id: str | None = Field( + None, alias="SessionId", description="ConversationRelay Session ID" + ) + session_status: str | None = Field( + None, + alias="SessionStatus", + description="ConversationRelay session status (e.g., 'ended')", + ) + session_duration: str | None = Field( + None, alias="SessionDuration", description="Session duration in seconds" + ) + + model_config = {"populate_by_name": True} + + +class LanguageConfig(BaseModel): + """A single ```` child for multi-language ConversationRelay setups. + + Maps to the ```` element documented at + https://www.twilio.com/docs/voice/twiml/connect/conversationrelay#language-element + """ + + code: str = Field( + ..., + description="Language code, e.g. 'es-MX'. Can be 'multi' for automatic language " + "detection (requires ElevenLabs TTS and Deepgram STT).", + ) + voice: str | None = Field(None, description="TTS voice name for this language") + tts_provider: str | None = Field(None, description="TTS provider, e.g. 'google'") + transcription_provider: str | None = Field( + None, description="Transcription provider, e.g. 'deepgram'" + ) + speech_model: str | None = Field( + None, + description="Speech model for STT. Choices vary by transcription_provider; " + "see the provider's documentation.", + ) + + model_config = {"populate_by_name": True} + + class TwiMLOptions(BaseModel): - """Options for generating ConversationRelay TwiML.""" + """Options for the TwiML inside ```` (plus the + ```` URL). + + Fields map to the attributes documented at + https://www.twilio.com/docs/voice/twiml/connect/conversationrelay . + All fields are optional. ``VoiceChannel.handle_incoming_call`` merges these + values over TAC defaults using Pydantic's ``model_fields_set`` — only + fields explicitly set by the caller override TAC's defaults. + """ - websocket_url: str = Field(..., description="WebSocket URL for ConversationRelay") custom_parameters: CustomParameters | dict[str, Any] | None = Field( None, description="Custom parameters to pass to ConversationRelay", @@ -107,6 +179,11 @@ class TwiMLOptions(BaseModel): None, description="Initial greeting message for caller", ) + welcome_greeting_interruptible: InterruptMode | None = Field( + None, + description="What caller input can interrupt the welcome greeting. " + "Defaults to 'any' on Twilio.", + ) action_url: str | None = Field( None, description="URL for Twilio to request when call ends", @@ -117,25 +194,185 @@ class TwiMLOptions(BaseModel): "manage conversation creation and participants.", ) - model_config = {"populate_by_name": True} + # Language, TTS, STT + language: str | None = Field( + None, + description="Language for both STT and TTS, e.g. 'en-US'. Equivalent to setting " + "both tts_language and transcription_language.", + ) + tts_language: str | None = Field( + None, + description="TTS language code; overrides `language` for TTS.", + ) + transcription_language: str | None = Field( + None, + description="STT language code; overrides `language` for transcription. " + "Can be 'multi' for automatic language detection (Deepgram only).", + ) + voice: str | None = Field(None, description="TTS voice name (choices vary by tts_provider)") + tts_provider: str | None = Field( + None, + description="TTS provider: 'Google', 'Amazon', or 'ElevenLabs'. Defaults to 'ElevenLabs'.", + ) + transcription_provider: str | None = Field( + None, + description="STT provider: 'Google' or 'Deepgram'. Defaults to 'Deepgram' (or 'Google' " + "for accounts that used ConversationRelay before 2025-09-12).", + ) + speech_model: str | None = Field( + None, + description="Speech model for STT. Choices vary by transcription_provider.", + ) + elevenlabs_text_normalization: Literal["on", "auto", "off"] | None = Field( + None, + description="Text normalization for ElevenLabs TTS. Defaults to 'off'. " + "'auto' behaves like 'off' for ConversationRelay calls.", + ) + # Turn detection / interruption + eot_threshold: float | None = Field( + None, + description="Confidence required to finish a turn. " + "Only applies with Deepgram + flux speech model. " + "Twilio enforces the accepted range — see ConversationRelay docs.", + ) + partial_prompts: bool | None = Field( + None, + description="Send unfinalized prompts and eager end-of-turn events " + "(last=False). Only applies with Deepgram + flux speech model.", + ) + deepgram_smart_format: bool | None = Field( + None, + description="Use Deepgram Smart Format for transcription output. " + "Defaults to true when transcription_provider='Deepgram'.", + ) + speech_timeout: int | Literal["auto"] | None = Field( + None, + description="Silence (ms) after speech before finalizing the prompt. " + "Integer milliseconds or the literal 'auto' (the platform default). " + "Twilio enforces the accepted range — see ConversationRelay docs.", + ) + interruptible: InterruptMode | bool | None = Field( + None, + description="What caller input interrupts TTS playback. Boolean accepted " + "for backward compat: True='any', False='none'. Defaults to 'any'.", + ) + interrupt_sensitivity: Literal["high", "medium", "low"] | None = Field( + None, + description="How easily caller speech triggers an interrupt. Defaults to 'high'.", + ) + report_input_during_agent_speech: InterruptMode | None = Field( + None, + description="What caller input gets reported while the agent is speaking " + "(independent of whether playback is interrupted). Defaults to 'none' since May 2025.", + ) + ignore_backchannel: bool | None = Field( + None, + description="Filter short conversational feedback ('yeah', 'uh-huh', …) " + "so it doesn't interrupt the agent. Defaults to false.", + ) + preemptible: bool | None = Field( + None, + description="Allow text tokens from the next talk cycle to interrupt the current one. " + "Defaults to false.", + ) + dtmf_detection: bool | None = Field( + None, + description="Emit DTMF keypress events over the WebSocket.", + ) -class ConversationRelayCallbackPayload(BaseModel): - """Payload received from Twilio ConversationRelay callback webhook. + # Recognition hints / events / debug / intelligence + hints: str | None = Field( + None, + description="Comma-separated words/phrases likely to appear in speech. " + "Capitalize proper nouns.", + ) + events: str | None = Field( + None, + description="Space-separated event subscriptions, e.g. 'speaker-events tokens-played'.", + ) + debug: str | None = Field( + None, + description="Debug subscription, e.g. 'debugging'. Note: 'speaker-events' and " + "'tokens-played' have moved to the `events` attribute — only use them here " + "for backward compatibility.", + ) + intelligence_service: str | None = Field( + None, + description="Conversation Intelligence (classic) Service SID or unique name for " + "persisting transcripts and running Language Operators.", + ) - Sent via the URL when a call ends or transitions state. - Used in relay-only mode to signal conversation completion. - """ + # Nested children + languages: list[LanguageConfig] | None = Field( + None, description="Additional children for multi-language support" + ) - account_sid: str = Field(..., alias="AccountSid") - call_sid: str = Field(..., alias="CallSid") - call_status: str = Field(..., alias="CallStatus") - from_number: str = Field(..., alias="From") - to_number: str = Field(..., alias="To") - direction: str = Field(..., alias="Direction") - application_sid: str | None = Field(None, alias="ApplicationSid") - session_id: str | None = Field(None, alias="SessionId") - session_status: str | None = Field(None, alias="SessionStatus") - session_duration: str | None = Field(None, alias="SessionDuration") + extra: dict[str, str | bool | int] | None = Field( + None, + description="Escape hatch for ConversationRelay attributes not yet typed on " + "this model. Keys are emitted as-is on ; Twilio's SDK " + "converts snake_case to camelCase, lowercases bools to 'true'/'false', " + "and stringifies ints. Prefer a typed field when one exists — use " + "``extra`` only for newly-added Twilio attributes not yet in this SDK.", + ) model_config = {"populate_by_name": True} + + @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 + this model. Without this, the user's value would be silently dropped + by the TwiML serializer in favor of the typed default — a footgun. + """ + if not self.extra: + return self + typed = set(type(self).model_fields) - {"extra"} + shadowed = sorted(k for k in self.extra if k in typed) + if shadowed: + raise ValueError( + f"TwiMLOptions.extra keys {shadowed} shadow typed fields. " + "Set the typed field directly instead of using ``extra``." + ) + return self + + +class TwiMLRequest(BaseModel): + """Framework-neutral view of the Twilio TwiML webhook form. + + Populated by ``TACFastAPIServer`` from the incoming Twilio webhook, then + passed to a customizer registered via + ``VoiceChannel.on_inbound_call_twiml(...)`` so the application can + produce per-call ``TwiMLOptions`` overrides without depending on FastAPI + types. + """ + + from_number: str | None = Field(None, alias="From") + to_number: str | None = Field(None, alias="To") + call_sid: str | None = Field(None, alias="CallSid") + caller_country: str | None = Field(None, alias="CallerCountry") + caller_state: str | None = Field(None, alias="CallerState") + caller_city: str | None = Field(None, alias="CallerCity") + direction: str | None = Field(None, alias="Direction") + extra: dict[str, str] = Field( + default_factory=dict, + description="Any other fields from the Twilio webhook not captured above. " + "Values are always strings here (webhook form fields are url-encoded), " + "unlike TwiMLOptions.extra which accepts str | bool | int for emitted " + "TwiML attributes.", + ) + + model_config = {"populate_by_name": True, "extra": "ignore"} + + @classmethod + def from_form(cls, form: dict[str, str]) -> "TwiMLRequest": + """Build a context from a raw Twilio form dict, bucketing unknown keys into ``extra``.""" + known_aliases = {f.alias for f in cls.model_fields.values() if f.alias} + known: dict[str, str] = {} + extra: dict[str, str] = {} + for key, value in form.items(): + if key in known_aliases: + known[key] = value + else: + extra[key] = value + return cls(**known, extra=extra) diff --git a/src/tac/server/config.py b/src/tac/server/config.py index 37b83a9..44cf9bd 100644 --- a/src/tac/server/config.py +++ b/src/tac/server/config.py @@ -8,29 +8,19 @@ class TACServerConfig(BaseModel): """Configuration for TAC server implementations. - Controls host/port binding, public domain for WebSocket URLs, - and customizable webhook paths. + Controls host/port binding and the server-only webhook paths. Voice paths + (WebSocket and ConversationRelay action) live on ``TACConfig`` because + they're consumed by the voice channel regardless of which web framework + is used; this server reads them from there to register its routes. """ host: str = Field(default="0.0.0.0", description="Host to bind the server to") port: int = Field(default=8000, description="Port to bind the server to") - public_domain: str = Field( - default="", description="Public domain for WebSocket URL (e.g., 'example.ngrok.io')" - ) - welcome_greeting: str = Field( - default="Hello! How can I assist you today?", - description="Initial greeting message for callers", - ) conversation_webhook_path: str = Field( default="/webhook", description="Path for conversation webhook endpoint (for all channels)" ) twiml_path: str = Field(default="/twiml", description="Path for TwiML generation endpoint") - websocket_path: str = Field(default="/ws", description="Path for voice WebSocket endpoint") - conversation_relay_callback_path: str = Field( - default="/conversation-relay-callback", - description="Path for ConversationRelay action callback endpoint", - ) cintel_webhook_path: str | None = Field( default=None, description="Path for Conversation Intelligence webhook endpoint. " @@ -42,16 +32,13 @@ def from_env(cls) -> "TACServerConfig": """Create config from environment variables. Environment variables: - TWILIO_VOICE_PUBLIC_DOMAIN: Public domain for WebSocket URLs (required for voice) TWILIO_SERVER_HOST: Host to bind to (default: 0.0.0.0) TWILIO_SERVER_PORT: Port to bind to (default: 8000) + + TWILIO_VOICE_PUBLIC_DOMAIN is read by ``TACConfig.from_env`` instead. """ kwargs: dict[str, object] = {} - public_domain = os.environ.get("TWILIO_VOICE_PUBLIC_DOMAIN") - if public_domain: - kwargs["public_domain"] = public_domain - host = os.environ.get("TWILIO_SERVER_HOST") if host: kwargs["host"] = host diff --git a/src/tac/server/fastapi_server.py b/src/tac/server/fastapi_server.py index 452caa1..5b86c50 100644 --- a/src/tac/server/fastapi_server.py +++ b/src/tac/server/fastapi_server.py @@ -16,8 +16,8 @@ from tac.channels.websocket_protocol import WebSocketDisconnectError from tac.core.logging import get_logger from tac.core.tac import TAC +from tac.models.voice import TwiMLRequest from tac.server.config import TACServerConfig -from tac.tools.handoff import studio_voice_handoff_url if TYPE_CHECKING: from tac.channels.messaging import MessagingChannel @@ -112,6 +112,9 @@ def __init__( self.voice_channel = voice_channel self.messaging_channels: list[MessagingChannel] = messaging_channels or [] + if self.voice_channel is not None: + self._validate_voice_url_config() + # Gather all channels that need webhook processing self.webhook_channels: list[BaseChannel] = [] if self.voice_channel: @@ -121,6 +124,27 @@ def __init__( self.app: FastAPI = app if app is not None else FastAPI(title="TAC Server") self._register_routes(self.app) + def _validate_voice_url_config(self) -> None: + """Fail fast at server construction if the voice channel can't build a + WebSocket URL. + + Without this check, the misconfiguration would only surface on the + first inbound call as a 500 — an easy thing to miss in CI or smoke + tests that don't hit voice. Failing here means the server doesn't + start at all if the voice channel isn't wired up. + + Custom adapters (Flask/Django/etc.) constructing ``VoiceChannel`` + directly still get the runtime ``ValueError`` on first request — we + can't know what URL plumbing they're doing outside this server. + """ + assert self.voice_channel is not None # checked by caller + if self.tac.config.voice_public_domain: + return + raise ValueError( + "Voice channel is configured but TACConfig.voice_public_domain is " + "not set. Set it directly or via the TWILIO_VOICE_PUBLIC_DOMAIN env var." + ) + def _register_routes(self, app: FastAPI) -> None: """Register TAC routes (conversation webhook, voice, CI) onto the given FastAPI app.""" config = self.config @@ -181,49 +205,31 @@ async def conversation_webhook(request: Request) -> JSONResponse: if self.voice_channel is not None: vc = self.voice_channel - if not config.public_domain: - logger.warning( - "public_domain is not set — voice URLs will be malformed. " - "Set TWILIO_VOICE_PUBLIC_DOMAIN environment variable." - ) - @app.post(config.twiml_path, dependencies=[Depends(http_sig)]) - async def post_twiml() -> Response: + async def post_twiml(request: Request) -> Response: """Generate TwiML for incoming voice calls.""" - websocket_url = f"wss://{config.public_domain}{config.websocket_path}" - if self.tac.config.studio_handoff_flow_sid: - action_url = studio_voice_handoff_url( - self.tac.config.account_sid, - self.tac.config.studio_handoff_flow_sid, - ) - elif not self.tac.is_orchestrator_enabled(): - action_url = ( - f"https://{config.public_domain}{config.conversation_relay_callback_path}" - ) - else: - action_url = None - - twiml = await vc.handle_incoming_call( - options={ - "websocket_url": websocket_url, - "action_url": action_url, - "welcome_greeting": config.welcome_greeting, - }, - ) + form = await request.form() + form_dict = {k: v for k, v in form.items() if isinstance(v, str)} + twiml_request = TwiMLRequest.from_form(form_dict) + + twiml = await vc.handle_incoming_call(twiml_request=twiml_request) return Response(content=twiml, media_type="application/xml") - @app.websocket(config.websocket_path) + @app.websocket(self.tac.config.voice_websocket_path) async def websocket_endpoint(websocket: WebSocket, _: None = Depends(ws_sig)) -> None: """Handle voice WebSocket connections.""" adapter = FastAPIWebSocketAdapter(websocket) await vc.handle_websocket(adapter) - @app.post(config.conversation_relay_callback_path) + @app.post( + self.tac.config.voice_action_path, + dependencies=[Depends(http_sig)], + ) async def conversation_relay_callback(request: Request) -> Response: """Handle ConversationRelay action callback (call ended).""" try: form_data = await request.form() - payload_dict = {k: str(v) for k, v in form_data.items()} + payload_dict = {k: v for k, v in form_data.items() if isinstance(v, str)} await vc.handle_conversation_relay_callback(payload_dict) except Exception: logger.error("Failed to process ConversationRelay callback", exc_info=True) diff --git a/tests/test_outbound.py b/tests/test_outbound.py index 51cfa4f..b278581 100644 --- a/tests/test_outbound.py +++ b/tests/test_outbound.py @@ -443,6 +443,32 @@ async def test_returns_call_sid(self) -> None: assert result.call_sid == "CAsid789" +class TestInitiateVoiceConversationOptionsForbidsExtra: + """Migration safety: removed fields raise ValidationError instead of + being silently dropped, so callers upgrading from older TAC versions + get a clear signal that their code needs updating.""" + + def test_removed_welcome_greeting_raises(self) -> None: + from pydantic import ValidationError + + with pytest.raises(ValidationError, match="welcome_greeting"): + InitiateVoiceConversationOptions(to="+15551234567", welcome_greeting="Hi!") # type: ignore[call-arg] + + def test_removed_action_url_raises(self) -> None: + from pydantic import ValidationError + + with pytest.raises(ValidationError, match="action_url"): + InitiateVoiceConversationOptions( + to="+15551234567", action_url="https://example.com/end" + ) # type: ignore[call-arg] + + def test_removed_custom_parameters_raises(self) -> None: + from pydantic import ValidationError + + with pytest.raises(ValidationError, match="custom_parameters"): + InitiateVoiceConversationOptions(to="+15551234567", custom_parameters={"k": "v"}) # type: ignore[call-arg] + + # ============================================================================= # isOwnMessage 2-tier # ============================================================================= @@ -662,6 +688,8 @@ async def test_reraises_twilio_rest_error(self) -> None: @pytest.mark.asyncio async def test_custom_parameters_in_twiml(self) -> None: + from tac.models.voice import TwiMLOptions + tac = TAC(get_test_config()) channel = VoiceChannel(tac) @@ -675,7 +703,7 @@ async def test_custom_parameters_in_twiml(self) -> None: InitiateVoiceConversationOptions( to="+15559876543", websocket_url="wss://example.com/ws", - custom_parameters={"foo": "bar"}, + twiml_options=TwiMLOptions(custom_parameters={"foo": "bar"}), ) ) @@ -685,6 +713,8 @@ async def test_custom_parameters_in_twiml(self) -> None: @pytest.mark.asyncio async def test_welcome_greeting_in_twiml(self) -> None: + from tac.models.voice import TwiMLOptions + tac = TAC(get_test_config()) channel = VoiceChannel(tac) @@ -698,13 +728,99 @@ async def test_welcome_greeting_in_twiml(self) -> None: InitiateVoiceConversationOptions( to="+15559876543", websocket_url="wss://example.com/ws", - welcome_greeting="Hi there!", + twiml_options=TwiMLOptions(welcome_greeting="Hi there!"), ) ) call_kwargs = mock_client.calls.create.call_args.kwargs assert "Hi there!" in call_kwargs["twiml"] + @pytest.mark.asyncio + async def test_channel_twiml_options_applied(self) -> None: + """VoiceChannelConfig.twiml_options flows into outbound TwiML.""" + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLOptions + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(voice="en-US-Journey-D", interruptible="speech"), + ), + ) + + mock_call = MagicMock() + mock_call.sid = "CAchan" + mock_client = MagicMock() + mock_client.calls.create.return_value = mock_call + + with patch.object(channel, "_get_twilio_client", return_value=mock_client): + await channel.initiate_outbound_conversation( + InitiateVoiceConversationOptions( + to="+15559876543", + websocket_url="wss://example.com/ws", + ) + ) + + twiml_xml = mock_client.calls.create.call_args.kwargs["twiml"] + assert 'voice="en-US-Journey-D"' in twiml_xml + assert 'interruptible="speech"' in twiml_xml + + @pytest.mark.asyncio + async def test_per_call_twiml_options_override_channel(self) -> None: + """Per-call twiml_options win over channel-static twiml_options.""" + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLOptions + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(voice="en-US-Journey-D"), + ), + ) + + mock_call = MagicMock() + mock_call.sid = "CApercall" + mock_client = MagicMock() + mock_client.calls.create.return_value = mock_call + + with patch.object(channel, "_get_twilio_client", return_value=mock_client): + await channel.initiate_outbound_conversation( + InitiateVoiceConversationOptions( + to="+15559876543", + websocket_url="wss://example.com/ws", + twiml_options=TwiMLOptions(voice="es-MX-Neural2-A"), + ) + ) + + twiml_xml = mock_client.calls.create.call_args.kwargs["twiml"] + assert 'voice="es-MX-Neural2-A"' in twiml_xml + assert "en-US-Journey-D" not in twiml_xml + + @pytest.mark.asyncio + async def test_studio_handoff_used_when_no_action_url(self) -> None: + """Studio handoff URL drives action_url on outbound when no override.""" + flow_sid = "FW" + "a" * 32 + tac = TAC({**get_test_config(), "studio_handoff_flow_sid": flow_sid}) + channel = VoiceChannel(tac) + + mock_call = MagicMock() + mock_call.sid = "CAstudio" + mock_client = MagicMock() + mock_client.calls.create.return_value = mock_call + + with patch.object(channel, "_get_twilio_client", return_value=mock_client): + await channel.initiate_outbound_conversation( + InitiateVoiceConversationOptions( + to="+15559876543", + websocket_url="wss://example.com/ws", + ) + ) + + twiml_xml = mock_client.calls.create.call_args.kwargs["twiml"] + assert f"Flows/{flow_sid}" in twiml_xml + # ============================================================================= # RCS outbound diff --git a/tests/test_relay_only_mode.py b/tests/test_relay_only_mode.py index 9ea021c..214df26 100644 --- a/tests/test_relay_only_mode.py +++ b/tests/test_relay_only_mode.py @@ -17,10 +17,7 @@ from tac.channels.sms import SMSChannel from tac.channels.voice import VoiceChannel from tac.models.session import ConversationSession -from tac.models.voice import ( - ConversationRelayCallbackPayload, - TwiMLOptions, -) +from tac.models.voice import ConversationRelayCallbackPayload def relay_only_config() -> dict: @@ -31,6 +28,7 @@ def relay_only_config() -> dict: "api_key": "SK123", "api_secret": "test_api_secret", "phone_number": "+15551234567", + "voice_public_domain": "example.com", } @@ -90,16 +88,19 @@ async def test_retrieve_memory_returns_empty_in_relay_only(self) -> None: @pytest.mark.asyncio async def test_handle_incoming_call_twiml_omits_conversation_configuration(self) -> None: """TwiML does not include conversationConfiguration in relay-only mode.""" - tac = TAC(relay_only_config()) - channel = VoiceChannel(tac) + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLOptions - twiml = await channel.handle_incoming_call( - TwiMLOptions( - websocket_url="wss://example.com/ws", - welcome_greeting="Hello", - ) + tac = TAC(relay_only_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(welcome_greeting="Hello"), + ), ) + twiml = await channel.handle_incoming_call() + assert "conversationConfiguration" not in twiml @pytest.mark.no_auto_mock diff --git a/tests/test_server.py b/tests/test_server.py index e41e9e4..7a95784 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -19,6 +19,7 @@ def get_test_config() -> dict: "api_secret": "test_api_token", "conversation_configuration_id": "conv_configuration_test123", "phone_number": "+15551234567", + "voice_public_domain": "test.ngrok.io", } @@ -31,48 +32,38 @@ class TestTACServerConfig: """Test TACServerConfig.""" def test_defaults(self) -> None: - config = TACServerConfig(public_domain="example.ngrok.io") + config = TACServerConfig() assert config.host == "0.0.0.0" assert config.port == 8000 - assert config.public_domain == "example.ngrok.io" - assert config.welcome_greeting == "Hello! How can I assist you today?" assert config.conversation_webhook_path == "/webhook" assert config.twiml_path == "/twiml" - assert config.websocket_path == "/ws" assert config.cintel_webhook_path is None def test_custom_paths(self) -> None: config = TACServerConfig( - public_domain="my.domain.com", host="127.0.0.1", port=3000, conversation_webhook_path="/conversations", twiml_path="/voice/twiml", - websocket_path="/voice/ws", cintel_webhook_path="/ci", ) assert config.host == "127.0.0.1" assert config.port == 3000 assert config.conversation_webhook_path == "/conversations" assert config.twiml_path == "/voice/twiml" - assert config.websocket_path == "/voice/ws" assert config.cintel_webhook_path == "/ci" def test_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("TWILIO_VOICE_PUBLIC_DOMAIN", "my.ngrok.io") monkeypatch.setenv("TWILIO_SERVER_HOST", "127.0.0.1") monkeypatch.setenv("TWILIO_SERVER_PORT", "3000") config = TACServerConfig.from_env() - assert config.public_domain == "my.ngrok.io" assert config.host == "127.0.0.1" assert config.port == 3000 def test_from_env_defaults(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("TWILIO_VOICE_PUBLIC_DOMAIN", raising=False) monkeypatch.delenv("TWILIO_SERVER_HOST", raising=False) monkeypatch.delenv("TWILIO_SERVER_PORT", raising=False) config = TACServerConfig.from_env() - assert config.public_domain == "" assert config.host == "0.0.0.0" assert config.port == 8000 @@ -90,6 +81,43 @@ def test_studio_handoff_flow_sid_accepts_value(self) -> None: assert tac.config.studio_handoff_flow_sid == flow_sid +class TestVoiceUrlConfigValidationAtStartup: + """TACFastAPIServer fails fast at construction if a voice channel is + attached but no public URL is configured. Catches the misconfiguration + before the first inbound call rather than returning a 500.""" + + def test_raises_when_voice_channel_without_public_domain(self) -> None: + from tac.channels.voice import VoiceChannel + from tac.server import TACFastAPIServer + + # Config without voice_public_domain + cfg = {**get_test_config()} + cfg.pop("voice_public_domain", None) + tac = TAC(cfg) + vc = VoiceChannel(tac) + + with pytest.raises(ValueError, match="voice_public_domain"): + TACFastAPIServer(tac=tac, voice_channel=vc) + + def test_passes_when_voice_public_domain_set(self) -> None: + from tac.channels.voice import VoiceChannel + from tac.server import TACFastAPIServer + + tac = TAC(get_test_config()) # has voice_public_domain + vc = VoiceChannel(tac) + TACFastAPIServer(tac=tac, voice_channel=vc) # no raise + + def test_passes_without_voice_channel(self) -> None: + """No voice channel attached: validation skipped, server starts fine + even without voice_public_domain.""" + from tac.server import TACFastAPIServer + + cfg = {**get_test_config()} + cfg.pop("voice_public_domain", None) + tac = TAC(cfg) + TACFastAPIServer(tac=tac) # no raise + + class TestWebSocketDisconnectError: """Test WebSocketDisconnectError.""" @@ -220,7 +248,7 @@ async def test_messaging_webhook_fanout(self) -> None: server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[sms, chat], ) app = server.app @@ -270,7 +298,7 @@ async def test_conversation_webhook_handles_channel_errors(self) -> None: server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[sms, chat], ) app = server.app @@ -316,7 +344,7 @@ def test_create_app_voice_only(self) -> None: vc = VoiceChannel(tac) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), voice_channel=vc, ) app = server.app @@ -335,7 +363,7 @@ def test_create_app_messaging_only(self) -> None: sms = SMSChannel(tac) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[sms], ) app = server.app @@ -352,9 +380,7 @@ def test_create_app_with_cintel(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig( - public_domain="test.ngrok.io", cintel_webhook_path="/ci-webhook" - ), + config=TACServerConfig(cintel_webhook_path="/ci-webhook"), ) app = server.app @@ -362,18 +388,25 @@ def test_create_app_with_cintel(self) -> None: assert "/ci-webhook" in route_paths def test_create_app_custom_paths(self) -> None: + """Custom server paths come from TACServerConfig (server-only paths) + and TACConfig (voice paths, which are also consumed by the channel + for URL construction).""" from tac.channels import SMSChannel from tac.channels.voice import VoiceChannel from tac.server import TACFastAPIServer - tac = TAC(get_test_config()) + tac = TAC( + { + **get_test_config(), + "voice_websocket_path": "/voice/ws", + "voice_action_path": "/voice/cleanup", + } + ) server = TACFastAPIServer( tac=tac, config=TACServerConfig( - public_domain="test.ngrok.io", conversation_webhook_path="/conversations", twiml_path="/voice/twiml", - websocket_path="/voice/ws", ), voice_channel=VoiceChannel(tac), messaging_channels=[SMSChannel(tac)], @@ -384,6 +417,7 @@ def test_create_app_custom_paths(self) -> None: assert "/conversations" in route_paths assert "/voice/twiml" in route_paths assert "/voice/ws" in route_paths + assert "/voice/cleanup" in route_paths def test_custom_app_is_used(self) -> None: """User-supplied FastAPI instance is used directly and metadata preserved.""" @@ -395,7 +429,7 @@ def test_custom_app_is_used(self) -> None: custom_app = FastAPI(title="My Custom Service", version="9.9.9") server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), app=custom_app, ) assert server.app is custom_app @@ -413,7 +447,7 @@ def test_custom_app_has_tac_routes(self) -> None: custom_app = FastAPI() server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[SMSChannel(tac)], app=custom_app, ) @@ -429,7 +463,7 @@ def test_default_app_created(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), ) assert isinstance(server.app, FastAPI) assert server.app.title == "TAC Server" @@ -444,7 +478,7 @@ async def test_can_add_custom_route_post_construction(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), ) @server.app.get("/health") @@ -468,7 +502,7 @@ async def test_can_add_middleware_post_construction(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), ) class HeaderMiddleware(BaseHTTPMiddleware): @@ -500,7 +534,7 @@ async def test_can_add_exception_handler(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), ) class MyError(Exception): @@ -528,7 +562,7 @@ def test_routes_registered_exactly_once(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[SMSChannel(tac)], ) route_paths = [r.path for r in server.app.routes if hasattr(r, "path")] @@ -547,7 +581,7 @@ def _build_server(self, **tac_overrides: object) -> object: tac = TAC({**get_test_config(), **tac_overrides}) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), voice_channel=VoiceChannel(tac), ) return TestClient(server.app) @@ -571,7 +605,10 @@ def test_connect_action_uses_studio_webhook_when_flow_sid_set(self) -> None: ) assert expected in resp.text - def test_connect_action_omitted_when_no_handoff_flow(self) -> None: + def test_connect_action_uses_cleanup_url_when_no_handoff_flow(self) -> None: + """Without Studio handoff, action_url falls back to the server's + session-cleanup URL — no-op in orchestrated mode, drives cleanup in + relay-only mode.""" client = self._build_server() # no studio_handoff_flow_sid resp = client.post( # type: ignore[attr-defined] "/twiml", @@ -579,9 +616,7 @@ def test_connect_action_omitted_when_no_handoff_flow(self) -> None: ) assert resp.status_code == 200 - # No action URL when no handoff flow configured - assert "" in resp.text - assert "action=" not in resp.text + assert 'action="https://test.ngrok.io/conversation-relay-callback"' in resp.text class TestSignatureValidation: @@ -597,7 +632,7 @@ async def test_webhook_rejects_missing_signature(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[SMSChannel(tac)], ) transport = ASGITransport(app=server.app) @@ -615,7 +650,7 @@ async def test_webhook_rejects_invalid_signature(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[SMSChannel(tac)], ) transport = ASGITransport(app=server.app) @@ -640,7 +675,7 @@ async def test_webhook_accepts_valid_signature(self) -> None: sms = SMSChannel(tac) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), messaging_channels=[sms], ) with patch.object(sms, "process_webhook", new_callable=AsyncMock): @@ -664,13 +699,58 @@ def test_twiml_rejects_missing_signature(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), voice_channel=VoiceChannel(tac), ) client = TestClient(server.app) resp = client.post("/twiml") assert resp.status_code == 403 + def test_conversation_relay_callback_rejects_missing_signature(self) -> None: + from fastapi.testclient import TestClient + + from tac.channels.voice import VoiceChannel + from tac.server import TACFastAPIServer + + tac = TAC(get_test_config()) + server = TACFastAPIServer( + tac=tac, + config=TACServerConfig(), + voice_channel=VoiceChannel(tac), + ) + client = TestClient(server.app) + resp = client.post("/conversation-relay-callback") + assert resp.status_code == 403 + + def test_conversation_relay_callback_accepts_valid_signature(self) -> None: + from fastapi.testclient import TestClient + + from tac.channels.voice import VoiceChannel + from tac.server import TACFastAPIServer + + tac = TAC(get_test_config()) + server = TACFastAPIServer( + tac=tac, + config=TACServerConfig(), + voice_channel=VoiceChannel(tac), + ) + client = TestClient(server.app) + form_data = { + "AccountSid": "ACtest123", + "CallSid": "CA123", + "CallStatus": "completed", + "From": "+15551234567", + "To": "+15559876543", + "Direction": "inbound", + } + signature = compute_signature("http://testserver/conversation-relay-callback", form_data) + resp = client.post( + "/conversation-relay-callback", + data=form_data, + headers={"X-Twilio-Signature": signature}, + ) + assert resp.status_code == 200 + def test_twiml_accepts_valid_form_signature(self) -> None: from fastapi.testclient import TestClient @@ -680,7 +760,7 @@ def test_twiml_accepts_valid_form_signature(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), voice_channel=VoiceChannel(tac), ) client = TestClient(server.app) @@ -702,9 +782,7 @@ async def test_cintel_webhook_rejects_missing_signature(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig( - public_domain="test.ngrok.io", cintel_webhook_path="/ci-webhook" - ), + config=TACServerConfig(cintel_webhook_path="/ci-webhook"), ) transport = ASGITransport(app=server.app) async with AsyncClient(transport=transport, base_url="http://test") as client: @@ -720,7 +798,7 @@ async def test_custom_routes_not_affected(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), ) @server.app.get("/health") @@ -766,7 +844,7 @@ def test_websocket_rejects_missing_signature(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), voice_channel=VoiceChannel(tac), ) client = TestClient(server.app) @@ -784,10 +862,61 @@ def test_websocket_rejects_invalid_signature(self) -> None: tac = TAC(get_test_config()) server = TACFastAPIServer( tac=tac, - config=TACServerConfig(public_domain="test.ngrok.io"), + config=TACServerConfig(), voice_channel=VoiceChannel(tac), ) client = TestClient(server.app) with pytest.raises(WebSocketDisconnect): with client.websocket_connect("/ws", headers={"X-Twilio-Signature": "invalid"}): pass + + +class TestTwiMLCustomizerEndToEnd: + """Smoke test: server parses Twilio form and channel customizer shapes TwiML.""" + + def test_customizer_on_voice_channel_receives_parsed_context_and_overrides_twiml( + self, + ) -> None: + from fastapi.testclient import TestClient + + from tac.channels.voice import VoiceChannel + from tac.models.voice import TwiMLOptions, TwiMLRequest + from tac.server import TACFastAPIServer + + captured: dict[str, TwiMLRequest] = {} + + async def customizer(ctx: TwiMLRequest) -> TwiMLOptions: + captured["ctx"] = ctx + return TwiMLOptions(voice="en-US-Journey-D", language="en-US") + + tac = TAC(get_test_config()) + vc = VoiceChannel(tac) + + vc.on_inbound_call_twiml(customizer) + server = TACFastAPIServer( + tac=tac, + config=TACServerConfig(), + voice_channel=vc, + ) + client = TestClient(server.app) + form_data = { + "From": "+14155551234", + "To": "+15551234567", + "CallerCountry": "US", + "ApiVersion": "2010-04-01", + } + signature = compute_signature("http://testserver/twiml", form_data) + resp = client.post( + "/twiml", + data=form_data, + headers={"X-Twilio-Signature": signature}, + ) + assert resp.status_code == 200 + ctx = captured["ctx"] + assert ctx.from_number == "+14155551234" + assert ctx.caller_country == "US" + assert ctx.extra == {"ApiVersion": "2010-04-01"} + body = resp.text + assert 'voice="en-US-Journey-D"' in body + assert 'language="en-US"' in body + assert 'url="wss://test.ngrok.io/ws"' in body diff --git a/tests/test_voice_channel.py b/tests/test_voice_channel.py index d3f1501..83778b8 100644 --- a/tests/test_voice_channel.py +++ b/tests/test_voice_channel.py @@ -13,7 +13,6 @@ from tac.models.session import ConversationSession from tac.models.tac import TACMemoryResponse from tac.models.voice import ( - CustomParameters, InterruptMessage, PromptMessage, TwiMLOptions, @@ -29,6 +28,7 @@ def get_test_config() -> dict: "api_secret": "test_api_token", "conversation_configuration_id": "conv_configuration_test123", "phone_number": "+15551234567", + "voice_public_domain": "example.com", } @@ -451,24 +451,23 @@ async def message_callback( @pytest.mark.asyncio async def test_handle_incoming_call(self) -> None: """Test handle_incoming_call generates valid TwiML with conversation_configuration.""" - tac = TAC(get_test_config()) - channel = VoiceChannel(tac) + from tac.channels.voice import VoiceChannelConfig - # Generate TwiML (no need to mock - ConversationRelay handles conversation creation) - twiml = await channel.handle_incoming_call( - options={ - "websocket_url": "wss://example.ngrok.io/ws", - "action_url": "https://example.ngrok.io/flex_handoff", - "welcome_greeting": "Welcome!", - }, + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(welcome_greeting="Welcome!"), + ), ) - # Verify TwiML contains expected elements + twiml = await channel.handle_incoming_call() + assert '' in twiml assert "" in twiml - assert '' in twiml + assert '' in twiml assert "" in twiml @@ -480,15 +479,8 @@ async def test_handle_incoming_call_default_greeting(self) -> None: tac = TAC(get_test_config()) channel = VoiceChannel(tac) - # Generate TwiML without custom greeting (uses default) - twiml = await channel.handle_incoming_call( - options={ - "websocket_url": "wss://test.ngrok.io/ws", - "action_url": "https://example.ngrok.io/flex_handoff", - }, - ) + twiml = await channel.handle_incoming_call() - # Verify default greeting is used assert 'welcomeGreeting="Hello! How can I assist you today?"' in twiml assert 'conversationConfiguration="conv_configuration_test123"' in twiml @@ -969,7 +961,7 @@ async def stream_response() -> AsyncGenerator[str, None]: def test_generate_twiml_minimal(self) -> None: """Test TwiML generation with only websocket URL.""" - twiml = generate_twiml(TwiMLOptions(websocket_url="wss://example.com/voice")) + twiml = generate_twiml("wss://example.com/voice") assert '' in twiml assert "" in twiml @@ -983,10 +975,10 @@ def test_generate_twiml_minimal(self) -> None: def test_generate_twiml_with_welcome_greeting(self) -> None: """Test TwiML generation with welcome greeting.""" twiml = generate_twiml( + "wss://example.com/voice", TwiMLOptions( - websocket_url="wss://example.com/voice", welcome_greeting="Hello! How can I help you?", - ) + ), ) assert 'welcomeGreeting="Hello! How can I help you?"' in twiml @@ -994,10 +986,10 @@ def test_generate_twiml_with_welcome_greeting(self) -> None: def test_generate_twiml_with_action_url(self) -> None: """Test TwiML generation with action URL.""" twiml = generate_twiml( + "wss://example.com/voice", TwiMLOptions( - websocket_url="wss://example.com/voice", action_url="https://example.com/callback", - ) + ), ) assert '' in twiml @@ -1005,15 +997,15 @@ def test_generate_twiml_with_action_url(self) -> None: def test_generate_twiml_with_standard_custom_parameters(self) -> None: """Test TwiML generation with standard TAC custom parameters.""" twiml = generate_twiml( + "wss://example.com/voice", TwiMLOptions( - websocket_url="wss://example.com/voice", custom_parameters={ "conversationId": "CH123", "profileId": "mem_profile_123", "customerParticipantId": "PA_cust", "aiAgentParticipantId": "PA_agent", }, - ) + ), ) assert '' in twiml @@ -1024,91 +1016,42 @@ def test_generate_twiml_with_standard_custom_parameters(self) -> None: def test_generate_twiml_with_arbitrary_custom_parameters(self) -> None: """Test TwiML generation with arbitrary custom parameters.""" twiml = generate_twiml( + "wss://example.com/voice", TwiMLOptions( - websocket_url="wss://example.com/voice", custom_parameters={ "custom_field_1": "value1", "custom_field_2": "value2", "session_id": "sess_123", }, - ) + ), ) assert '' in twiml assert '' in twiml assert '' in twiml - def test_generate_twiml_with_pydantic_model(self) -> None: - """Test TwiML generation using Pydantic CustomParameters model.""" - custom_params = CustomParameters(conversationId="CH123", profileId="mem_profile_123") - - twiml = generate_twiml( - TwiMLOptions( - websocket_url="wss://example.com/voice", - custom_parameters=custom_params, - ) - ) - - # Should use camelCase aliases - assert '' in twiml - assert '' in twiml - - def test_generate_twiml_with_dict_options(self) -> None: - """Test TwiML generation accepting plain dict instead of TwiMLOptions.""" - twiml = generate_twiml( - { - "websocket_url": "wss://example.com/voice", - "custom_parameters": {"key": "value"}, - "welcome_greeting": "Hi there!", - } - ) - - assert 'url="wss://example.com/voice"' in twiml - assert '' in twiml - assert 'welcomeGreeting="Hi there!"' in twiml - def test_generate_twiml_filters_none_values(self) -> None: """Test that None values are excluded from parameters.""" twiml = generate_twiml( + "wss://example.com/voice", TwiMLOptions( - websocket_url="wss://example.com/voice", custom_parameters={ "field1": "value1", "field2": None, "field3": "value3", }, - ) + ), ) assert '' in twiml assert "field2" not in twiml # None should be filtered assert '' in twiml - def test_generate_twiml_escapes_xml_special_chars(self) -> None: - """Test XML character escaping in parameter values.""" - twiml = generate_twiml( - TwiMLOptions( - websocket_url="wss://example.com/voice", - custom_parameters={ - "field": 'value with "quotes" & ampersand', - }, - ) - ) - - # Twilio SDK automatically escapes XML special characters - assert "&" in twiml - assert """ in twiml - # Verify the full escaped parameter is present - expected_param = ( - '' - ) - assert expected_param in twiml - def test_generate_twiml_complete_example(self) -> None: """Test complete TwiML generation with all options.""" twiml = generate_twiml( + "wss://example.ngrok.io/voice", TwiMLOptions( - websocket_url="wss://example.ngrok.io/voice", custom_parameters={ "conversationId": "CH_abc123", "profileId": "mem_profile_xyz", @@ -1116,7 +1059,7 @@ def test_generate_twiml_complete_example(self) -> None: }, welcome_greeting="Welcome to our support line!", action_url="https://example.com/call-ended", - ) + ), ) # Verify all components present @@ -1131,10 +1074,10 @@ def test_generate_twiml_complete_example(self) -> None: def test_generate_twiml_with_conversation_configuration(self) -> None: """Test TwiML generation with conversation_configuration.""" twiml = generate_twiml( + "wss://example.com/voice", TwiMLOptions( - websocket_url="wss://example.com/voice", conversation_configuration="conv_configuration_test_service_123", - ) + ), ) assert 'conversationConfiguration="conv_configuration_test_service_123"' in twiml @@ -1142,61 +1085,517 @@ def test_generate_twiml_with_conversation_configuration(self) -> None: def test_generate_twiml_without_conversation_configuration(self) -> None: """Test TwiML generation without conversation_configuration.""" + twiml = generate_twiml("wss://example.com/voice", TwiMLOptions()) + + # Should not have conversation_configuration in output + assert "conversationConfiguration" not in twiml + + +class TestGenerateTwiMLConversationRelayAttrs: + """The widened TwiMLOptions surface should emit every documented attribute.""" + + def test_voice_and_language_attrs(self) -> None: twiml = generate_twiml( + "wss://example.com/ws", TwiMLOptions( - websocket_url="wss://example.com/voice", - ) + voice="en-US-Journey-D", + language="en-US", + transcription_provider="deepgram", + tts_provider="elevenlabs", + ), ) + assert 'voice="en-US-Journey-D"' in twiml + assert 'language="en-US"' in twiml + assert 'transcriptionProvider="deepgram"' in twiml + assert 'ttsProvider="elevenlabs"' in twiml - # Should not have conversation_configuration in output - assert "conversationConfiguration" not in twiml + def test_interruptible_dtmf_debug(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + interruptible="speech", + dtmf_detection=True, + debug="speaker-events", + ), + ) + assert 'interruptible="speech"' in twiml + assert 'dtmfDetection="true"' in twiml + assert 'debug="speaker-events"' in twiml + + def test_interruptible_bool_normalized_to_enum(self) -> None: + """Twilio accepts True/False on interruptible for backward-compat, + but the documented enum values are 'any'/'none'. Normalize.""" + twiml_true = generate_twiml("wss://example.com/ws", TwiMLOptions(interruptible=True)) + twiml_false = generate_twiml("wss://example.com/ws", TwiMLOptions(interruptible=False)) + assert 'interruptible="any"' in twiml_true + assert 'interruptible="none"' in twiml_false + + def test_language_children_emitted(self) -> None: + from tac.models.voice import LanguageConfig + + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + languages=[ + LanguageConfig( + code="es-MX", + voice="es-MX-Neural2-A", + tts_provider="google", + transcription_provider="google", + speech_model="long", + ), + LanguageConfig(code="fr-FR"), + ], + ), + ) + assert '' in twiml + + def test_welcome_greeting_interruptible(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + welcome_greeting="Hi", + welcome_greeting_interruptible="dtmf", + ), + ) + assert 'welcomeGreetingInterruptible="dtmf"' in twiml + + def test_language_override_attrs(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + tts_language="en-US", + transcription_language="fr-FR", + ), + ) + assert 'ttsLanguage="en-US"' in twiml + assert 'transcriptionLanguage="fr-FR"' in twiml + + def test_speech_model_and_elevenlabs_normalization(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + speech_model="nova-3-general", + elevenlabs_text_normalization="on", + ), + ) + assert 'speechModel="nova-3-general"' in twiml + assert 'elevenlabsTextNormalization="on"' in twiml + + def test_turn_detection_attrs(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + eot_threshold=0.75, + partial_prompts=True, + deepgram_smart_format=False, + speech_timeout=1500, + ), + ) + assert 'eotThreshold="0.75"' in twiml + assert 'partialPrompts="true"' in twiml + assert 'deepgramSmartFormat="false"' in twiml + assert 'speechTimeout="1500"' in twiml + + def test_interrupt_sensitivity_and_report_input(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + interrupt_sensitivity="medium", + report_input_during_agent_speech="speech", + ), + ) + assert 'interruptSensitivity="medium"' in twiml + assert 'reportInputDuringAgentSpeech="speech"' in twiml + + def test_ignore_backchannel_and_preemptible(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions(ignore_backchannel=True, preemptible=True), + ) + assert 'ignoreBackchannel="true"' in twiml + assert 'preemptible="true"' in twiml + + def test_hints_events_intelligence_service(self) -> None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions( + hints="TwiML,ConversationRelay", + events="speaker-events tokens-played", + intelligence_service="GAaabbcc", + ), + ) + assert 'hints="TwiML,ConversationRelay"' in twiml + assert 'events="speaker-events tokens-played"' in twiml + assert 'intelligenceService="GAaabbcc"' in twiml + + def test_speech_timeout_accepts_auto(self) -> None: + opts = TwiMLOptions(speech_timeout="auto") + assert opts.speech_timeout == "auto" + twiml = generate_twiml("wss://example.com/ws", opts) + assert 'speechTimeout="auto"' in twiml + + def test_speech_timeout_rejects_other_strings(self) -> None: + import pytest + from pydantic import ValidationError + + with pytest.raises(ValidationError): + TwiMLOptions(speech_timeout="fast") # type: ignore[arg-type] + + def test_omitted_fields_absent_from_output(self) -> None: + twiml = generate_twiml("wss://example.com/ws") + for attr in ( + "voice=", + "language=", + "transcriptionProvider=", + "ttsProvider=", + "interruptible=", + "dtmfDetection=", + "debug=", + "welcomeGreetingInterruptible=", + "ttsLanguage=", + "transcriptionLanguage=", + "speechModel=", + "elevenlabsTextNormalization=", + "eotThreshold=", + "partialPrompts=", + "deepgramSmartFormat=", + "speechTimeout=", + "interruptSensitivity=", + "reportInputDuringAgentSpeech=", + "ignoreBackchannel=", + "preemptible=", + "hints=", + "events=", + "intelligenceService=", + " None: + twiml = generate_twiml( + "wss://example.com/ws", + TwiMLOptions(extra={"future_feature": "on", "another_attr": True}), + ) + # Twilio SDK snake_case → camelCase + assert 'futureFeature="on"' in twiml + assert 'anotherAttr="true"' in twiml + + def test_extra_shadowing_typed_field_raises(self) -> None: + """A typed-field name in ``extra`` raises at construction. Silent + drop-and-warn would be a footgun: the user explicitly set the field + via ``extra`` and got nothing back.""" + from pydantic import ValidationError + + with pytest.raises(ValidationError, match="shadow typed fields"): + TwiMLOptions(voice="en-US-Journey-D", extra={"voice": "should-not-appear"}) + + with pytest.raises(ValidationError, match="shadow typed fields"): + # Even when the typed field is unset, a shadow key must use the + # typed field directly so validators / type coercion run. + TwiMLOptions(extra={"speech_timeout": 800}) + + def test_extra_none_emits_nothing(self) -> None: + twiml = generate_twiml("wss://example.com/ws", TwiMLOptions()) + # Sanity: no trailing garbage from extra handling when it's unset. + assert " None: - """Test handle_incoming_call includes additional custom parameters.""" + async def test_tac_defaults_applied(self) -> None: tac = TAC(get_test_config()) channel = VoiceChannel(tac) + twiml = await channel.handle_incoming_call() + assert 'welcomeGreeting="Hello! How can I assist you today?"' in twiml + assert 'conversationConfiguration="conv_configuration_test123"' in twiml - # Generate TwiML with additional parameters - twiml = await channel.handle_incoming_call( - options={ - "websocket_url": "wss://example.ngrok.io/ws", - "action_url": "https://example.ngrok.io/callback", - "welcome_greeting": "Welcome!", - "custom_parameters": { - "session_id": "sess_abc123", - "user_language": "es", - "priority": "high", - }, - }, + @pytest.mark.asyncio + async def test_static_options_override_conversation_configuration(self) -> None: + from tac.channels.voice import VoiceChannelConfig + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions( + conversation_configuration="conv_configuration_custom" + ), + ), ) + twiml = await channel.handle_incoming_call() + assert 'conversationConfiguration="conv_configuration_custom"' in twiml + assert "conv_configuration_test123" not in twiml - # Verify conversation_configuration is present - assert 'conversationConfiguration="conv_configuration_test123"' in twiml + @pytest.mark.asyncio + async def test_studio_handoff_used_when_flow_sid_set(self) -> None: + """Studio handoff URL is used when configured and no higher layer set action_url.""" + flow_sid = "FW" + "a" * 32 + tac = TAC({**get_test_config(), "studio_handoff_flow_sid": flow_sid}) + channel = VoiceChannel(tac) + twiml = await channel.handle_incoming_call() + expected = ( + f'action="https://webhooks.twilio.com/v1/Accounts/ACtest123' + f'/Flows/{flow_sid}?Trigger=incomingCall"' + ) + assert expected in twiml + + @pytest.mark.asyncio + async def test_studio_handoff_beats_default_action_url(self) -> None: + """Studio handoff is a user-expressed intent (explicit config) and + wins over the derived cleanup URL. Setting Studio handoff is the + signal that the user wants Studio's cleanup, not the SDK's default.""" + flow_sid = "FW" + "a" * 32 + tac = TAC({**get_test_config(), "studio_handoff_flow_sid": flow_sid}) + channel = VoiceChannel(tac) + twiml = await channel.handle_incoming_call() + expected = ( + f'action="https://webhooks.twilio.com/v1/Accounts/ACtest123' + f'/Flows/{flow_sid}?Trigger=incomingCall"' + ) + assert expected in twiml + # Default cleanup URL must not also appear. + assert "conversation-relay-callback" not in twiml + + @pytest.mark.asyncio + async def test_action_url_falls_back_to_derived_default(self) -> None: + """With nothing else configured, action_url is derived from + TACConfig.voice_public_domain + voice_action_path.""" + tac = TAC(get_test_config()) + channel = VoiceChannel(tac) + twiml = await channel.handle_incoming_call() + assert 'action="https://example.com/conversation-relay-callback"' in twiml + + @pytest.mark.asyncio + async def test_static_options_action_url_beats_derived_default(self) -> None: + """A static action_url on default_twiml_options is an explicit user + choice and wins over the derived cleanup URL.""" + from tac.channels.voice import VoiceChannelConfig + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(action_url="https://static.example.com/end"), + ), + ) + twiml = await channel.handle_incoming_call() + assert 'action="https://static.example.com/end"' in twiml + assert "conversation-relay-callback" not in twiml + + @pytest.mark.asyncio + async def test_action_url_three_layer_resolution(self) -> None: + """Customizer setting only `voice` (no action_url) must not clobber a + default_twiml_options.action_url. Verifies the _overlay_fields skip + invariant — every layer's action_url is funneled through + _resolve_action_url, never via the field overlay.""" + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLRequest + + async def customizer(req: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(voice="en-US-Journey-D") # no action_url + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(action_url="https://static.example.com/end"), + ), + ) + channel.on_inbound_call_twiml(customizer) + twiml = await channel.handle_incoming_call(twiml_request=TwiMLRequest()) + assert 'action="https://static.example.com/end"' in twiml + assert 'voice="en-US-Journey-D"' in twiml + + @pytest.mark.asyncio + async def test_customizer_action_url_none_suppresses(self) -> None: + """Explicit action_url=None on the customizer suppresses the + attribute, even if Studio handoff or a channel + default would otherwise populate it.""" + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLRequest + + async def customizer(req: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(action_url=None) + + flow_sid = "FW" + "a" * 32 + tac = TAC({**get_test_config(), "studio_handoff_flow_sid": flow_sid}) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(action_url="https://static.example.com/end"), + ), + ) + channel.on_inbound_call_twiml(customizer) + twiml = await channel.handle_incoming_call(twiml_request=TwiMLRequest()) + assert "action=" not in twiml + + @pytest.mark.asyncio + async def test_default_options_action_url_none_suppresses(self) -> None: + """Explicit action_url=None on default_twiml_options suppresses + channel-wide, even with Studio handoff configured.""" + from tac.channels.voice import VoiceChannelConfig + + flow_sid = "FW" + "a" * 32 + tac = TAC({**get_test_config(), "studio_handoff_flow_sid": flow_sid}) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(action_url=None), + ), + ) + twiml = await channel.handle_incoming_call() + assert "action=" not in twiml + + +class TestStaticTwiMLOptions: + """VoiceChannelConfig.twiml_options applies to every call without a callback.""" + + @pytest.mark.asyncio + async def test_static_options_applied(self) -> None: + from tac.channels.voice import VoiceChannelConfig + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(voice="en-US-Journey-D", language="en-US"), + ), + ) + twiml = await channel.handle_incoming_call() + assert 'voice="en-US-Journey-D"' in twiml + assert 'language="en-US"' in twiml + + @pytest.mark.asyncio + async def test_welcome_greeting_via_twiml_options(self) -> None: + from tac.channels.voice import VoiceChannelConfig - # Verify additional custom parameters are present - assert '' in twiml - assert '' in twiml - assert '' in twiml + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(welcome_greeting="Bonjour!"), + ), + ) + twiml = await channel.handle_incoming_call() + assert 'welcomeGreeting="Bonjour!"' in twiml + + +class TestCustomizeTwiMLOptions: + """Per-call customizer runs on top of static options and TAC defaults.""" + + @pytest.mark.asyncio + async def test_customizer_skipped_without_twiml_request(self) -> None: + from tac.models.voice import TwiMLRequest + + called = False + + async def customizer(ctx: TwiMLRequest) -> TwiMLOptions: + nonlocal called + called = True + return TwiMLOptions(voice="en-US-Journey-D") + + tac = TAC(get_test_config()) + channel = VoiceChannel(tac) + + channel.on_inbound_call_twiml(customizer) + twiml = await channel.handle_incoming_call() + assert called is False + assert "voice=" not in twiml @pytest.mark.asyncio - async def test_handle_incoming_call_without_additional_parameters(self) -> None: - """Test handle_incoming_call works without additional parameters.""" + async def test_customizer_invoked_with_twiml_request(self) -> None: + from tac.models.voice import TwiMLRequest + + seen: dict[str, TwiMLRequest] = {} + + async def customizer(ctx: TwiMLRequest) -> TwiMLOptions: + seen["ctx"] = ctx + return TwiMLOptions(voice="en-US-Journey-D", interruptible="speech") + tac = TAC(get_test_config()) channel = VoiceChannel(tac) - # Generate TwiML without additional parameters + channel.on_inbound_call_twiml(customizer) + ctx = TwiMLRequest(from_number="+14155551234", caller_country="US") twiml = await channel.handle_incoming_call( - options={ - "websocket_url": "wss://example.ngrok.io/ws", - }, + twiml_request=ctx, ) + assert seen["ctx"] is ctx + assert 'voice="en-US-Journey-D"' in twiml + assert 'interruptible="speech"' in twiml - # Verify conversation_configuration is present - assert 'conversationConfiguration="conv_configuration_test123"' in twiml - # Verify no custom parameters - assert "session_id" not in twiml - assert "user_language" not in twiml + @pytest.mark.asyncio + async def test_customizer_output_beats_static(self) -> None: + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLRequest + + async def customizer(ctx: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(voice="es-MX-Neural2-A") + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(voice="en-US-Journey-D"), + ), + ) + channel.on_inbound_call_twiml(customizer) + twiml = await channel.handle_incoming_call( + twiml_request=TwiMLRequest(), + ) + assert 'voice="es-MX-Neural2-A"' in twiml + + @pytest.mark.asyncio + async def test_customizer_unset_fields_keep_lower_layers(self) -> None: + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLRequest + + async def customizer(ctx: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(voice="en-US-Journey-D") + + tac = TAC(get_test_config()) + channel = VoiceChannel( + tac, + config=VoiceChannelConfig( + default_twiml_options=TwiMLOptions(welcome_greeting="Channel default"), + ), + ) + channel.on_inbound_call_twiml(customizer) + twiml = await channel.handle_incoming_call( + twiml_request=TwiMLRequest(), + ) + # Customizer didn't set welcome_greeting; channel default survives. + assert 'welcomeGreeting="Channel default"' in twiml + assert 'voice="en-US-Journey-D"' in twiml + + @pytest.mark.asyncio + async def test_customizer_action_url_wins_over_studio_handoff(self) -> None: + from tac.models.voice import TwiMLRequest + + flow_sid = "FW" + "a" * 32 + + async def customizer(ctx: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(action_url="https://customizer.example.com/end") + + tac = TAC({**get_test_config(), "studio_handoff_flow_sid": flow_sid}) + channel = VoiceChannel(tac) + + channel.on_inbound_call_twiml(customizer) + twiml = await channel.handle_incoming_call( + twiml_request=TwiMLRequest(), + ) + assert 'action="https://customizer.example.com/end"' in twiml class TestConversationInitializationFlow: @@ -1606,3 +2005,43 @@ async def running_callback(message, context, memory): # Verify task was cancelled assert task_cancelled == [True] assert session_state.stream_task.done() + + +class TestVoicePublicDomainNormalization: + """TACConfig.voice_public_domain strips schemes/whitespace/trailing slashes.""" + + @pytest.mark.parametrize( + "raw,expected", + [ + ("https://example.ngrok.app", "example.ngrok.app"), + ("http://example.ngrok.app", "example.ngrok.app"), + ("wss://example.ngrok.app", "example.ngrok.app"), + ("example.ngrok.app/", "example.ngrok.app"), + ("https://example.ngrok.app/", "example.ngrok.app"), + (" example.ngrok.app ", "example.ngrok.app"), + ("example.ngrok.app", "example.ngrok.app"), + ("", None), + ], + ) + def test_normalizes(self, raw: str, expected: str | None) -> None: + tac = TAC({**get_test_config(), "voice_public_domain": raw}) + assert tac.config.voice_public_domain == expected + + @pytest.mark.asyncio + async def test_normalized_value_produces_valid_twiml(self) -> None: + """Regression: a sloppy https://... / value used to concatenate to + wss://https://example.ngrok.app//ws.""" + tac = TAC({**get_test_config(), "voice_public_domain": "https://example.ngrok.app/"}) + channel = VoiceChannel(tac) + twiml = await channel.handle_incoming_call() + assert 'url="wss://example.ngrok.app/ws"' in twiml + + +class TestVoicePathsOnTACConfig: + """Voice paths live on TACConfig (one source of truth for both the + channel's URL construction and the server's route registration).""" + + def test_default_paths(self) -> None: + tac = TAC(get_test_config()) + assert tac.config.voice_websocket_path == "/ws" + assert tac.config.voice_action_path == "/conversation-relay-callback" diff --git a/tests/test_voice_models.py b/tests/test_voice_models.py index e415e33..3831b36 100644 --- a/tests/test_voice_models.py +++ b/tests/test_voice_models.py @@ -3,8 +3,11 @@ from tac.models.voice import ( CustomParameters, InterruptMessage, + LanguageConfig, PromptMessage, SetupMessage, + TwiMLOptions, + TwiMLRequest, ) @@ -212,3 +215,72 @@ def test_interrupt_aliases(self) -> None: assert msg.utterance_until_interrupt == "Test utterance" assert msg.duration_until_interrupt_ms == 2000 + + +class TestTwiMLRequest: + """TwiMLRequest parses Twilio webhook form fields.""" + + def test_from_form_known_fields(self) -> None: + ctx = TwiMLRequest.from_form( + { + "From": "+14155551234", + "To": "+15551234567", + "CallSid": "CA" + "1" * 32, + "CallerCountry": "US", + "CallerState": "CA", + "CallerCity": "San Francisco", + "Direction": "inbound", + } + ) + assert ctx.from_number == "+14155551234" + assert ctx.to_number == "+15551234567" + assert ctx.call_sid == "CA" + "1" * 32 + assert ctx.caller_country == "US" + assert ctx.caller_state == "CA" + assert ctx.caller_city == "San Francisco" + assert ctx.direction == "inbound" + assert ctx.extra == {} + + def test_from_form_unknown_fields_bucketed_into_extra(self) -> None: + ctx = TwiMLRequest.from_form( + { + "From": "+14155551234", + "ApiVersion": "2010-04-01", + "ForwardedFrom": "+15559999999", + } + ) + assert ctx.from_number == "+14155551234" + assert ctx.extra == { + "ApiVersion": "2010-04-01", + "ForwardedFrom": "+15559999999", + } + + def test_from_form_empty(self) -> None: + ctx = TwiMLRequest.from_form({}) + assert ctx.from_number is None + assert ctx.extra == {} + + +class TestTwiMLOptionsFieldsSet: + """Merge semantics rely on Pydantic's model_fields_set.""" + + def test_unset_scalar_fields_are_none(self) -> None: + options = TwiMLOptions() + assert options.voice is None + assert "voice" not in options.model_fields_set + + def test_explicitly_set_fields_tracked(self) -> None: + options = TwiMLOptions( + voice="en-US-Journey-D", + dtmf_detection=False, + ) + assert "voice" in options.model_fields_set + assert "dtmf_detection" in options.model_fields_set + # Unset fields not tracked + assert "interruptible" not in options.model_fields_set + + def test_language_config_optional_fields(self) -> None: + lang = LanguageConfig(code="es-MX") + assert lang.voice is None + assert lang.tts_provider is None + assert lang.transcription_provider is None