transfer call

api/services/looptalk/audio_streamer.py

@@ -9,6 +9,7 @@ import asyncio
 from typing import Dict, Set
 
 from loguru import logger
+
 from pipecat.audio.utils import mix_audio
 from pipecat.frames.frames import (
     Frame,

api/services/pipecat/event_handlers.py

@@ -80,14 +80,29 @@ def register_event_handlers(
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(_transport, _participant):
         call_disposed = engine.is_call_disposed()
+        transfer_in_progress = getattr(engine, '_transfer_in_progress', False)
+        
+        logger.info(f"[TRANSFER-DEBUG] on_client_disconnected triggered")
+        logger.info(f"[TRANSFER-DEBUG] Engine instance ID in event_handler: {id(engine)}")
+        logger.info(f"[TRANSFER-DEBUG] Engine type in event_handler: {type(engine)}")
+        logger.info(f"[TRANSFER-DEBUG] transfer_in_progress attribute exists: {hasattr(engine, '_transfer_in_progress')}")
+        logger.info(f"[TRANSFER-DEBUG] transfer_in_progress value: {transfer_in_progress}")
+        logger.info(f"[TRANSFER-DEBUG] Call disposed: {call_disposed}")
+        
         logger.debug(
-            f"In on_client_disconnected callback handler. Call disposed: {call_disposed}"
+            f"In on_client_disconnected callback handler. Call disposed: {call_disposed}, "
+            f"Transfer in progress: {transfer_in_progress}"
         )
 
         # Stop recordings
         await audio_buffer.stop_recording()
 
-        # End the call
+        # Skip auto hang-up if transfer is in progress
+        if transfer_in_progress:
+            logger.info("Transfer in progress - skipping auto hang-up, letting redirect handle call")
+            return
+        logger.info("Transfer in progress - False, proceeding with hang up")
+        # Normal hang-up logic for non-transfer disconnections
         await engine.end_call_with_reason(
             EndTaskReason.USER_HANGUP.value, abort_immediately=True
         )

api/services/pipecat/pipeline_builder.py

@@ -71,6 +71,12 @@ def build_pipeline(
             user_context_aggregator,
             llm,  # LLM
             pipeline_engine_callback_processor,
+        ]
+    )
+    
+    
+    processors.extend(
+        [
             tts,  # TTS
             transport.output(),  # Transport bot output
             audio_buffer,  # AudioBufferProcessor - records both input and output audio

api/services/pipecat/run_pipeline.py

@@ -115,6 +115,7 @@ async def run_pipeline_twilio(
     # Create audio configuration for Twilio
     audio_config = create_audio_config(WorkflowRunMode.TWILIO.value)
 
+
     transport = await create_twilio_transport(
         websocket_client,
         stream_sid,
@@ -125,6 +126,7 @@ async def run_pipeline_twilio(
         vad_config,
         ambient_noise_config,
     )
+    
     await _run_pipeline(
         transport,
         workflow_id,
@@ -543,6 +545,7 @@ async def _run_pipeline(
 
     engine = PipecatEngine(
         llm=llm,
+        transport=transport,
         workflow=workflow_graph,
         call_context_vars=merged_call_context_vars,
         workflow_run_id=workflow_run_id,
@@ -552,7 +555,7 @@ async def _run_pipeline(
         embeddings_base_url=embeddings_base_url,
     )
 
-    # Create pipeline components with audio configuration
+    # Create pipeline components
     audio_buffer, context = create_pipeline_components(audio_config)
 
     # Set the context and audio_buffer after creation

api/services/pipecat/transport_setup.py

@@ -24,6 +24,7 @@ from pipecat.transports.websocket.fastapi import (
     FastAPIWebsocketParams,
     FastAPIWebsocketTransport,
 )
+from loguru import logger
 
 librnnoise_path = os.path.normpath(
     str(APP_ROOT_DIR / "native" / "rnnoise" / "librnnoise.so")

api/services/smart_turn/app.py

@@ -21,9 +21,10 @@ from fastapi import (
     status,
 )
 from fastapi.websockets import WebSocketState
-from pipecat.audio.turn.smart_turn.local_smart_turn_v2 import LocalSmartTurnAnalyzerV2
 from scipy.io import wavfile
 
+from pipecat.audio.turn.smart_turn.local_smart_turn_v2 import LocalSmartTurnAnalyzerV2
+
 LOG_LEVEL = (
     logging.DEBUG
     if os.environ.get("LOG_LEVEL", "DEBUG").lower() == "debug"

api/services/telephony/base.py

@@ -309,3 +309,45 @@ class TelephonyProvider(ABC):
             Tuple of (Response, media_type) - Response object and content type
         """
         pass
+
+    # ======== CALL TRANSFER METHODS ========
+
+    @abstractmethod
+    async def transfer_call(
+        self,
+        destination: str,
+        tool_call_id: str,
+        timeout: int = 30,
+        **kwargs: Any
+    ) -> Dict[str, Any]:
+        """
+        Initiate a call transfer to a destination number.
+
+        Args:
+            destination: The destination phone number (E.164 format)
+            tool_call_id: Unique identifier for tracking this transfer
+            timeout: Transfer timeout in seconds
+            **kwargs: Provider-specific additional parameters
+
+        Returns:
+            Dict containing:
+                - call_sid: Provider's call identifier
+                - status: Transfer initiation status
+                - provider: Provider name
+                - webhook_urls: Dict with answer and status callback URLs
+
+        Raises:
+            NotImplementedError: If provider doesn't support transfers
+            ValueError: If provider configuration is invalid
+        """
+        pass
+
+    @abstractmethod
+    def supports_transfers(self) -> bool:
+        """
+        Check if this provider supports call transfers.
+
+        Returns:
+            True if provider supports call transfers, False otherwise
+        """
+        pass

api/services/telephony/factory.py

@@ -128,3 +128,29 @@ async def get_all_telephony_providers() -> List[Type[TelephonyProvider]]:
         List of provider classes that can be used for webhook detection
     """
     return [CloudonixProvider, TwilioProvider, VobizProvider, VonageProvider]
+
+
+async def get_transfer_provider(organization_id: int) -> TelephonyProvider:
+    """
+    Get telephony provider that supports call transfers.
+
+    Args:
+        organization_id: Organization ID for provider lookup
+
+    Returns:
+        Configured telephony provider that supports transfers
+
+    Raises:
+        ValueError: If provider doesn't support transfers or org not configured
+    """
+    provider = await get_telephony_provider(organization_id)
+    
+    if not provider.supports_transfers():
+        config = await load_telephony_config(organization_id)
+        provider_type = config.get("provider", "unknown")
+        raise ValueError(
+            f"Organization telephony provider '{provider_type}' does not support call transfers. "
+            f"Only Twilio provider supports transfers."
+        )
+    
+    return provider

api/services/telephony/providers/cloudonix_provider.py

@@ -680,3 +680,29 @@ class CloudonixProvider(TelephonyProvider):
 </Response>"""
 
         return Response(content=twiml, media_type="application/xml"), "application/xml"
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        tool_call_id: str,
+        timeout: int = 30,
+        **kwargs: Any
+    ) -> Dict[str, Any]:
+        """
+        Cloudonix provider does not support call transfers.
+
+        Raises:
+            NotImplementedError: Always, as Cloudonix doesn't support transfers
+        """
+        raise NotImplementedError("Cloudonix provider does not support call transfers")
+
+    def supports_transfers(self) -> bool:
+        """
+        Cloudonix does not support call transfers.
+
+        Returns:
+            False - Cloudonix provider does not support call transfers
+        """
+        return False

api/services/telephony/providers/twilio_provider.py

@@ -10,6 +10,7 @@ import aiohttp
 from fastapi import HTTPException
 from loguru import logger
 from twilio.request_validator import RequestValidator
+from pipecat.utils.context import set_current_call_sid
 
 from api.enums import WorkflowRunMode
 from api.services.telephony.base import (
@@ -72,6 +73,7 @@ class TwilioProvider(TelephonyProvider):
         if from_number is None:
             from_number = random.choice(self.from_numbers)
         logger.info(f"Selected phone number {from_number} for outbound call")
+        logger.info(f"Webhook url received - {webhook_url}")
 
         # Prepare call data
         data = {"To": to_number, "From": from_number, "Url": webhook_url}
@@ -172,6 +174,7 @@ class TwilioProvider(TelephonyProvider):
     </Connect>
     <Pause length="40"/>
 </Response>"""
+        logger.info(f"Twiml content generated - {twiml_content}")
         return twiml_content
 
     async def get_call_cost(self, call_id: str) -> Dict[str, Any]:
@@ -280,6 +283,11 @@ class TwilioProvider(TelephonyProvider):
             try:
                 stream_sid = start_msg["start"]["streamSid"]
                 call_sid = start_msg["start"]["callSid"]
+                
+                # Set call SID in Pipecat context for use throughout the pipeline
+                set_current_call_sid(call_sid)
+                logger.info(f"Set call SID context: {call_sid}")
+                
             except KeyError:
                 logger.error("Missing streamSid or callSid in start message")
                 await websocket.close(code=4400, reason="Missing stream identifiers")
@@ -459,3 +467,111 @@ class TwilioProvider(TelephonyProvider):
 </Response>"""
 
         return Response(content=twiml_content, media_type="application/xml")
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        tool_call_id: str,
+        timeout: int = 30,
+        **kwargs: Any
+    ) -> Dict[str, Any]:
+        """
+        Initiate a call transfer via Twilio.
+
+        Args:
+            destination: The destination phone number (E.164 format)
+            tool_call_id: Unique identifier for tracking this transfer
+            timeout: Transfer timeout in seconds
+            **kwargs: Additional Twilio-specific parameters
+
+        Returns:
+            Dict containing transfer result information
+
+        Raises:
+            ValueError: If provider configuration is invalid
+            Exception: If Twilio API call fails
+        """
+        if not self.validate_config():
+            raise ValueError("Twilio provider not properly configured")
+
+        # Select a random phone number for the transfer
+        from_number = random.choice(self.from_numbers)
+        logger.info(f"Selected phone number {from_number} for transfer call")
+
+        backend_endpoint, _ = await get_backend_endpoints()
+
+        # Generate webhook URLs for the transfer call
+        call_url = f"{backend_endpoint}/api/v1/telephony/transfer-call-handler/{tool_call_id}"
+        status_callback_url = f"{backend_endpoint}/api/v1/telephony/transfer-result/{tool_call_id}"
+
+        logger.debug(f"Transfer webhook URLs - Answer: {call_url}, Status: {status_callback_url}")
+
+        # Prepare Twilio API call data
+        endpoint = f"{self.base_url}/Calls.json"
+        data = {
+            "To": destination,
+            "From": from_number,
+            "Timeout": timeout,
+            "Url": call_url,
+            "StatusCallback": status_callback_url,
+            "StatusCallbackEvent": ["answered", "no-answer", "busy", "failed", "completed"],
+            "StatusCallbackMethod": "POST"
+        }
+
+        # Add any additional kwargs
+        data.update(kwargs)
+
+        try:
+            # Make Twilio API call
+            logger.info(f"Making Twilio transfer API call to: {endpoint}")
+            logger.info(f"Transfer call data: {data}")
+
+            async with aiohttp.ClientSession() as session:
+                auth = aiohttp.BasicAuth(self.account_sid, self.auth_token)
+                async with session.post(endpoint, data=data, auth=auth) as response:
+                    response_status = response.status
+                    response_text = await response.text()
+                    
+                    logger.info(f"Twilio transfer API response status: {response_status}")
+                    logger.info(f"Twilio transfer API response body: {response_text}")
+
+                    if response_status in [200, 201]:
+                        try:
+                            response_data = await response.json()
+                            call_sid = response_data.get("sid")
+                            logger.info(f"Transfer call initiated successfully: {call_sid}")
+
+                            return {
+                                "call_sid": call_sid,
+                                "status": response_data.get("status", "queued"),
+                                "provider": self.PROVIDER_NAME,
+                                "webhook_urls": {
+                                    "answer": call_url,
+                                    "status": status_callback_url
+                                },
+                                "from_number": from_number,
+                                "to_number": destination,
+                                "raw_response": response_data
+                            }
+                        except Exception as e:
+                            logger.error(f"Failed to parse Twilio transfer response JSON: {e}")
+                            raise Exception(f"Failed to parse transfer response: {e}")
+                    else:
+                        error_msg = f"Twilio API call failed with status {response_status}: {response_text}"
+                        logger.error(error_msg)
+                        raise Exception(error_msg)
+
+        except Exception as e:
+            logger.error(f"Exception during Twilio transfer call: {e}")
+            raise
+
+    def supports_transfers(self) -> bool:
+        """
+        Twilio supports call transfers.
+
+        Returns:
+            True - Twilio provider supports call transfers
+        """
+        return True

api/services/telephony/providers/vobiz_provider.py

@@ -533,3 +533,29 @@ class VobizProvider(TelephonyProvider):
 </Response>"""
 
         return Response(content=vobiz_xml_content, media_type="application/xml")
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        tool_call_id: str,
+        timeout: int = 30,
+        **kwargs: Any
+    ) -> Dict[str, Any]:
+        """
+        Vobiz provider does not support call transfers.
+
+        Raises:
+            NotImplementedError: Always, as Vobiz doesn't support transfers
+        """
+        raise NotImplementedError("Vobiz provider does not support call transfers")
+
+    def supports_transfers(self) -> bool:
+        """
+        Vobiz does not support call transfers.
+
+        Returns:
+            False - Vobiz provider does not support call transfers
+        """
+        return False

api/services/telephony/providers/vonage_provider.py

@@ -484,3 +484,29 @@ class VonageProvider(TelephonyProvider):
         ]
 
         return Response(content=json.dumps(error_ncco), media_type="application/json")
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        tool_call_id: str,
+        timeout: int = 30,
+        **kwargs: Any
+    ) -> Dict[str, Any]:
+        """
+        Vonage provider does not support call transfers.
+
+        Raises:
+            NotImplementedError: Always, as Vonage doesn't support transfers
+        """
+        raise NotImplementedError("Vonage provider does not support call transfers")
+
+    def supports_transfers(self) -> bool:
+        """
+        Vonage does not support call transfers.
+
+        Returns:
+            False - Vonage provider does not support call transfers
+        """
+        return False

api/services/telephony/transfer_coordination.py

@@ -0,0 +1,204 @@
+"""Redis-based transfer coordination service for multi-instance scaling.
+
+Handles transfer event publishing, subscription, and context storage
+across multiple API server instances.
+"""
+
+import asyncio
+import time
+from typing import Optional, Dict, Any
+from loguru import logger
+import redis.asyncio as aioredis
+
+from api.constants import REDIS_URL
+from api.services.telephony.transfer_event_protocol import (
+    TransferEvent,
+    TransferContext,
+    TransferEventType,
+    TransferRedisChannels
+)
+
+
+class TransferCoordinator:
+    """Coordinates transfer events and context storage using Redis."""
+    
+    def __init__(self, redis_client: Optional[aioredis.Redis] = None):
+        self._redis_client = redis_client
+        self._pubsub_connections: Dict[str, aioredis.client.PubSub] = {}
+        
+    async def _get_redis(self) -> aioredis.Redis:
+        """Get Redis client instance."""
+        if not self._redis_client:
+            self._redis_client = await aioredis.from_url(
+                REDIS_URL, decode_responses=True
+            )
+        return self._redis_client
+        
+    async def store_transfer_context(self, context: TransferContext, ttl: int = 300) -> None:
+        """Store transfer context in Redis with TTL.
+        
+        Args:
+            context: Transfer context data
+            ttl: Time to live in seconds (default 5 minutes)
+        """
+        try:
+            redis = await self._get_redis()
+            key = TransferRedisChannels.transfer_context_key(context.tool_call_id)
+            await redis.setex(key, ttl, context.to_json())
+            logger.debug(f"Stored transfer context for {context.tool_call_id}")
+        except Exception as e:
+            logger.error(f"Failed to store transfer context: {e}")
+            
+    async def get_transfer_context(self, tool_call_id: str) -> Optional[TransferContext]:
+        """Retrieve transfer context from Redis.
+        
+        Args:
+            tool_call_id: Tool call identifier
+            
+        Returns:
+            Transfer context if found, None otherwise
+        """
+        try:
+            redis = await self._get_redis()
+            key = TransferRedisChannels.transfer_context_key(tool_call_id)
+            data = await redis.get(key)
+            if data:
+                return TransferContext.from_json(data)
+            return None
+        except Exception as e:
+            logger.error(f"Failed to get transfer context: {e}")
+            return None
+            
+    async def remove_transfer_context(self, tool_call_id: str) -> None:
+        """Remove transfer context from Redis.
+        
+        Args:
+            tool_call_id: Tool call identifier
+        """
+        try:
+            redis = await self._get_redis()
+            key = TransferRedisChannels.transfer_context_key(tool_call_id)
+            await redis.delete(key)
+            logger.debug(f"Removed transfer context for {tool_call_id}")
+        except Exception as e:
+            logger.error(f"Failed to remove transfer context: {e}")
+            
+    async def publish_transfer_event(self, event: TransferEvent) -> None:
+        """Publish transfer event to Redis channel.
+        
+        Args:
+            event: Transfer event to publish
+        """
+        try:
+            # Add timestamp if not present
+            if event.timestamp is None:
+                event.timestamp = time.time()
+                
+            redis = await self._get_redis()
+            channel = TransferRedisChannels.transfer_events(event.tool_call_id)
+            await redis.publish(channel, event.to_json())
+            logger.info(f"Published {event.type} event for {event.tool_call_id}")
+        except Exception as e:
+            logger.error(f"Failed to publish transfer event: {e}")
+            
+    async def wait_for_transfer_completion(
+        self, 
+        tool_call_id: str, 
+        timeout_seconds: float = 30.0
+    ) -> Optional[TransferEvent]:
+        """Wait for transfer completion event using Redis pub/sub.
+        
+        Args:
+            tool_call_id: Tool call identifier to wait for
+            timeout_seconds: Maximum time to wait
+            
+        Returns:
+            Transfer completion event if received, None on timeout
+        """
+        channel = TransferRedisChannels.transfer_events(tool_call_id)
+        redis = await self._get_redis()
+        pubsub = redis.pubsub()
+        
+        try:
+            await pubsub.subscribe(channel)
+            logger.info(f"Waiting for transfer completion on {channel} (timeout: {timeout_seconds}s)")
+            
+            # Wait for completion event with timeout
+            async def wait_for_message():
+                async for message in pubsub.listen():
+                    if message["type"] == "message":
+                        try:
+                            event = TransferEvent.from_json(message["data"])
+                            logger.info(f"Received {event.type} event for {tool_call_id}")
+                            
+                            # 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
+                            ]:
+                                return event
+                        except Exception as e:
+                            logger.error(f"Failed to parse transfer event: {e}")
+                            continue
+                return None
+            
+            # Wait with timeout
+            result = await asyncio.wait_for(wait_for_message(), timeout=timeout_seconds)
+            return result
+            
+        except asyncio.TimeoutError:
+            logger.error(f"Transfer completion wait timed out for {tool_call_id}")
+            # Publish timeout event for other instances
+            timeout_event = TransferEvent(
+                type=TransferEventType.TRANSFER_TIMEOUT,
+                tool_call_id=tool_call_id,
+                workflow_run_id=0,  # Will be updated by caller
+                original_call_sid="",
+                status="failed",
+                reason="timeout",
+                end_call=True
+            )
+            await self.publish_transfer_event(timeout_event)
+            return None
+        except Exception as e:
+            logger.error(f"Error waiting for transfer completion: {e}")
+            return None
+        finally:
+            try:
+                await pubsub.unsubscribe(channel)
+                await pubsub.close()
+            except Exception as e:
+                logger.error(f"Error closing pubsub connection: {e}")
+                
+    async def cleanup(self):
+        """Clean up Redis connections."""
+        try:
+            # Close pubsub connections
+            for pubsub in self._pubsub_connections.values():
+                try:
+                    await pubsub.close()
+                except:
+                    pass
+            self._pubsub_connections.clear()
+            
+            # Close main Redis connection
+            if self._redis_client:
+                await self._redis_client.close()
+                self._redis_client = None
+        except Exception as e:
+            logger.error(f"Error during transfer coordinator cleanup: {e}")
+
+
+# Global transfer coordinator instance
+_transfer_coordinator: Optional[TransferCoordinator] = None
+
+
+async def get_transfer_coordinator() -> TransferCoordinator:
+    """Get or create the global transfer coordinator instance."""
+    global _transfer_coordinator
+    if not _transfer_coordinator:
+        _transfer_coordinator = TransferCoordinator()
+    return _transfer_coordinator

api/services/telephony/transfer_event_protocol.py

@@ -0,0 +1,105 @@
+"""Redis communication protocol for transfer coordination.
+
+Defines event formats and Redis channels for coordinating call transfers
+across multiple API server instances.
+"""
+
+import json
+from dataclasses import asdict, dataclass
+from enum import Enum
+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"
+    TRANSFER_FAILED = "transfer_failed"
+    TRANSFER_CANCELLED = "transfer_cancelled"
+    TRANSFER_TIMEOUT = "transfer_timeout"
+
+
+@dataclass
+class TransferEvent:
+    """Event data structure for transfer coordination."""
+    
+    type: TransferEventType
+    tool_call_id: str
+    workflow_run_id: int
+    original_call_sid: str
+    transfer_call_sid: Optional[str] = None
+    target_number: Optional[str] = None
+    conference_name: Optional[str] = None
+    message: Optional[str] = None
+    status: Optional[str] = None
+    action: Optional[str] = None
+    reason: Optional[str] = None
+    end_call: bool = False
+    timestamp: Optional[float] = None
+    
+    def to_json(self) -> str:
+        """Convert event to JSON string."""
+        return json.dumps(asdict(self))
+    
+    @classmethod 
+    def from_json(cls, data: str) -> "TransferEvent":
+        """Create event from JSON string."""
+        return cls(**json.loads(data))
+    
+    def to_result_dict(self) -> Dict[str, Any]:
+        """Convert to function call result format."""
+        result = {
+            "status": self.status or "success",
+            "message": self.message or "",
+            "action": self.action or self.type,
+            "conference_id": self.conference_name,
+            "transfer_call_sid": self.transfer_call_sid,
+            "original_call_sid": self.original_call_sid,
+            "caller_number": None,  # Will be populated by webhook handler
+            "end_call": self.end_call,
+            "reason": self.reason
+        }
+        return result
+
+
+@dataclass
+class TransferContext:
+    """Transfer context data stored in Redis."""
+    
+    tool_call_id: str
+    call_sid: Optional[str]
+    target_number: str
+    tool_uuid: str
+    original_call_sid: str
+    caller_number: Optional[str]
+    initiated_at: float
+    workflow_run_id: int
+    
+    def to_json(self) -> str:
+        """Convert context to JSON string."""
+        return json.dumps(asdict(self))
+    
+    @classmethod
+    def from_json(cls, data: str) -> "TransferContext":
+        """Create context from JSON string.""" 
+        return cls(**json.loads(data))
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary."""
+        return asdict(self)
+
+
+class TransferRedisChannels:
+    """Redis channel naming conventions for transfer events."""
+    
+    @staticmethod
+    def transfer_events(tool_call_id: str) -> str:
+        """Channel for transfer events for a specific tool call."""
+        return f"transfer:events:{tool_call_id}"
+    
+    @staticmethod
+    def transfer_context_key(tool_call_id: str) -> str:
+        """Redis key for transfer context storage."""
+        return f"transfer:context:{tool_call_id}"

api/services/telephony/transfer_test.py

@@ -0,0 +1,135 @@
+"""Test utilities for transfer coordination.
+
+This module provides utilities to test Redis-based transfer coordination
+across multiple instances.
+"""
+
+import asyncio
+import time
+import uuid
+from typing import Optional
+
+from loguru import logger
+
+from api.services.telephony.transfer_coordination import get_transfer_coordinator
+from api.services.telephony.transfer_event_protocol import (
+    TransferContext,
+    TransferEvent,
+    TransferEventType
+)
+
+
+async def test_redis_coordination():
+    """Test basic Redis pub/sub coordination for transfers."""
+    logger.info("Testing Redis-based transfer coordination...")
+    
+    transfer_coordinator = await get_transfer_coordinator()
+    
+    # Test 1: Store and retrieve transfer context
+    tool_call_id = str(uuid.uuid4())
+    test_context = TransferContext(
+        tool_call_id=tool_call_id,
+        call_sid="test_call_123",
+        target_number="+1234567890",
+        tool_uuid="test_tool_uuid",
+        original_call_sid="original_call_123",
+        caller_number="+0987654321",
+        initiated_at=time.time(),
+        workflow_run_id=123
+    )
+    
+    logger.info("Test 1: Storing transfer context...")
+    await transfer_coordinator.store_transfer_context(test_context)
+    
+    logger.info("Test 1: Retrieving transfer context...")
+    retrieved_context = await transfer_coordinator.get_transfer_context(tool_call_id)
+    
+    if retrieved_context and retrieved_context.tool_call_id == tool_call_id:
+        logger.info("✅ Test 1 PASSED: Context storage/retrieval works")
+    else:
+        logger.error("❌ Test 1 FAILED: Context storage/retrieval failed")
+        return False
+    
+    # Test 2: Event publishing and waiting
+    logger.info("Test 2: Testing event publishing...")
+    
+    # Start waiting for completion in background
+    async def wait_for_completion():
+        return await transfer_coordinator.wait_for_transfer_completion(tool_call_id, 5.0)
+    
+    wait_task = asyncio.create_task(wait_for_completion())
+    
+    # Give it a moment to start waiting
+    await asyncio.sleep(0.5)
+    
+    # Publish completion event
+    test_event = TransferEvent(
+        type=TransferEventType.TRANSFER_COMPLETED,
+        tool_call_id=tool_call_id,
+        workflow_run_id=123,
+        original_call_sid="original_call_123",
+        transfer_call_sid="transfer_call_456",
+        conference_name="test-conference",
+        message="Test transfer completed successfully",
+        status="success",
+        action="transfer_success"
+    )
+    
+    logger.info("Test 2: Publishing completion event...")
+    await transfer_coordinator.publish_transfer_event(test_event)
+    
+    # Wait for the completion
+    received_event = await wait_task
+    
+    if received_event and received_event.tool_call_id == tool_call_id:
+        logger.info("✅ Test 2 PASSED: Event pub/sub works")
+    else:
+        logger.error("❌ Test 2 FAILED: Event pub/sub failed")
+        return False
+    
+    # Test 3: Cleanup
+    logger.info("Test 3: Testing cleanup...")
+    await transfer_coordinator.remove_transfer_context(tool_call_id)
+    
+    cleanup_context = await transfer_coordinator.get_transfer_context(tool_call_id)
+    if cleanup_context is None:
+        logger.info("✅ Test 3 PASSED: Cleanup works")
+    else:
+        logger.error("❌ Test 3 FAILED: Cleanup failed")
+        return False
+    
+    logger.info("✅ All tests PASSED! Redis coordination is working correctly.")
+    return True
+
+
+async def test_timeout_handling():
+    """Test timeout handling in transfer coordination."""
+    logger.info("Testing timeout handling...")
+    
+    transfer_coordinator = await get_transfer_coordinator()
+    tool_call_id = str(uuid.uuid4())
+    
+    # Wait for completion with short timeout (should timeout)
+    start_time = time.time()
+    result = await transfer_coordinator.wait_for_transfer_completion(tool_call_id, 2.0)
+    elapsed = time.time() - start_time
+    
+    if result is None and elapsed >= 2.0:
+        logger.info("✅ Timeout test PASSED: Properly timed out after 2 seconds")
+        return True
+    else:
+        logger.error(f"❌ Timeout test FAILED: Expected timeout, got {result} in {elapsed}s")
+        return False
+
+
+if __name__ == "__main__":
+    async def main():
+        success1 = await test_redis_coordination()
+        success2 = await test_timeout_handling()
+        
+        if success1 and success2:
+            logger.info("🎉 All transfer coordination tests PASSED!")
+        else:
+            logger.error("💥 Some tests FAILED!")
+    
+    asyncio.run(main())

api/services/workflow/pipecat_engine.py

@@ -112,6 +112,9 @@ class PipecatEngine:
         self._embeddings_api_key: Optional[str] = embeddings_api_key
         self._embeddings_model: Optional[str] = embeddings_model
         self._embeddings_base_url: Optional[str] = embeddings_base_url
+        
+        # Transfer state tracking - prevents auto hang-up during call transfers
+        self._transfer_in_progress: bool = False
 
     async def _get_organization_id(self) -> Optional[int]:
         """Get and cache the organization ID from workflow run."""
@@ -207,14 +210,14 @@ class PipecatEngine:
             )
             logger.info(f"Arguments: {function_call_params.arguments}")
 
-            # Perform variable extraction before transitioning to new node
-            await self._perform_variable_extraction_if_needed(self._current_node)
-
-            # Set context for the new node, so that when the function call result
-            # frame is received by LLMContextAggregator and an LLM generation
-            # is done, we have updated context and functions
-            await self.set_node(transition_to_node)
             try:
+                # Perform variable extraction before transitioning to new node
+                await self._perform_variable_extraction_if_needed(self._current_node)
+
+                # Set context for the new node, so that when the function call result
+                # frame is received by LLMContextAggregator and an LLM generation
+                # is done, we have updated context and functions
+                await self.set_node(transition_to_node)
 
                 async def on_context_updated() -> None:
                     """
@@ -245,6 +248,7 @@ class PipecatEngine:
                 await function_call_params.result_callback(
                     result, properties=properties
                 )
+                
             except Exception as e:
                 logger.error(f"Error in transition function {name}: {str(e)}")
                 error_result = {"status": "error", "error": str(e)}
@@ -277,6 +281,7 @@ class PipecatEngine:
         async def calculate_func(function_call_params: FunctionCallParams) -> None:
             logger.info(f"LLM Function Call EXECUTED: safe_calculator")
             logger.info(f"Arguments: {function_call_params.arguments}")
+            
             try:
                 expr = function_call_params.arguments.get("expression", "")
                 result = safe_calculator(expr)
@@ -292,6 +297,7 @@ class PipecatEngine:
         ) -> None:
             logger.info(f"LLM Function Call EXECUTED: get_current_time")
             logger.info(f"Arguments: {function_call_params.arguments}")
+            
             try:
                 timezone = function_call_params.arguments.get("timezone", "UTC")
                 result = get_current_time(timezone)
@@ -302,6 +308,7 @@ class PipecatEngine:
         async def convert_time_func(function_call_params: FunctionCallParams) -> None:
             logger.info(f"LLM Function Call EXECUTED: convert_time")
             logger.info(f"Arguments: {function_call_params.arguments}")
+            
             try:
                 result = convert_time(
                     function_call_params.arguments.get("source_timezone"),
@@ -332,6 +339,7 @@ class PipecatEngine:
         async def retrieve_kb_func(function_call_params: FunctionCallParams) -> None:
             logger.info("LLM Function Call EXECUTED: retrieve_from_knowledge_base")
             logger.info(f"Arguments: {function_call_params.arguments}")
+            
             try:
                 query = function_call_params.arguments.get("query", "")
                 organization_id = await self._get_organization_id()
@@ -532,7 +540,7 @@ class PipecatEngine:
             self._current_node, run_in_background=False
         )
 
-        frame_to_push = CancelFrame() if abort_immediately else EndFrame()
+        frame_to_push = CancelFrame(reason=reason) if abort_immediately else EndFrame(reason=reason)
 
         # Apply disposition mapping - first try call_disposition if it is,
         # extracted from the call conversation then fall back to reason
@@ -705,6 +713,18 @@ class PipecatEngine:
                 f"Stasis connection set for immediate transfers: {connection.channel_id}"
             )
 
+    def set_mute_pipeline(self, mute: bool) -> None:
+        """Set the pipeline mute state.
+
+        This controls whether user input should be muted via the CallbackUserMuteStrategy.
+        When muted, the user's audio input will be blocked.
+
+        Args:
+            mute: True to mute user input, False to allow input
+        """
+        logger.debug(f"Setting pipeline mute state to: {mute}")
+        self._mute_pipeline = mute
+
     async def handle_llm_text_frame(self, text: str):
         """Accumulate LLM text frames to build reference text."""
         self._current_llm_generation_reference_text += text

api/services/workflow/pipecat_engine_custom_tools.py

@@ -6,8 +6,11 @@ during workflow execution.
 
 from __future__ import annotations
 
+import asyncio
+import re
 from typing import TYPE_CHECKING, Any, Optional
 
+import aiohttp
 from loguru import logger
 
 from api.db import db_client
@@ -21,9 +24,24 @@ from api.services.workflow.tools.custom_tool import (
     tool_to_function_schema,
 )
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.frames.frames import FunctionCallResultProperties, TTSSpeakFrame
+from pipecat.frames.frames import (
+    FunctionCallResultProperties,
+    TTSSpeakFrame,
+    OutputAudioRawFrame,
+)
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.utils.enums import EndTaskReason
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketClient
+
+from api.utils.hold_audio import load_hold_audio
+from api.services.telephony.transfer_coordination import get_transfer_coordinator
+from api.services.telephony.transfer_event_protocol import (
+    TransferEvent,
+    TransferContext,
+    TransferEventType,
+)
+
+from dograh.api.utils.common import get_backend_endpoints
 
 if TYPE_CHECKING:
     from api.services.workflow.pipecat_engine import PipecatEngine
@@ -138,6 +156,8 @@ class CustomToolManager:
         """
         if tool.category == ToolCategory.END_CALL.value:
             return self._create_end_call_handler(tool, function_name)
+        elif tool.category == ToolCategory.TRANSFER_CALL.value:
+            return self._create_transfer_call_handler(tool, function_name)
 
         return self._create_http_tool_handler(tool, function_name)
 
@@ -230,3 +250,505 @@ class CustomToolManager:
                 )
 
         return end_call_handler
+
+    def _create_transfer_call_handler(self, tool: Any, function_name: str):
+        """Create a handler function for a transfer call tool.
+
+        Args:
+            tool: The ToolModel instance
+            function_name: The function name used by the LLM
+
+        Returns:
+            Async handler function for the transfer call tool
+        """
+        # Don't run LLM after starting transfer - we're handling async response
+        properties = FunctionCallResultProperties(run_llm=False)
+
+        async def transfer_call_handler(
+            function_call_params: FunctionCallParams,
+        ) -> None:
+            logger.info(f"Transfer Call Tool EXECUTED: {function_name}")
+            logger.info(f"Arguments: {function_call_params.arguments}")
+
+            try:
+                # Get the transfer call configuration
+                config = tool.definition.get("config", {})
+                destination = config.get("destination", "")
+                message_type = config.get("messageType", "none")
+                custom_message = config.get("customMessage", "")
+                timeout_seconds = config.get(
+                    "timeout", 30
+                )  # Default 30 seconds if not configured
+
+                # Validate destination phone number
+                if not destination or not destination.strip():
+                    validation_error_result = {
+                        "status": "failed",
+                        "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_pattern = r"^\+[1-9]\d{1,14}$"
+                if not re.match(e164_pattern, 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
+
+                # Provider validation handled by telephony endpoint
+
+                # Note: User muting and hold music are handled automatically by
+
+                # Play pre-transfer message if configured
+                if message_type == "custom" and custom_message:
+                    logger.info(f"Playing pre-transfer message: {custom_message}")
+                    await self._engine.task.queue_frame(TTSSpeakFrame(custom_message))
+
+                # Get original call information from Pipecat context
+                from pipecat.utils.context import get_current_call_sid
+
+                original_call_sid = get_current_call_sid()
+                caller_number = None  # Skip caller number for now as requested
+
+                logger.info(f"Found original call context: call_id={original_call_sid}")
+
+                # Get organization ID for provider configuration
+                organization_id = await self.get_organization_id()
+                if not organization_id:
+                    validation_error_result = {
+                        "status": "failed",
+                        "message": "I'm sorry, but I can't determine which organization this call belongs to. 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
+
+                # Prepare transfer request data
+                transfer_data = {
+                    "destination": destination,
+                    "organization_id": organization_id,  # Required for provider configuration
+                    "tool_call_id": function_call_params.tool_call_id,  # Use LLM's tool call ID for pipeline coordination
+                    "tool_uuid": tool.tool_uuid,  # Add tool UUID for tracing and validation
+                    "original_call_sid": original_call_sid,  # Original caller's call SID
+                    "caller_number": caller_number,  # Original caller's phone number
+                }
+
+                # Initialize Redis-based transfer coordination
+                import httpx
+                import time
+
+                # Get backend endpoint URL
+                backend_url, _ = await get_backend_endpoints()
+
+                # Get transfer coordinator for Redis-based coordination
+                transfer_coordinator = await get_transfer_coordinator()
+
+                # Now initiate the transfer call
+                transfer_url = f"{backend_url}/api/v1/telephony/call-transfer"
+
+                async with httpx.AsyncClient(timeout=30.0) as client:
+                    response = await client.post(
+                        transfer_url,
+                        json=transfer_data,
+                        headers={"Content-Type": "application/json"},
+                        # Authentication headers added by provider if needed
+                    )
+
+                    if response.status_code == 200:
+                        result_data = response.json()
+                        logger.info(f"Transfer initiated successfully: {result_data}")
+
+                        # Wait for webhook completion using standard Pipecat async pattern
+                        logger.info(
+                            f"Transfer call initiated for {destination}, waiting for webhook completion..."
+                        )
+
+                        # Start hold music during transfer waiting period
+                        hold_music_stop_event = asyncio.Event()
+                        hold_music_task = None
+
+                        try:
+                            # Mute the pipeline to prevent further LLM generations during transfer
+                            logger.info("Muting pipeline during transfer call")
+                            self._engine.set_mute_pipeline(True)
+
+                            # Determine sample rate from transport (default to 8000Hz for Twilio)
+                            sample_rate = 8000
+                            if hasattr(self._engine.transport, "output") and hasattr(
+                                self._engine.transport.output(), "sample_rate"
+                            ):
+                                sample_rate = getattr(
+                                    self._engine.transport.output(), "sample_rate", 8000
+                                )
+
+                            logger.info(
+                                f"Starting hold music at {sample_rate}Hz while waiting for transfer"
+                            )
+
+                            # Start hold music as background task
+                            hold_music_task = asyncio.create_task(
+                                self.play_hold_music_loop(
+                                    hold_music_stop_event, sample_rate
+                                )
+                            )
+
+                            # Wait for transfer completion using Redis pub/sub
+                            logger.info(
+                                "Waiting for transfer completion via Redis pub/sub..."
+                            )
+                            transfer_event = (
+                                await transfer_coordinator.wait_for_transfer_completion(
+                                    transfer_data["tool_call_id"], timeout_seconds
+                                )
+                            )
+
+                            # Stop hold music and unmute pipeline
+                            logger.info(
+                                "Transfer completed, stopping hold music and unmuting pipeline"
+                            )
+                            hold_music_stop_event.set()
+                            if hold_music_task:
+                                await hold_music_task
+                            self._engine.set_mute_pipeline(False)
+
+                            if transfer_event:
+                                # Get result from transfer event
+                                final_result = transfer_event.to_result_dict()
+
+                                # Get transfer context for caller number
+                                transfer_context = (
+                                    await transfer_coordinator.get_transfer_context(
+                                        transfer_data["tool_call_id"]
+                                    )
+                                )
+                                if transfer_context and transfer_context.caller_number:
+                                    final_result["caller_number"] = (
+                                        transfer_context.caller_number
+                                    )
+
+                                # Handle the transfer result and inform user appropriately
+                                await self._handle_transfer_result(
+                                    final_result, function_call_params, properties
+                                )
+                            else:
+                                # Handle timeout case
+                                logger.error(
+                                    f"Transfer call timed out after {timeout_seconds} seconds"
+                                )
+
+                                # Create timeout result and handle it through the same flow
+                                timeout_result = {
+                                    "status": "failed",
+                                    "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
+                                )
+
+                        except Exception as e:
+                            logger.error(f"Error during transfer wait: {e}")
+
+                            # Stop hold music and unmute pipeline on error
+                            logger.info(
+                                "Transfer error, stopping hold music and unmuting pipeline"
+                            )
+                            hold_music_stop_event.set()
+                            if hold_music_task:
+                                await hold_music_task
+                            self._engine.set_mute_pipeline(False)
+
+                            # Handle error case
+                            error_result = {
+                                "status": "failed",
+                                "message": "I'm sorry, but there was an issue processing the transfer. Please try again.",
+                                "action": "transfer_failed",
+                                "reason": "system_error",
+                                "end_call": True,
+                            }
+                            await self._handle_transfer_result(
+                                error_result, function_call_params, properties
+                            )
+
+                    else:
+                        error_data = (
+                            response.json()
+                            if response.content
+                            else {"error": "Unknown error"}
+                        )
+                        logger.error(
+                            f"Transfer initiation failed: {response.status_code} - {error_data}"
+                        )
+
+                        # No cleanup needed for Redis-based coordination
+
+                        # Handle initiation failure with user-friendly message
+                        initiation_failure_result = {
+                            "status": "failed",
+                            "message": "I'm sorry, but I'm having trouble setting up the call transfer right now. There might be a technical issue. Please try again later or contact support.",
+                            "action": "transfer_failed",
+                            "reason": "initiation_failed",
+                            "end_call": True,
+                        }
+
+                        await self._handle_transfer_result(
+                            initiation_failure_result, function_call_params, properties
+                        )
+
+            except httpx.TimeoutException:
+                logger.error(f"Transfer call '{function_name}' HTTP request timed out")
+
+                # No cleanup needed for Redis-based coordination
+
+                # Handle HTTP timeout with user-friendly message
+                http_timeout_result = {
+                    "status": "failed",
+                    "message": "I'm sorry, but there seems to be a network issue preventing me from setting up the call transfer. Please try again in a moment.",
+                    "action": "transfer_failed",
+                    "reason": "network_timeout",
+                    "end_call": True,
+                }
+
+                await self._handle_transfer_result(
+                    http_timeout_result, function_call_params, properties
+                )
+
+            except Exception as e:
+                logger.error(
+                    f"Transfer call tool '{function_name}' execution failed: {e}"
+                )
+
+                # No cleanup needed for Redis-based coordination
+
+                # Handle generic exception with user-friendly message
+                exception_result = {
+                    "status": "failed",
+                    "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(
+                    exception_result, function_call_params, properties
+                )
+
+        return transfer_call_handler
+
+    async def _handle_transfer_result(
+        self, result: dict, function_call_params, properties
+    ):
+        """Handle different transfer call outcomes and take appropriate action."""
+        action = result.get("action", "")
+        status = result.get("status", "")
+        message = result.get("message", "")
+        should_end_call = result.get("end_call", False)
+
+        logger.info(f"Handling transfer result: action={action}, status={status}")
+
+        if action == "transfer_success":
+            # Successful transfer - add original caller to conference and end pipeline
+            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}"
+            )
+
+            # First inform LLM of success (but don't continue call)
+            response_properties = FunctionCallResultProperties(
+                run_llm=False
+            )  # We'll handle the transfer ourselves
+            await function_call_params.result_callback(
+                {
+                    "status": "transfer_success",
+                    "message": "Transfer successful - connecting to conference",
+                    "conference_id": conference_id,
+                },
+                properties=response_properties,
+            )
+
+            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
+            reason = result.get("reason", "unknown")
+            logger.info(f"Transfer failed ({reason}), informing user and ending call")
+
+            # Use system message pattern to direct LLM response for transfer failure
+            # This is more reliable than function call results
+            from pipecat.frames.frames import LLMMessagesAppendFrame
+
+            # Create system message with clear instructions for transfer failure
+            failure_instruction = {
+                "role": "system",
+                "content": f"IMPORTANT: The transfer call has FAILED. Reason: {reason}. You must inform the customer about this failure using this message: '{message}' Then immediately say goodbye and end the conversation. Do NOT ask if they need anything else or continue the conversation. Do NOT continue with transfer language.",
+            }
+
+            # Push the system message to LLM context
+            await self._engine.task.queue_frame(
+                LLMMessagesAppendFrame([failure_instruction], run_llm=True)
+            )
+
+            # Also send the function call result for consistency
+            response_properties = FunctionCallResultProperties(
+                run_llm=False
+            )  # LLM will be triggered by system message
+            await function_call_params.result_callback(
+                {"status": "transfer_failed", "reason": reason, "message": message},
+                properties=response_properties,
+            )
+
+            # Set appropriate disposition for analytics
+            disposition_map = {
+                "no_answer": "transfer_no_answer",
+                "busy": "transfer_busy",
+                "call_failed": "transfer_failed",
+                "timeout": "transfer_timeout",
+                "no_destination": "transfer_config_error",
+                "invalid_destination": "transfer_config_error",
+                "initiation_failed": "transfer_system_error",
+                "network_timeout": "transfer_system_error",
+                "execution_error": "transfer_system_error",
+            }
+
+            disposition = disposition_map.get(reason, "transfer_failed")
+            logger.info(
+                f"Setting disposition: {disposition} for transfer failure reason: {reason}"
+            )
+
+            # Give the LLM time to speak the message, then end the call with disposition
+            # We'll schedule the end call after a brief delay to allow TTS
+            logger.info("Scheduling call end after LLM delivers failure message")
+
+            # Import here to avoid circular dependencies
+            import asyncio
+
+            # Schedule call end after 3 seconds to allow LLM to speak
+            async def delayed_end_call():
+                import asyncio
+
+                await asyncio.sleep(3)
+                await self._engine.end_call_with_reason(
+                    f"transfer_failed_{reason}",  # Include specific reason in end reason
+                    abort_immediately=False,  # Allow any queued speech to complete
+                )
+
+            # Create task to end call asynchronously
+            asyncio.create_task(delayed_end_call())
+
+        elif action == "transfer_completed":
+            # This should no longer happen since we ignore "completed" status in webhook
+            # to avoid overriding successful transfers
+            logger.warning(
+                "Received unexpected 'transfer_completed' action - this should be ignored by webhook now"
+            )
+            logger.warning(
+                "If you see this message, there might be an issue with the webhook status filtering"
+            )
+
+            # For safety, treat it as a generic result without ending the call
+            await function_call_params.result_callback(result, properties=properties)
+
+        else:
+            # Unknown action, treat as generic success
+            logger.warning(f"Unknown transfer action: {action}, treating as success")
+            await function_call_params.result_callback(result, properties=properties)
+
+    async def play_hold_music_loop(
+        self, stop_event: asyncio.Event, sample_rate: int = 8000
+    ):
+        """Play hold music in a loop until stop event is triggered.
+
+        Args:
+            stop_event: Event to stop the hold music loop
+            sample_rate: Sample rate for the hold music (default 8000Hz for Twilio)
+        """
+        try:
+            import os
+
+            # Path to hold music file based on sample rate
+            assets_dir = os.path.join(
+                os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "assets"
+            )
+
+            # Select appropriate hold music file
+            if sample_rate == 16000:
+                hold_music_file = os.path.join(
+                    assets_dir, "transfer_hold_ring_16000.wav"
+                )
+            else:  # Default to 8000Hz for Twilio
+                hold_music_file = os.path.join(
+                    assets_dir, "transfer_hold_ring_8000.wav"
+                )
+
+            logger.info(f"Starting hold music loop with file: {hold_music_file}")
+
+            # Load hold music audio data
+            hold_audio_data = load_hold_audio(hold_music_file, sample_rate)
+            if not hold_audio_data:
+                logger.error("Failed to load hold music data")
+                return
+
+            # Convert bytes to audio frames - each frame should be about 20ms worth of audio
+            # For 8000Hz: 20ms = 160 samples = 320 bytes (16-bit)
+            # For 16000Hz: 20ms = 320 samples = 640 bytes (16-bit)
+            frame_size = 320 if sample_rate == 8000 else 640
+
+            audio_data = hold_audio_data
+            total_length = len(audio_data)
+            position = 0
+
+            logger.info(
+                f"Hold music loaded: {total_length} bytes, frame size: {frame_size}"
+            )
+
+            while not stop_event.is_set():
+                # Extract audio chunk
+                if position + frame_size > total_length:
+                    # Reached end of audio, loop back to beginning
+                    position = 0
+
+                audio_chunk = audio_data[position : position + frame_size]
+                position += frame_size
+
+                # Create audio frame
+                audio_frame = OutputAudioRawFrame(
+                    audio=audio_chunk,
+                    sample_rate=sample_rate,
+                    num_channels=1,
+                )
+
+                # Queue the frame
+                await self._engine.task.queue_frame(audio_frame)
+
+                # Sleep for frame duration (20ms)
+                await asyncio.sleep(0.02)
+
+            logger.info("Hold music loop stopped")
+
+        except Exception as e:
+            logger.error(f"Error in hold music loop: {e}")