diff --git a/api/routes/telephony.py b/api/routes/telephony.py index cc7ce55..5c57cc2 100644 --- a/api/routes/telephony.py +++ b/api/routes/telephony.py @@ -1670,7 +1670,7 @@ async def complete_transfer_function_call(transfer_id: str, request: Request): result = { "status": "success", "message": "Great! The destination number answered. Let me transfer you now.", - "action": "transfer_success", + "action": "destination_answered", "conference_id": conference_name, "transfer_call_sid": call_sid, # The outbound transfer call SID "original_call_sid": original_call_sid, # The original caller's SID @@ -1714,9 +1714,7 @@ async def complete_transfer_function_call(transfer_id: str, request: Request): try: # Determine event type based on result status if result["status"] == "success": - event_type = TransferEventType.TRANSFER_COMPLETED - elif result.get("reason") == "timeout": - event_type = TransferEventType.TRANSFER_TIMEOUT + event_type = TransferEventType.DESTINATION_ANSWERED else: event_type = TransferEventType.TRANSFER_FAILED @@ -1730,7 +1728,6 @@ async def complete_transfer_function_call(transfer_id: str, request: Request): status=result["status"], action=result.get("action", ""), reason=result.get("reason"), - end_call=result.get("end_call", False), ) # Publish the event via Redis diff --git a/api/routes/tool.py b/api/routes/tool.py index d54589e..ba8e708 100644 --- a/api/routes/tool.py +++ b/api/routes/tool.py @@ -71,7 +71,7 @@ class TransferCallConfig(BaseModel): """Configuration for Transfer Call tools.""" destination: str = Field( - description="Phone number to transfer the call to (E.164 format, e.g., +1234567890)" + description="Phone number or SIP endpoint to transfer the call to (E.164 format e.g., +1234567890, or SIP endpoint e.g., PJSIP/1234)" ) messageType: Literal["none", "custom"] = Field( default="none", description="Type of message to play before transfer" @@ -89,16 +89,23 @@ class TransferCallConfig(BaseModel): @field_validator("destination") @classmethod def validate_destination(cls, v: str) -> str: - """Validate that destination is a valid E.164 phone number.""" + """Validate that destination is a valid E.164 phone number or SIP endpoint.""" # Allow empty string for initial creation (like HTTP API tools with empty URL) if not v.strip(): return v # E.164 format: +[1-9]\d{1,14} e164_pattern = r"^\+[1-9]\d{1,14}$" - if not re.match(e164_pattern, v): + + # SIP endpoint format: PJSIP/extension or SIP/extension + sip_pattern = r"^(PJSIP|SIP)/[\w\-\.@]+$" + + is_valid_e164 = re.match(e164_pattern, v) + is_valid_sip = re.match(sip_pattern, v, re.IGNORECASE) + + if not (is_valid_e164 or is_valid_sip): raise ValueError( - "Destination must be a valid E.164 phone number (e.g., +1234567890)" + "Destination must be a valid E.164 phone number (e.g., +1234567890) or SIP endpoint (e.g., PJSIP/1234)" ) return v diff --git a/api/services/pipecat/transport_setup.py b/api/services/pipecat/transport_setup.py index db18380..c7b56ff 100644 --- a/api/services/pipecat/transport_setup.py +++ b/api/services/pipecat/transport_setup.py @@ -6,6 +6,14 @@ from api.constants import APP_ROOT_DIR from api.db import db_client from api.enums import OrganizationConfigurationKey from api.services.pipecat.audio_config import AudioConfig +from api.services.telephony.providers.ari_call_strategies import ( + ARIBridgeSwapStrategy, + ARIHangupStrategy, +) +from api.services.telephony.providers.twilio_call_strategies import ( + TwilioConferenceStrategy, + TwilioHangupStrategy, +) from pipecat.audio.mixers.silence_mixer import SilenceAudioMixer from pipecat.audio.mixers.soundfile_mixer import SoundfileMixer from pipecat.serializers.asterisk import AsteriskFrameSerializer @@ -54,12 +62,17 @@ async def create_twilio_transport( raise ValueError( f"Incomplete Twilio configuration for organization {organization_id}" ) + # Create strategy instances + transfer_strategy = TwilioConferenceStrategy() + hangup_strategy = TwilioHangupStrategy() serializer = TwilioFrameSerializer( stream_sid=stream_sid, call_sid=call_sid, account_sid=account_sid, auth_token=auth_token, + transfer_strategy=transfer_strategy, + hangup_strategy=hangup_strategy, ) return FastAPIWebsocketTransport( @@ -178,12 +191,17 @@ async def create_ari_transport( f"Incomplete ARI configuration for organization {organization_id}. " f"Required: ari_endpoint, app_name, app_password" ) + # Create strategy instances + transfer_strategy = ARIBridgeSwapStrategy() + hangup_strategy = ARIHangupStrategy() serializer = AsteriskFrameSerializer( channel_id=channel_id, ari_endpoint=ari_endpoint, app_name=app_name, app_password=app_password, + transfer_strategy=transfer_strategy, + hangup_strategy=hangup_strategy, params=AsteriskFrameSerializer.InputParams( asterisk_sample_rate=audio_config.transport_in_sample_rate, sample_rate=audio_config.pipeline_sample_rate, diff --git a/api/services/telephony/ari_manager.py b/api/services/telephony/ari_manager.py index ea8e414..98dacfc 100644 --- a/api/services/telephony/ari_manager.py +++ b/api/services/telephony/ari_manager.py @@ -14,6 +14,7 @@ setup_logging() import asyncio import json import signal +import time from typing import Dict, Optional, Set from urllib.parse import urlparse @@ -26,6 +27,11 @@ from api.constants import REDIS_URL from api.db import db_client from api.enums import CallType, OrganizationConfigurationKey, WorkflowRunMode from api.services.quota_service import check_dograh_quota_by_user_id +from api.services.telephony.call_transfer_manager import get_call_transfer_manager +from api.services.telephony.transfer_event_protocol import ( + TransferEvent, + TransferEventType, +) # Redis key pattern and TTL for channel-to-run mapping _CHANNEL_KEY_PREFIX = "ari:channel:" @@ -62,6 +68,9 @@ class ARIConnection: # Redis client for channel-to-run reverse mapping (lazy init) self._redis_client: Optional[aioredis.Redis] = None + # Transfer manager for handling call transfers + self._call_transfer_manager = None + async def _get_redis(self) -> aioredis.Redis: """Get Redis client instance (lazy init).""" if not self._redis_client: @@ -70,6 +79,12 @@ class ARIConnection: ) return self._redis_client + async def _get_transfer_manager(self): + """Get transfer manager instance.""" + if not self._call_transfer_manager: + self._call_transfer_manager = await get_call_transfer_manager() + return self._call_transfer_manager + async def _set_channel_run(self, channel_id: str, workflow_run_id: str): """Store channel_id -> workflow_run_id mapping in Redis.""" r = await self._get_redis() @@ -229,6 +244,11 @@ class ARIConnection: channel_id = channel.get("id", "unknown") channel_state = channel.get("state", "unknown") + # Log all events for each channel + logger.debug( + f"[ARI EVENT org={self.organization_id}] {event_type}: channel={channel_id}, state={channel_state}" + ) + if event_type == "StasisStart": # Skip external media channels we created — they fire # their own StasisStart but need no further handling. @@ -255,6 +275,19 @@ class ARIConnection: ) else: # Outbound call (state == "Up") — originated by us + # Check if this is a transfer destination channel (app_args starts with "transfer") + # Transfer destinations run externally - we only track status to publish transfer event, not run the pipeline + transfer_id = self._get_transfer_id(app_args) + if transfer_id: + logger.info( + f"[ARI org={self.organization_id}] Transfer destination answered: " + f"channel={channel_id}, transfer_id={transfer_id}" + ) + asyncio.create_task( + self._handle_destination_answered(transfer_id, channel_id) + ) + return + # Parse args to extract workflow context args_dict = {} for arg in app_args: @@ -298,13 +331,26 @@ class ARIConnection: ) elif event_type == "ChannelDestroyed": - cause = channel.get("cause", 0) - cause_txt = channel.get("cause_txt", "unknown") + cause = event.get("cause", 0) + cause_txt = event.get("cause_txt", "unknown") + tech_cause = event.get("tech_cause", "unknown") logger.info( f"[ARI org={self.organization_id}] ChannelDestroyed: " - f"channel={channel_id}, cause={cause} ({cause_txt})" + f"channel={channel_id}, cause={cause} ({cause_txt}), tech_cause = {tech_cause}" ) + # Check if this is a transfer destination that failed + transfer_id = await self._get_transfer_id_for_channel(channel_id) + if transfer_id: + failure_message = self._map_hangup_cause_to_message( + cause, tech_cause, cause_txt + ) + asyncio.create_task( + self._handle_transfer_failed( + transfer_id, channel_id, failure_message + ) + ) + elif event_type == "ChannelDtmfReceived": digit = event.get("digit", "") logger.debug( @@ -580,7 +626,38 @@ class ARIConnection: call_id = ctx.get("call_id") ext_channel_id = ctx.get("ext_channel_id") bridge_id = ctx.get("bridge_id") + transfer_state = ctx.get("transfer_state") + # Check if this is a call transfer scenario external channel. Skip full teardown if + # transfer is in progress and this is the external media channel + # During call transfer, we preserve the caller-destination bridge + if ( + transfer_state == "in-progress" + and channel_id == ext_channel_id + and ext_channel_id is not None + ): + logger.info( + f"[ARI org={self.organization_id}] Transfer in progress - skipping full teardown " + f"for external channel {channel_id}, preserving bridge {bridge_id} and caller {call_id}" + ) + + # Update transfer state to complete + ctx["transfer_state"] = "complete" + await db_client.update_workflow_run( + run_id=int(workflow_run_id), gathered_context=ctx + ) + + # Clean up only Redis markers for external channel + await self._delete_channel_run(channel_id) + await self._delete_ext_channel(channel_id) + + logger.info( + f"[ARI org={self.organization_id}] Transfer cleanup complete - preserved caller {call_id} " + f"in bridge {bridge_id}" + ) + return + + # Normal full teardown for non-transfer scenarios (transfer_state is None or not in-progress) # Delete the bridge first (removes channels from it) if bridge_id: await self._delete_bridge(bridge_id) @@ -633,6 +710,124 @@ class ARIConnection: f"{response.status} {text}" ) + # ======== CALL TRANSFER HELPER METHODS ======== + + def _map_hangup_cause_to_message( + self, cause: int, tech_cause: str, cause_txt: str + ) -> str: + """Map Asterisk cause codes to user-friendly transfer failure messages.""" + if cause == 17 and tech_cause == "486": # User busy/declined + return "The person declined the call or their line is busy." + elif cause == 19 and tech_cause == "480": # No answer + return "The transfer call was not answered. The person may be busy or unavailable right now." + elif cause == 21: # Call rejected + return "The transfer call failed to connect. There may be a network issue or the number is unavailable." + else: + return f"Transfer failed: {cause_txt}" + + def _get_transfer_id(self, app_args: list) -> Optional[str]: + """Get transfer_id if this is a transfer channel, None otherwise. + + Args format: ['transfer', '{transfer_id}', '{conf_name}'] + """ + if len(app_args) > 1 and app_args[0] == "transfer": + transfer_id = app_args[1] + logger.debug( + f"[ARI org={self.organization_id}] Detected transfer channel with transfer_id: {transfer_id}" + ) + return transfer_id + return None + + async def _get_transfer_id_for_channel(self, channel_id: str) -> Optional[str]: + """Get transfer_id for a channel by checking Redis mapping.""" + try: + r = await self._get_redis() + transfer_id = await r.get(f"ari:transfer_channel:{channel_id}") + logger.debug( + f"[ARI Transfer] Looking up transfer_id for channel {channel_id}: {transfer_id}" + ) + return transfer_id + except Exception as e: + logger.error( + f"[ARI org={self.organization_id}] Error getting transfer ID for channel {channel_id}: {e}" + ) + return None + + async def _handle_destination_answered( + self, transfer_id: str, destination_channel_id: str + ): + """Handle transfer destination channel answered - publish success event.""" + try: + logger.info( + f"[ARI Transfer org={self.organization_id}] Destination {destination_channel_id} " + f"answered for transfer {transfer_id}" + ) + + # Store channel mapping for potential future events and get transfer context + transfer_manager = await self._get_transfer_manager() + await transfer_manager.store_transfer_channel_mapping( + destination_channel_id, transfer_id + ) + context = await transfer_manager.get_transfer_context(transfer_id) + if not context: + logger.error( + f"[ARI Transfer org={self.organization_id}] No transfer context found for {transfer_id}" + ) + return + + logger.info( + f"[ARI Transfer org={self.organization_id}] Transfer {transfer_id} success: " + f"caller={context.original_call_sid} -> destination={destination_channel_id}" + ) + + # Publish destination answered event - this will trigger the bridge swap in serializer + success_event = TransferEvent( + type=TransferEventType.DESTINATION_ANSWERED, + transfer_id=transfer_id, + original_call_sid=context.original_call_sid, + transfer_call_sid=destination_channel_id, + conference_name=context.conference_name, + message="Transfer destination answered", + status="success", + action="destination_answered", + ) + await transfer_manager.publish_transfer_event(success_event) + + except Exception as e: + logger.error( + f"[ARI Transfer org={self.organization_id}] Error handling transfer answer: {e}" + ) + # On error, publish failure event + await self._handle_transfer_failed( + transfer_id, destination_channel_id, f"Transfer processing error: {e}" + ) + + async def _handle_transfer_failed( + self, transfer_id: str, channel_id: str, reason: str + ): + """Handle transfer failure - publish failure event.""" + try: + logger.info(f"[ARI Transfer] Transfer {transfer_id} failed: {reason}") + + transfer_manager = await self._get_transfer_manager() + context = await transfer_manager.get_transfer_context(transfer_id) + + # Publish failure event + failure_event = TransferEvent( + type=TransferEventType.TRANSFER_FAILED, + transfer_id=transfer_id, + original_call_sid=context.original_call_sid if context else "", + transfer_call_sid=channel_id, + message=f"Transfer failed: {reason}", + status="failed", + action="transfer_failed", + reason=reason, + ) + await transfer_manager.publish_transfer_event(failure_event) + + except Exception as e: + logger.error(f"[ARI Transfer] Error handling transfer failure: {e}") + async def _delete_channel(self, channel_id: str): """Delete (hang up) an ARI channel. Ignores 404 (already gone).""" diff --git a/api/services/telephony/call_transfer_manager.py b/api/services/telephony/call_transfer_manager.py index f843850..a6f0f8c 100644 --- a/api/services/telephony/call_transfer_manager.py +++ b/api/services/telephony/call_transfer_manager.py @@ -85,6 +85,28 @@ class CallTransferManager: except Exception as e: logger.error(f"Failed to remove transfer context: {e}") + async def store_transfer_channel_mapping( + self, channel_id: str, transfer_id: str + ) -> None: + """Store channel->transfer mapping in Redis for event correlation. + + Args: + channel_id: ARI channel ID + transfer_id: Transfer identifier + """ + try: + redis = await self._get_redis() + await redis.setex( + f"ari:transfer_channel:{channel_id}", 300, transfer_id + ) # 5 minute TTL + logger.debug( + f"[Transfer Manager] Stored channel mapping: channel={channel_id}, transfer_id={transfer_id}" + ) + except Exception as e: + logger.error( + f"[Transfer Manager] Error storing transfer channel mapping: {e}" + ) + async def publish_transfer_event(self, event: TransferEvent) -> None: """Publish transfer event to Redis channel. @@ -136,16 +158,10 @@ class CallTransferManager: ) # Check if this is a completion event - if ( - event.type - in [ - TransferEventType.TRANSFER_ANSWERED, # Call answered = transfer successful - TransferEventType.TRANSFER_COMPLETED, - TransferEventType.TRANSFER_FAILED, - TransferEventType.TRANSFER_CANCELLED, - TransferEventType.TRANSFER_TIMEOUT, - ] - ): + if event.type in [ + TransferEventType.DESTINATION_ANSWERED, + TransferEventType.TRANSFER_FAILED, + ]: return event except Exception as e: logger.error(f"Failed to parse transfer event: {e}") @@ -169,6 +185,31 @@ class CallTransferManager: except Exception as e: logger.error(f"Error closing pubsub connection: {e}") + async def find_transfer_context_for_call(self, caller_channel_id: str): + """Find the active transfer context for this caller channel.""" + + redis = await self._get_redis() + + try: + # Search Redis for transfer contexts where original_call_sid matches this caller + transfer_keys = await redis.keys("transfer:context:*") + + for key in transfer_keys: + try: + context_data = await redis.get(key) + if context_data: + context = TransferContext.from_json(context_data) + if context.original_call_sid == caller_channel_id: + return context + except Exception: + continue + + return None + + except Exception as e: + logger.error(f"[ARI Transfer] Error finding transfer context: {e}") + return None + async def cleanup(self): """Clean up Redis connections.""" try: diff --git a/api/services/telephony/providers/ari_call_strategies.py b/api/services/telephony/providers/ari_call_strategies.py new file mode 100644 index 0000000..5225a29 --- /dev/null +++ b/api/services/telephony/providers/ari_call_strategies.py @@ -0,0 +1,226 @@ +"""ARI-specific call operation strategies. + +This module contains the business logic for Asterisk ARI call operations. +""" + +from typing import Any, Dict + +from loguru import logger + +from pipecat.serializers.call_strategies import HangupStrategy, TransferStrategy + + +class ARIBridgeSwapStrategy(TransferStrategy): + """Implements bridge swap transfer for Asterisk ARI. + + This strategy handles transferring calls by swapping channels in existing + bridges, managing transfer contexts, and publishing + transfer completion events. + """ + + async def execute_transfer(self, context: Dict[str, Any]) -> bool: + """Execute bridge swap transfer for Asterisk ARI.""" + try: + import aiohttp + import redis.asyncio as aioredis + from aiohttp import BasicAuth + + channel_id = context["channel_id"] + ari_endpoint = context["ari_endpoint"] + app_name = context["app_name"] + app_password = context["app_password"] + + if not channel_id or not ari_endpoint: + logger.warning( + "Cannot execute transfer: missing channel_id or ari_endpoint" + ) + return False + + logger.info( + f"[ARI Transfer] Executing bridge swap for channel {channel_id}" + ) + + from api.constants import REDIS_URL + from api.db import db_client + from api.services.telephony.call_transfer_manager import ( + get_call_transfer_manager, + ) + auth = BasicAuth(app_name, app_password) + + # Get call transfer manager instance + call_transfer_manager = await get_call_transfer_manager() + + # 1. Find active transfer context for this caller channel + transfer_context = await call_transfer_manager.find_transfer_context_for_call(channel_id) + if not transfer_context: + logger.error( + f"[ARI Transfer] No active transfer context found for caller {channel_id}" + ) + return False + + logger.info( + f"[ARI Transfer] Found transfer context: {transfer_context.transfer_id}, " + f"destination: {transfer_context.call_sid}" + ) + + # 2. Get workflow run to find current bridge and external media channel + redis = aioredis.from_url(REDIS_URL, decode_responses=True) + workflow_run_id = await redis.get(f"ari:channel:{channel_id}") + if not workflow_run_id: + logger.error( + f"[ARI Transfer] No workflow run found for caller {channel_id}" + ) + return False + + workflow_run = await db_client.get_workflow_run_by_id(int(workflow_run_id)) + if not workflow_run or not workflow_run.gathered_context: + logger.error( + f"[ARI Transfer] No workflow context found for run {workflow_run_id}" + ) + return False + + ctx = workflow_run.gathered_context + bridge_id = ctx.get("bridge_id") + ext_channel_id = ctx.get("ext_channel_id") + + if not bridge_id or not ext_channel_id: + logger.error( + f"[ARI Transfer] Missing bridge/external channel info: {ctx}" + ) + return False + + destination_channel_id = transfer_context.call_sid + if not destination_channel_id: + logger.error( + f"[ARI Transfer] No destination channel in transfer context" + ) + return False + + logger.info( + f"[ARI Transfer] Bridge swap: bridge={bridge_id}, caller={channel_id}, " + f"destination={destination_channel_id}, ext_media={ext_channel_id}" + ) + + # 3. Set transfer state to prevent StasisEnd auto-teardown + workflow_run.gathered_context["transfer_state"] = "in-progress" + await db_client.update_workflow_run( + run_id=int(workflow_run_id), + gathered_context=workflow_run.gathered_context, + ) + logger.debug( + f"[ARI Transfer] Set transfer_state=in-progress for workflow {workflow_run_id}" + ) + + # 4. Execute bridge swap operations via ARI REST API + async with aiohttp.ClientSession() as session: + # Add destination channel to existing bridge + add_url = f"{ari_endpoint}/ari/bridges/{bridge_id}/addChannel" + async with session.post( + add_url, auth=auth, params={"channel": destination_channel_id} + ) as response: + if response.status in (200, 204): + logger.info( + f"[ARI Transfer] Added destination {destination_channel_id} to bridge {bridge_id}" + ) + else: + error_text = await response.text() + logger.error( + f"[ARI Transfer] Failed to add destination to bridge: {response.status} {error_text}" + ) + return False + + # Remove external media channel from bridge + remove_url = f"{ari_endpoint}/ari/bridges/{bridge_id}/removeChannel" + async with session.post( + remove_url, auth=auth, params={"channel": ext_channel_id} + ) as response: + if response.status in (200, 204): + logger.info( + f"[ARI Transfer] Removed external media {ext_channel_id} from bridge {bridge_id}" + ) + else: + error_text = await response.text() + logger.error( + f"[ARI Transfer] Failed to remove external media from bridge: {response.status} {error_text}" + ) + + # Hang up the external media channel + hangup_url = f"{ari_endpoint}/ari/channels/{ext_channel_id}" + async with session.delete(hangup_url, auth=auth) as response: + if response.status in (200, 204): + logger.info( + f"[ARI Transfer] Hung up external media channel {ext_channel_id}" + ) + elif response.status == 404: + logger.debug( + f"[ARI Transfer] External media channel {ext_channel_id} already gone" + ) + else: + error_text = await response.text() + logger.warning( + f"[ARI Transfer] Failed to hang up external media: {response.status} {error_text}" + ) + + logger.info( + f"[ARI Transfer] Bridge swap completed successfully for transfer {transfer_context.transfer_id}, " + f"caller {channel_id} connected to destination {destination_channel_id} via bridge {bridge_id}" + ) + + # 5. Clean up transfer context after successful completion + + call_transfer_manager = await get_call_transfer_manager() + await call_transfer_manager.remove_transfer_context( + transfer_context.transfer_id + ) + return True + + except Exception as e: + logger.exception(f"Failed to execute ARI transfer: {e}") + return False + +class ARIHangupStrategy(HangupStrategy): + """Implements hangup for Asterisk ARI channels.""" + + async def execute_hangup(self, context: Dict[str, Any]) -> bool: + """Hang up the Asterisk channel via ARI REST API.""" + try: + import aiohttp + from aiohttp import BasicAuth + + channel_id = context["channel_id"] + ari_endpoint = context["ari_endpoint"] + app_name = context["app_name"] + app_password = context["app_password"] + + if not channel_id or not ari_endpoint: + logger.warning( + "Cannot hang up Asterisk channel: missing channel_id or ari_endpoint" + ) + return False + + endpoint = f"{ari_endpoint}/ari/channels/{channel_id}" + auth = BasicAuth(app_name, app_password) + + async with aiohttp.ClientSession() as session: + async with session.delete(endpoint, auth=auth) as response: + if response.status in (200, 204): + logger.info( + f"Successfully terminated Asterisk channel {channel_id}" + ) + return True + elif response.status == 404: + logger.debug( + f"Asterisk channel {channel_id} was already terminated" + ) + return True + else: + error_text = await response.text() + logger.error( + f"Failed to terminate Asterisk channel {channel_id}: " + f"Status {response.status}, Response: {error_text}" + ) + return False + + except Exception as e: + logger.exception(f"Failed to hang up Asterisk channel: {e}") + return False \ No newline at end of file diff --git a/api/services/telephony/providers/ari_provider.py b/api/services/telephony/providers/ari_provider.py index 139065a..7d7fce1 100644 --- a/api/services/telephony/providers/ari_provider.py +++ b/api/services/telephony/providers/ari_provider.py @@ -349,8 +349,8 @@ class ARIProvider(TelephonyProvider): # ======== CALL TRANSFER METHODS ======== def supports_transfers(self) -> bool: - """ARI does not currently support call transfers.""" - return False + """ARI supports call transfers via bridge manipulation.""" + return True async def transfer_call( self, @@ -360,8 +360,104 @@ class ARIProvider(TelephonyProvider): timeout: int = 30, **kwargs: Any, ) -> Dict[str, Any]: - """ARI call transfers are not yet implemented.""" - raise NotImplementedError("ARI provider does not support call transfers") + """Initiate ARI call transfer by creating an outbound channel to the destination. + + This method creates the destination channel and returns immediately. The transfer + process completes asynchronously - success/failure is determined by ARI events + and communicated through the transfer event system. + + Args: + destination: Destination phone number (SIP endpoint) + transfer_id: Unique identifier for this transfer attempt + conference_name: Conference name (unused in ARI, kept for interface compatibility) + timeout: Transfer timeout in seconds + **kwargs: Additional arguments + + Returns: + Dict containing: + - call_sid: Destination channel ID + - status: "initiated" + - provider: "ari" + - raw_response: Full ARI channel creation response + """ + if not self.validate_config(): + raise ValueError("ARI provider not properly configured") + + logger.info( + f"[ARI Transfer] Initiating transfer {transfer_id} to {destination} " + f"(timeout: {timeout}s)" + ) + + from api.services.telephony.call_transfer_manager import ( + get_call_transfer_manager, + ) + + # Get call transfer manager for event correlation mapping + call_transfer_manager = await get_call_transfer_manager() + + # Build SIP endpoint + if destination.startswith("SIP/") or destination.startswith("PJSIP/"): + sip_endpoint = destination + else: + sip_endpoint = f"PJSIP/{destination}" + + # Build transfer appArgs for event correlation + app_args = f"transfer,{transfer_id}" + + try: + endpoint = f"{self.base_url}/channels" + params = { + "endpoint": sip_endpoint, + "app": self.app_name, + "appArgs": app_args, + "timeout": timeout, # Keep timeout for transfer calls + } + + async with aiohttp.ClientSession() as session: + async with session.post( + endpoint, + params=params, + auth=self._get_auth(), + ) as response: + response_text = await response.text() + + if response.status != 200: + error_msg = f"ARI channel creation failed: {response.status} {response_text}" + logger.error(f"[ARI Transfer] {error_msg}") + await call_transfer_manager.remove_transfer_context(transfer_id) + raise Exception(error_msg) + + result = json.loads(response_text) + + destination_channel_id = result.get("id", "") + if not destination_channel_id: + logger.error( + f"[ARI Transfer] Failed to get channel ID from response: {result}" + ) + await call_transfer_manager.remove_transfer_context(transfer_id) + raise Exception("Failed to create destination channel") + + # Store transfer channel mapping for event correlation + await call_transfer_manager.store_transfer_channel_mapping( + destination_channel_id, transfer_id + ) + + logger.info( + f"[ARI Transfer] Originated destination channel {destination_channel_id} " + f"for transfer {transfer_id}" + ) + + return { + "call_sid": destination_channel_id, + "status": "initiated", + "provider": self.PROVIDER_NAME, + "raw_response": result, + } + + except Exception as e: + logger.error(f"[ARI Transfer] Failed to originate call transfer destination channel: {e}") + await call_transfer_manager.remove_transfer_context(transfer_id) + raise # ======== ARI-SPECIFIC METHODS ======== diff --git a/api/services/telephony/providers/twilio_call_strategies.py b/api/services/telephony/providers/twilio_call_strategies.py new file mode 100644 index 0000000..d48cff3 --- /dev/null +++ b/api/services/telephony/providers/twilio_call_strategies.py @@ -0,0 +1,186 @@ +"""Twilio-specific call operation strategies. + +This module contains the business logic for Twilio call operations, +maintaining proper separation of concerns between protocol handling and business logic. +""" + +from typing import Any, Dict + +import aiohttp +from loguru import logger + +from pipecat.serializers.call_strategies import HangupStrategy, TransferStrategy + + +class TwilioConferenceStrategy(TransferStrategy): + """Implements conference-based call transfer for Twilio. + + This strategy transfers calls by placing them into a Twilio conference, + with cleanup of transfer contexts upon successful completion. + """ + + async def execute_transfer(self, context: Dict[str, Any]) -> bool: + """Execute conference transfer for Twilio call.""" + try: + account_sid = context["account_sid"] + auth_token = context["auth_token"] + call_sid = context["call_sid"] + region = context.get("region") + edge = context.get("edge") + + # 1. Find active transfer context for this call + transfer_context = await self._find_transfer_context_for_call(call_sid) + if not transfer_context: + logger.error( + f"[Twilio Transfer] No active transfer context found for call {call_sid}" + ) + return False + + logger.info( + f"[Twilio Transfer] Found transfer context: {transfer_context.transfer_id}, " + f"original: {transfer_context.original_call_sid}" + ) + + region_prefix = f"{region}." if region else "" + edge_prefix = f"{edge}." if edge else "" + + # Twilio API endpoint for updating calls + endpoint = f"https://api.{edge_prefix}{region_prefix}twilio.com/2010-04-01/Accounts/{account_sid}/Calls/{call_sid}.json" + + # Create basic auth from account_sid and auth_token + auth = aiohttp.BasicAuth(account_sid, auth_token) + + conference_name = transfer_context.conference_name + twiml = f""" + + + {conference_name} + +""" + + logger.debug( + f"[Twilio Transfer] Transferring call to conference: {conference_name}" + ) + + # 2. Make the POST request to transfer the call + async with aiohttp.ClientSession() as session: + async with session.post( + endpoint, auth=auth, data={"Twiml": twiml} + ) as response: + response_text = await response.text() + + if response.status == 200: + logger.info( + f"[Twilio Transfer] Conference transfer completed successfully for call {call_sid}, " + f"joined conference {conference_name}" + ) + + # 3. Clean up transfer context after successful transfer + await self._cleanup_transfer_context(transfer_context.transfer_id) + return True + elif response.status == 404: + logger.error( + f"Failed to transfer Twilio call {call_sid}: Call not found (404)" + ) + await self._cleanup_transfer_context(transfer_context.transfer_id) + return False + else: + logger.error( + f"Failed to transfer Twilio call {call_sid} to conference {conference_name}: " + f"Status {response.status}, Response: {response_text}" + ) + await self._cleanup_transfer_context(transfer_context.transfer_id) + return False + + except Exception as e: + logger.error(f"Failed to transfer Twilio call: {e}") + if transfer_context: + await self._cleanup_transfer_context(transfer_context.transfer_id) + return False + + async def _find_transfer_context_for_call(self, call_sid: str): + """Find the active transfer context for this call.""" + try: + import redis.asyncio as aioredis + + from api.constants import REDIS_URL + from api.services.telephony.transfer_event_protocol import TransferContext + + # Search Redis for transfer contexts where original_call_sid matches + redis = aioredis.from_url(REDIS_URL, decode_responses=True) + transfer_keys = await redis.keys("transfer:context:*") + + for key in transfer_keys: + try: + context_data = await redis.get(key) + if context_data: + context = TransferContext.from_json(context_data) + if context.original_call_sid == call_sid: + return context + except Exception: + continue + + return None + + except Exception as e: + logger.error(f"[Twilio Transfer] Error finding transfer context: {e}") + return None + + async def _cleanup_transfer_context(self, transfer_id: str): + """Clean up transfer context after completion or failure.""" + try: + from api.services.telephony.call_transfer_manager import ( + get_call_transfer_manager, + ) + + call_transfer_manager = await get_call_transfer_manager() + await call_transfer_manager.remove_transfer_context(transfer_id) + except Exception as e: + logger.error(f"[Twilio Transfer] Error cleaning up transfer context: {e}") + + +class TwilioHangupStrategy(HangupStrategy): + """Implements hangup for Twilio calls.""" + + async def execute_hangup(self, context: Dict[str, Any]) -> bool: + """Hang up the Twilio call via REST API.""" + try: + account_sid = context["account_sid"] + auth_token = context["auth_token"] + call_sid = context["call_sid"] + region = context.get("region") + edge = context.get("edge") + + if not account_sid or not auth_token or not call_sid: + logger.warning( + "Cannot hang up Twilio call: missing required credentials or call_sid" + ) + return False + + region_prefix = f"{region}." if region else "" + edge_prefix = f"{edge}." if edge else "" + + endpoint = f"https://api.{edge_prefix}{region_prefix}twilio.com/2010-04-01/Accounts/{account_sid}/Calls/{call_sid}.json" + auth = aiohttp.BasicAuth(account_sid, auth_token) + + async with aiohttp.ClientSession() as session: + async with session.post( + endpoint, auth=auth, data={"Status": "completed"} + ) as response: + if response.status == 200: + logger.info(f"Successfully terminated Twilio call {call_sid}") + return True + elif response.status == 404: + logger.debug(f"Twilio call {call_sid} was already terminated") + return True + else: + response_text = await response.text() + logger.error( + f"Failed to terminate Twilio call {call_sid}: " + f"Status {response.status}, Response: {response_text}" + ) + return False + + except Exception as e: + logger.exception(f"Failed to hang up Twilio call: {e}") + return False \ No newline at end of file diff --git a/api/services/telephony/transfer_event_protocol.py b/api/services/telephony/transfer_event_protocol.py index 6260676..7e7815e 100644 --- a/api/services/telephony/transfer_event_protocol.py +++ b/api/services/telephony/transfer_event_protocol.py @@ -13,12 +13,8 @@ from typing import Any, Dict, Optional class TransferEventType(str, Enum): """Types of transfer events sent between instances.""" - TRANSFER_INITIATED = "transfer_initiated" - TRANSFER_ANSWERED = "transfer_answered" - TRANSFER_COMPLETED = "transfer_completed" + DESTINATION_ANSWERED = "destination_answered" TRANSFER_FAILED = "transfer_failed" - TRANSFER_CANCELLED = "transfer_cancelled" - TRANSFER_TIMEOUT = "transfer_timeout" @dataclass @@ -56,7 +52,6 @@ class TransferEvent: "conference_id": self.conference_name, "transfer_call_sid": self.transfer_call_sid, "original_call_sid": self.original_call_sid, - "end_call": self.end_call, "reason": self.reason, } return result diff --git a/api/services/workflow/pipecat_engine_custom_tools.py b/api/services/workflow/pipecat_engine_custom_tools.py index f226791..c915fb8 100644 --- a/api/services/workflow/pipecat_engine_custom_tools.py +++ b/api/services/workflow/pipecat_engine_custom_tools.py @@ -318,7 +318,6 @@ class CustomToolManager: "message": "I'm sorry, but call transfers are not available for web calls. Please try a telephony call.", "action": "transfer_failed", "reason": "webrtc_not_supported", - "end_call": True, } await self._handle_transfer_result( webrtc_error_result, function_call_params, properties @@ -332,27 +331,46 @@ class CustomToolManager: "message": "I'm sorry, but I don't have a phone number configured for the transfer. Please contact support to set up call transfer.", "action": "transfer_failed", "reason": "no_destination", - "end_call": True, } await self._handle_transfer_result( validation_error_result, function_call_params, properties ) return - # Validate E.164 format - E164_PHONE_REGEX = r"^\+[1-9]\d{1,14}$" - if not re.match(E164_PHONE_REGEX, destination): - validation_error_result = { - "status": "failed", - "message": "I'm sorry, but the transfer phone number appears to be invalid. Please contact support to verify the transfer settings.", - "action": "transfer_failed", - "reason": "invalid_destination", - "end_call": True, - } - await self._handle_transfer_result( - validation_error_result, function_call_params, properties - ) - return + # Validate destination format based on workflow run mode + if workflow_run.mode == WorkflowRunMode.ARI.value: + # For ARI provider, also accept SIP endpoints + SIP_ENDPOINT_REGEX = r"^(PJSIP|SIP)\/[\w\-\.@]+$" + E164_PHONE_REGEX = r"^\+[1-9]\d{1,14}$" + + is_valid_sip = re.match(SIP_ENDPOINT_REGEX, destination) + is_valid_e164 = re.match(E164_PHONE_REGEX, destination) + + if not (is_valid_sip or is_valid_e164): + validation_error_result = { + "status": "failed", + "message": "I'm sorry, but the transfer destination appears to be invalid. Please contact support to verify the transfer settings.", + "action": "transfer_failed", + "reason": "invalid_destination", + } + await self._handle_transfer_result( + validation_error_result, function_call_params, properties + ) + return + else: + # For non-ARI providers (Twilio, etc), use E.164 validation + E164_PHONE_REGEX = r"^\+[1-9]\d{1,14}$" + if not re.match(E164_PHONE_REGEX, destination): + validation_error_result = { + "status": "failed", + "message": "I'm sorry, but the transfer phone number appears to be invalid. Please contact support to verify the transfer settings.", + "action": "transfer_failed", + "reason": "invalid_destination", + } + await self._handle_transfer_result( + validation_error_result, function_call_params, properties + ) + return if message_type == "custom" and custom_message: logger.info(f"Playing pre-transfer message: {custom_message}") @@ -366,14 +384,12 @@ class CustomToolManager: "message": "I'm sorry, there's an issue with this call transfer. Please contact support.", "action": "transfer_failed", "reason": "no_organization_id", - "end_call": False, } await self._handle_transfer_result( validation_error_result, function_call_params, properties ) return - # Get telephony provider directly (no HTTP round-trip) provider = await get_telephony_provider(organization_id) if not provider.supports_transfers() or not provider.validate_config(): validation_error_result = { @@ -381,7 +397,6 @@ class CustomToolManager: "message": "I'm sorry, there's an issue with this call transfer. Please contact support.", "action": "transfer_failed", "reason": "provider_does_not_support_transfer", - "end_call": False, } await self._handle_transfer_result( validation_error_result, function_call_params, properties @@ -396,6 +411,19 @@ class CustomToolManager: # Compute conference name from original call SID conference_name = f"transfer-{original_call_sid}" + # Store initial transfer context in Redis before provider call to avoid race condition + call_transfer_manager = await get_call_transfer_manager() + transfer_context = TransferContext( + transfer_id=transfer_id, + call_sid=None, # Will be updated after provider response + target_number=destination, + tool_uuid=tool.tool_uuid, + original_call_sid=original_call_sid, + conference_name=conference_name, + initiated_at=time.time(), + ) + await call_transfer_manager.store_transfer_context(transfer_context) + # Mute the pipeline self._engine.set_mute_pipeline(True) @@ -410,21 +438,8 @@ class CustomToolManager: call_sid = transfer_result.get("call_sid") logger.info(f"Transfer call initiated successfully: {call_sid}") - # TODO: Possible race here between saving the transfer context - # and getting a callback response from Twilio? Should we store_transfer_context - # before sending request to Twilio and update the transfer context afterwards? - - # Store transfer context in Redis - call_transfer_manager = await get_call_transfer_manager() - transfer_context = TransferContext( - transfer_id=transfer_id, - call_sid=call_sid, - target_number=destination, - tool_uuid=tool.tool_uuid, - original_call_sid=original_call_sid, - conference_name=conference_name, - initiated_at=time.time(), - ) + # Update transfer context with actual call_sid from provider response + transfer_context.call_sid = call_sid await call_transfer_manager.store_transfer_context(transfer_context) # Wait for status callback completion using Redis pub/sub @@ -466,15 +481,15 @@ class CustomToolManager: transfer_event = None finally: - # Single cleanup point: stop hold music, unmute pipeline, remove context + # Cleanup hold music and pipeline state + # Transfer context cleanup is handled by respective transfer call strategies logger.info( - "Transfer wait ended, cleaning up hold music, pipeline state, and transfer context" + "Transfer wait ended, cleaning up hold music and pipeline state" ) hold_music_stop_event.set() if hold_music_task: await hold_music_task self._engine.set_mute_pipeline(False) - await call_transfer_manager.remove_transfer_context(transfer_id) # Handle result (after cleanup) if transfer_event: @@ -491,7 +506,6 @@ class CustomToolManager: "message": "I'm sorry, but the call is taking longer than expected to connect. The person might not be available right now. Please try calling back later.", "action": "transfer_failed", "reason": "timeout", - "end_call": True, } await self._handle_transfer_result( timeout_result, function_call_params, properties @@ -509,7 +523,6 @@ class CustomToolManager: "message": "I'm sorry, but something went wrong while trying to transfer your call. Please try again later or contact support if the problem persists.", "action": "transfer_failed", "reason": "execution_error", - "end_call": True, } await self._handle_transfer_result( @@ -521,41 +534,52 @@ class CustomToolManager: async def _handle_transfer_result( self, result: dict, function_call_params, properties ): - """Handle different transfer call outcomes and take appropriate action.""" + """Handle transfer call outcomes from any telephony provider (Twilio, ARI, etc). + + This method is provider-agnostic and processes standardized result dictionaries + from transfer completion events, validation failures, timeouts, and errors. + + Args: + result: Standardized result dict with keys: action, status, reason, message + function_call_params: LLM function call parameters for response callback + properties: Function call result properties (e.g., run_llm setting) + """ action = result.get("action", "") status = result.get("status", "") logger.info(f"Handling transfer result: action={action}, status={status}") - if action == "transfer_success": - # Successful transfer - add original caller to conference and end pipeline + if action == "destination_answered": + # Transfer destination answered - proceeding with bridge swap/conference join conference_id = result.get("conference_id") original_call_sid = result.get("original_call_sid") transfer_call_sid = result.get("transfer_call_sid") logger.info( - f"Transfer successful! Conference: {conference_id}, Original: {original_call_sid}, Transfer: {transfer_call_sid}" + f"Transfer destination answered! Conference/Bridge: {conference_id}, " + f"Original: {original_call_sid}, Transfer: {transfer_call_sid}" ) - # Inform LLM of success and end the call with Transfer call reason + # Inform LLM of success and end the call (no further LLM processing needed) response_properties = FunctionCallResultProperties(run_llm=False) await function_call_params.result_callback( { "status": "transfer_success", - "message": "Transfer successful - connecting to conference", + "message": "Transfer destination answered - connecting calls", "conference_id": conference_id, }, properties=response_properties, ) + # End pipeline - providers complete bridge swap/conference join as final transfer leg await self._engine.end_call_with_reason( EndTaskReason.TRANSFER_CALL.value, abort_immediately=False ) elif action == "transfer_failed": - # Transfer failed - inform user via LLM and then end the call + # Transfer failed - let LLM inform user with error details reason = result.get("reason", "unknown") - logger.info(f"Transfer failed ({reason}), informing user") + logger.info(f"Transfer failed ({reason}), informing user via LLM") await function_call_params.result_callback( { diff --git a/docker-compose.yaml b/docker-compose.yaml index 5e7ce32..949b40a 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -211,4 +211,4 @@ volumes: networks: app-network: - driver: bridge + driver: bridge \ No newline at end of file diff --git a/docs/docs.json b/docs/docs.json index c6e8d42..37c8809 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -43,7 +43,24 @@ "voice-agent/introduction", "voice-agent/editing-a-workflow", "voice-agent/template-variables", - "voice-agent/tools", + { + "group": "Tools", + "pages": [ + { + "group": "Built-in Tools", + "pages": [ + "voice-agent/tools/call-transfer", + "voice-agent/tools/end-call" + ] + }, + { + "group": "Custom Tools", + "pages": [ + "voice-agent/tools/http-api" + ] + } + ] + }, { "group": "Nodes", "pages": [ diff --git a/docs/voice-agent/tools/call-transfer.mdx b/docs/voice-agent/tools/call-transfer.mdx new file mode 100644 index 0000000..064df3c --- /dev/null +++ b/docs/voice-agent/tools/call-transfer.mdx @@ -0,0 +1,57 @@ +--- +title: "Call Transfer" +description: "Enable your AI agent to transfer calls to phone numbers or SIP endpoints with built-in call transfer functionality." +--- + +The Call Transfer tool enables your AI agent to transfer active calls to phone numbers or SIP endpoints. When configured, your agent can seamlessly transfer callers to human operators, departments, or other systems while maintaining a professional experience. + +## Supported Providers + +Call transfer is available for telephony calls using Twilio or Asterisk ARI providers. Web calls do not support transfer functionality. + +## How It Works + +The Call Transfer tool performs **blind transfers** where no call context is shared with the destination. Here's what happens: + +1. **Agent Decision**: Your AI agent determines a transfer is needed and calls the transfer function +2. **Pre-transfer Message**: (Optional) Agent plays a custom message like "Let me transfer you to our sales team" +3. **Hold Experience**: Caller hears hold music while the transfer is processed +4. **Connection**: Once the destination answers, the caller is connected directly +5. **Agent Handoff**: The AI agent ends its involvement in the call + +## Configuration + +### Basic Settings + +- **Destination**: Phone number or SIP endpoint (see formats below) +- **Timeout**: How long to wait for destination to answer (default 30 seconds) +- **Pre-transfer Message**: Optional custom message played before transfer + +### Destination Formats + +**For Twilio:** +- **Phone numbers**: E.164 format: `+1234567890` +- Must be a valid reachable phone number + +**For Asterisk ARI:** +- **SIP endpoints only**: `PJSIP/extension` or `SIP/endpoint` +- **Examples**: `PJSIP/sales-queue`, `SIP/1001`, `PJSIP/conference-room` + + +Asterisk ARI transfers only work with SIP endpoints configured on your Asterisk server. External phone numbers require additional PSTN trunk configuration. + + +## Setup Requirements + +1. **Organization Setup**: Ensure your organization has a supported telephony provider configured +2. **Tool Enablement**: Add the Call Transfer tool to your agent's available tools +3. **Destination Validation**: + - **Twilio**: Verify phone numbers are reachable + - **ARI**: Verify SIP endpoints exist on your Asterisk server +4. **Testing**: Test transfers in your specific provider environment + +## Troubleshooting + +- **Destination not reachable**: Verify destination number/endpoint is valid and reachable +- **Tool not available**: Check that Call Transfer tool is added to the correct agent node +- **Transfer failures**: Handle transfer failure scenarios within your agent prompts \ No newline at end of file diff --git a/docs/voice-agent/tools/end-call.mdx b/docs/voice-agent/tools/end-call.mdx new file mode 100644 index 0000000..6586d43 --- /dev/null +++ b/docs/voice-agent/tools/end-call.mdx @@ -0,0 +1,37 @@ +--- +title: "End Call" +description: "Allow your AI agent to terminate calls when conversations reach a natural conclusion or specific conditions are met." +--- + +The End Call tool enables your AI agent to terminate calls programmatically when conversations reach a natural conclusion, specific conditions are met, or when instructed by the caller. + +## How It Works + +The End Call tool provides a clean way to terminate conversations: + +1. **Agent Decision**: Your AI agent determines the call should end based on conversation context +2. **Optional Message**: Agent can provide a closing message before ending the call +3. **Clean Termination**: Call is terminated gracefully through the telephony provider +4. **Call Disposition**: Call disposition is updated in the system + +## Configuration + +### Basic Settings + +- **End Message**: Optional closing message played before termination +- **Reason**: Optional reason code for call termination tracking + +## Agent Usage + +Your AI agent can end calls during conversations: + +``` +end_call(message="Thank you for calling. Have a great day!", reason="conversation_complete") +``` + +Common scenarios: +- Conversation has reached natural conclusion +- User explicitly requests to end the call +- Agent has completed all requested tasks +- Call timeout or error conditions + diff --git a/docs/voice-agent/tools.mdx b/docs/voice-agent/tools/http-api.mdx similarity index 81% rename from docs/voice-agent/tools.mdx rename to docs/voice-agent/tools/http-api.mdx index f330220..255c778 100644 --- a/docs/voice-agent/tools.mdx +++ b/docs/voice-agent/tools/http-api.mdx @@ -1,10 +1,9 @@ --- -title: "Tools" +title: "HTTP API" +description: "Create REST API tools that your AI agent can invoke during conversations to integrate with external systems." --- -Dograh AI lets you attach multiple tools (e.g. API calls) directly to workflow nodes. This allows your voice agents to call any internal or external system during a live conversation-based on LLM judgment and your prompts. - -This works similar to a tool call on any agentic platforms and is 100% open source and fully customizable. +HTTP API tools allow you to attach extrernal REST API calls directly to workflow nodes, enabling your voice agents to call any internal or external system during live conversations based on LLM judgment and your prompts. This works similar to function calling on any agentic platform and is 100% open source and fully customizable. ## Video Tutorial @@ -19,9 +18,9 @@ This works similar to a tool call on any agentic platforms and is 100% open sour allowFullScreen > -## What Are Tools? +## What is an HTTP API Tool? -A tool is a REST API definition that the LLM can invoke at runtime. +An HTTP API Tool is a REST API definition that the LLM can invoke at runtime. **Typical use cases:** - Call your own backend endpoints @@ -44,7 +43,7 @@ A tool is a REST API definition that the LLM can invoke at runtime.
-## Defining Tools +## Defining an HTTP API Tool ### 1. Tool Name @@ -103,7 +102,7 @@ Guidelines: Example: - interest (string): - “Set to true if the user clearly shows intent to buy or wants follow-up. Otherwise false.” + "Set to true if the user clearly shows intent to buy or wants follow-up. Otherwise false." ![Parameter Example](/images/tool%20params.png) @@ -121,7 +120,7 @@ Inside the node, guide the LLM using **simple English instructions**. Example: -“If the user shows interest in speaking to sales or wants a callback, immediately call the capture_lead_interest tool and set interest to true.” +"If the user shows interest in speaking to sales or wants a callback, immediately call the capture_lead_interest tool and set interest to true." This instruction is often the deciding factor for correct tool usage. This instruction is often the **deciding factor** for correct tool usage. @@ -156,4 +155,4 @@ Poor naming or vague descriptions lead to: - Use plain English in node instructions - Attach only relevant tools to each node -**Well-defined tools + clear prompts = reliable, production-grade voice agents.** +**Well-defined tools + clear prompts = reliable, production-grade voice agents.** \ No newline at end of file diff --git a/pipecat b/pipecat index 6aa0502..791d241 160000 --- a/pipecat +++ b/pipecat @@ -1 +1 @@ -Subproject commit 6aa0502a9834d536aba9589cec87d827e66f2fad +Subproject commit 791d24196f07b36d64beddda4a9c79aa6ccdb245 diff --git a/ui/package-lock.json b/ui/package-lock.json index 2862dc7..cc11b0f 100644 --- a/ui/package-lock.json +++ b/ui/package-lock.json @@ -1,12 +1,12 @@ { "name": "ui", - "version": "1.13.0", + "version": "1.15.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "ui", - "version": "1.13.0", + "version": "1.15.0", "dependencies": { "@dagrejs/dagre": "^1.1.4", "@hey-api/client-fetch": "^0.10.0", diff --git a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx index 067fad7..a37902d 100644 --- a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx +++ b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx @@ -1,5 +1,7 @@ "use client"; +import {useState } from "react"; + import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card"; import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; @@ -37,20 +39,46 @@ export function TransferCallToolConfig({ timeout, onTimeoutChange, }: TransferCallToolConfigProps) { - // Basic E.164 validation pattern + const [sipMode, setSipMode] = useState(() => /^(PJSIP|SIP)\//i.test(destination)); + + // Validation patterns const isValidPhoneNumber = (phone: string): boolean => { const e164Pattern = /^\+[1-9]\d{1,14}$/; return e164Pattern.test(phone); }; - const phoneNumberError = destination && !isValidPhoneNumber(destination); + const isValidSipEndpoint = (endpoint: string): boolean => { + const sipPattern = /^(PJSIP|SIP)\/[\w\-\.@]+$/i; + return sipPattern.test(endpoint); + }; + + const getValidationError = (): string | null => { + if (!destination) return null; + + if (sipMode) { + return isValidSipEndpoint(destination) + ? null + : "Please enter a valid SIP endpoint (e.g., PJSIP/1234 or SIP/extension@domain.com)"; + } else { + return isValidPhoneNumber(destination) + ? null + : "Please enter a valid phone number in E.164 format (e.g., +1234567890)"; + } + }; + + const destinationError = getValidationError(); + + const handleSipModeToggle = () => { + setSipMode(!sipMode); + onDestinationChange(""); // Clear destination when switching modes + }; return ( Transfer Call Configuration - Configure call transfer settings (Twilio only) + Configure call transfer settings. Supports phone numbers (Twilio) and SIP endpoints (Asterisk ARI). @@ -80,21 +108,31 @@ export function TransferCallToolConfig({
- + onDestinationChange(e.target.value)} - placeholder="+1234567890" - className={phoneNumberError ? "border-red-500 focus:border-red-500" : ""} + placeholder={sipMode ? "PJSIP/1234 or SIP/extension@domain.com" : "+1234567890"} + className={destinationError ? "border-red-500 focus:border-red-500" : ""} /> - {phoneNumberError && ( + {destinationError && ( )} +
diff --git a/ui/src/app/tools/[toolUuid]/page.tsx b/ui/src/app/tools/[toolUuid]/page.tsx index b7a83c9..97e0a13 100644 --- a/ui/src/app/tools/[toolUuid]/page.tsx +++ b/ui/src/app/tools/[toolUuid]/page.tsx @@ -198,10 +198,14 @@ export default function ToolDetailPage() { // Validation based on tool type if (tool.category === "transfer_call") { - // Validate destination phone number for Transfer Call tools + // Validate destination for Transfer Call tools (supports both E.164 and SIP endpoints) const e164Pattern = /^\+[1-9]\d{1,14}$/; - if (!transferDestination || !e164Pattern.test(transferDestination)) { - setError("Please enter a valid phone number in E.164 format (e.g., +1234567890)"); + const sipPattern = /^(PJSIP|SIP)\/[\w\-\.@]+$/i; + const isValidE164 = e164Pattern.test(transferDestination); + const isValidSip = sipPattern.test(transferDestination); + + if (!transferDestination || (!isValidE164 && !isValidSip)) { + setError("Please enter a valid phone number (E.164 format) or SIP endpoint (e.g., PJSIP/1234)"); return; } } else if (tool.category !== "end_call") {