feat: telephony call transfer (#155)

* transfer call

* fix: ignore completed call status

* chore: refactor telephony

* chore: refactor pipecat engine custom tools and other telephony services

* chore: code refactor

* chore: put back office ambient sound files

* chore: remove transport from engine

* fix: fix alembic revision

* chore: remove set_transferring_call from engine

* fix: send OutputAudio frame and let transport chunk it

* fix: reinstate docker compose

* chore: remove unused transfer-twmil route for caller

* chore: update pipecat submodule

---------

Co-authored-by: Abhishek Kumar <abhishek@a6k.me>

api/alembic/versions/1a7d74d54e8f_add_transfer_call_tool_category.py

@@ -0,0 +1,50 @@
+"""add transfer_call tool category
+
+Revision ID: 1a7d74d54e8f
+Revises: 34c8537dfde5
+Create Date: 2026-02-03 11:18:11.417837
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+from alembic_postgresql_enum import TableReference
+
+# revision identifiers, used by Alembic.
+revision: str = "1a7d74d54e8f"
+down_revision: Union[str, None] = "34c8537dfde5"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.sync_enum_values(
+        enum_schema="public",
+        enum_name="tool_category",
+        new_values=["http_api", "end_call", "transfer_call", "native", "integration"],
+        affected_columns=[
+            TableReference(
+                table_schema="public", table_name="tools", column_name="category"
+            )
+        ],
+        enum_values_to_rename=[],
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.sync_enum_values(
+        enum_schema="public",
+        enum_name="tool_category",
+        new_values=["http_api", "end_call", "native", "integration"],
+        affected_columns=[
+            TableReference(
+                table_schema="public", table_name="tools", column_name="category"
+            )
+        ],
+        enum_values_to_rename=[],
+    )
+    # ### end Alembic commands ###

api/enums.py

@@ -122,7 +122,8 @@ class ToolCategory(Enum):
 
     HTTP_API = "http_api"  # Custom HTTP API calls (implemented)
     END_CALL = "end_call"  # End call tool
-    NATIVE = "native"  # Built-in integrations (future: call_transfer, dtmf_input)
+    TRANSFER_CALL = "transfer_call"  # Transfer call to phone number (Twilio only)
+    NATIVE = "native"  # Built-in integrations (future: dtmf_input)
     INTEGRATION = "integration"  # Third-party integrations (future: Google Calendar, Salesforce, etc.)
 
 

api/routes/telephony.py

@@ -8,9 +8,16 @@ import uuid
 from datetime import UTC, datetime
 from typing import Optional
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Request, WebSocket
+from fastapi import (
+    APIRouter,
+    Depends,
+    Header,
+    HTTPException,
+    Request,
+    WebSocket,
+)
 from loguru import logger
-from pydantic import BaseModel
+from pydantic import BaseModel, field_validator
 from sqlalchemy import text
 from sqlalchemy.future import select
 from starlette.responses import HTMLResponse
@@ -26,10 +33,15 @@ from api.services.auth.depends import get_user
 from api.services.campaign.campaign_call_dispatcher import campaign_call_dispatcher
 from api.services.campaign.campaign_event_publisher import get_campaign_event_publisher
 from api.services.quota_service import check_dograh_quota, check_dograh_quota_by_user_id
+from api.services.telephony.call_transfer_manager import get_call_transfer_manager
 from api.services.telephony.factory import (
     get_all_telephony_providers,
     get_telephony_provider,
 )
+from api.services.telephony.transfer_event_protocol import (
+    TransferEvent,
+    TransferEventType,
+)
 from api.utils.common import get_backend_endpoints
 from api.utils.telephony_helper import (
     generic_hangup_response,
@@ -157,7 +169,8 @@ async def initiate_call(
     if not phone_number:
         raise HTTPException(
             status_code=400,
-            detail="Phone number must be provided in request or set in user configuration",
+            detail="Phone number must be provided in request or set in user "
+            "configuration",
         )
 
     workflow_run_id = request.workflow_run_id
@@ -1480,3 +1493,210 @@ async def handle_cloudonix_cdr(request: Request):
     )
 
     return {"status": "success"}
+
+
+class TransferCallRequest(BaseModel):
+    """Request model for initiating a call transfer."""
+
+    destination: str  # E.164 format phone number (required)
+    organization_id: int  # Organization ID for provider configuration
+    transfer_id: str  # Unique identifier for tracking this transfer
+    conference_name: str  # Conference name for the transfer
+    timeout: Optional[int] = 20  # seconds to wait for answer
+
+    @field_validator("destination")
+    @classmethod
+    def validate_destination(cls, destination: str) -> str:
+        """Validate destination is in E.164 format."""
+        import re
+
+        if not destination or not destination.strip():
+            raise ValueError("Destination phone number is required")
+
+        E164_PHONE_REGEX = r"^\+[1-9]\d{1,14}$"
+        if not re.match(E164_PHONE_REGEX, destination.strip()):
+            raise ValueError(
+                f"Invalid phone number format: {destination}. Must be E.164 format (e.g., +1234567890)"
+            )
+
+        return destination.strip()
+
+
+@router.post("/call-transfer")
+async def initiate_call_transfer(request: TransferCallRequest):
+    """Initiate a call transfer via the telephony provider.
+
+    This endpoint only initiates the outbound call. Transfer context
+    (original_call_sid, etc.) is stored by the caller
+    before invoking this endpoint.
+    """
+    logger.info(
+        f"Starting call transfer to {request.destination} with transfer_id: {request.transfer_id}"
+    )
+
+    try:
+        try:
+            provider = await get_telephony_provider(request.organization_id)
+        except ValueError as e:
+            logger.error(f"Transfer provider validation failed: {e}")
+            raise HTTPException(
+                status_code=400, detail=f"Call transfer not supported: {str(e)}"
+            )
+
+        if not provider.supports_transfers():
+            raise HTTPException(
+                status_code=400,
+                detail=f"Provider '{provider.PROVIDER_NAME}' does not support call transfers",
+            )
+
+        if not provider.validate_config():
+            logger.error(f"Provider {provider.PROVIDER_NAME} configuration is invalid")
+            raise HTTPException(
+                status_code=400,
+                detail=f"Telephony provider '{provider.PROVIDER_NAME}' is not properly configured for transfers",
+            )
+
+        logger.info(f"Initiating transfer call via {provider.PROVIDER_NAME} provider")
+        try:
+            transfer_result = await provider.transfer_call(
+                destination=request.destination,
+                transfer_id=request.transfer_id,
+                conference_name=request.conference_name,
+                timeout=request.timeout,
+            )
+        except NotImplementedError as e:
+            logger.error(
+                f"Provider {provider.PROVIDER_NAME} doesn't support transfers: {e}"
+            )
+            raise HTTPException(
+                status_code=400,
+                detail=f"Provider '{provider.PROVIDER_NAME}' does not support call transfers",
+            )
+        except Exception as e:
+            logger.error(f"Provider transfer call failed: {e}")
+            raise HTTPException(
+                status_code=500, detail=f"Transfer call failed: {str(e)}"
+            )
+
+        call_sid = transfer_result.get("call_sid")
+        logger.info(f"Transfer call initiated successfully: {call_sid}")
+        logger.debug(f"Transfer result: {transfer_result}")
+
+        return {
+            "status": "transfer_initiated",
+            "call_id": call_sid,
+            "message": f"Calling {request.destination}...",
+            "transfer_id": request.transfer_id,
+            "provider": provider.PROVIDER_NAME,
+        }
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Unexpected error during transfer call: {e}")
+        raise HTTPException(
+            status_code=500, detail=f"Internal error during transfer: {str(e)}"
+        )
+
+
+@router.post("/transfer-result/{transfer_id}")
+async def complete_transfer_function_call(transfer_id: str, request: Request):
+    """Webhook endpoint to complete the function call with transfer result.
+
+    Called by Twilio's StatusCallback when the transfer call status changes.
+    """
+    form_data = await request.form()
+    data = dict(form_data)
+
+    call_status = data.get("CallStatus", "")
+    call_sid = data.get("CallSid", "")
+
+    logger.info(
+        f"Transfer result(call status) webhook: {transfer_id} status={call_status}"
+    )
+
+    # Get transfer context from Redis for additional information
+    call_transfer_manager = await get_call_transfer_manager()
+    transfer_context = await call_transfer_manager.get_transfer_context(transfer_id)
+
+    original_call_sid = transfer_context.original_call_sid if transfer_context else None
+    conference_name = transfer_context.conference_name if transfer_context else None
+
+    # Determine the result based on call status with user-friendly messaging
+    if call_status in ("in-progress", "answered"):
+        result = {
+            "status": "success",
+            "message": "Great! The destination number answered. Let me transfer you now.",
+            "action": "transfer_success",
+            "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
+            "end_call": False,  # Continue with transfer
+        }
+    elif call_status == "no-answer":
+        result = {
+            "status": "transfer_failed",
+            "reason": "no_answer",
+            "message": "The transfer call was not answered. The person may be busy or unavailable right now.",
+            "action": "transfer_failed",
+            "call_sid": call_sid,
+            "end_call": True,
+        }
+    elif call_status == "busy":
+        result = {
+            "status": "transfer_failed",
+            "reason": "busy",
+            "message": "The transfer call encountered a busy signal. The person is likely on another call.",
+            "action": "transfer_failed",
+            "call_sid": call_sid,
+            "end_call": True,
+        }
+    elif call_status == "failed":
+        result = {
+            "status": "transfer_failed",
+            "reason": "call_failed",
+            "message": "The transfer call failed to connect. There may be a network issue or the number is unavailable.",
+            "action": "transfer_failed",
+            "call_sid": call_sid,
+            "end_call": True,
+        }
+    else:
+        # Intermediate status (ringing, in-progress, etc.), don't complete yet
+        logger.info(
+            f"Received intermediate status {call_status}, waiting for final status"
+        )
+        return {"status": "pending"}
+
+    # Complete the function call with Redis event publishing
+    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
+        else:
+            event_type = TransferEventType.TRANSFER_FAILED
+
+        transfer_event = TransferEvent(
+            type=event_type,
+            transfer_id=transfer_id,
+            original_call_sid=original_call_sid or "",
+            transfer_call_sid=call_sid,
+            conference_name=conference_name,
+            message=result.get("message", ""),
+            status=result["status"],
+            action=result.get("action", ""),
+            reason=result.get("reason"),
+            end_call=result.get("end_call", False),
+        )
+
+        # Publish the event via Redis
+        await call_transfer_manager.publish_transfer_event(transfer_event)
+        logger.info(
+            f"Published {event_type} event for {transfer_id} with result: {result['status']}"
+        )
+
+    except Exception as e:
+        logger.error(f"Error completing transfer {transfer_id}: {e}")
+
+    return {"status": "completed", "result": result}

api/routes/tool.py

@@ -1,10 +1,11 @@
 """API routes for managing tools."""
 
+import re
 from datetime import datetime
 from typing import Annotated, Any, Dict, List, Literal, Optional, Union
 
 from fastapi import APIRouter, Depends, HTTPException
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 
 from api.db import db_client
 from api.db.models import UserModel
@@ -56,6 +57,42 @@ class EndCallConfig(BaseModel):
     )
 
 
+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)"
+    )
+    messageType: Literal["none", "custom"] = Field(
+        default="none", description="Type of message to play before transfer"
+    )
+    customMessage: Optional[str] = Field(
+        default=None, description="Custom message to play before transferring the call"
+    )
+    timeout: int = Field(
+        default=30,
+        ge=5,
+        le=120,
+        description="Maximum time in seconds to wait for destination to answer (5-120 seconds)",
+    )
+
+    @field_validator("destination")
+    @classmethod
+    def validate_destination(cls, v: str) -> str:
+        """Validate that destination is a valid E.164 phone number."""
+        # 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):
+            raise ValueError(
+                "Destination must be a valid E.164 phone number (e.g., +1234567890)"
+            )
+        return v
+
+
 class HttpApiToolDefinition(BaseModel):
     """Tool definition for HTTP API tools."""
 
@@ -72,9 +109,17 @@ class EndCallToolDefinition(BaseModel):
     config: EndCallConfig = Field(description="End Call configuration")
 
 
+class TransferCallToolDefinition(BaseModel):
+    """Tool definition for Transfer Call tools."""
+
+    schema_version: int = Field(default=1, description="Schema version")
+    type: Literal["transfer_call"] = Field(description="Tool type")
+    config: TransferCallConfig = Field(description="Transfer Call configuration")
+
+
 # Union type for tool definitions - Pydantic will discriminate based on 'type' field
 ToolDefinition = Annotated[
-    Union[HttpApiToolDefinition, EndCallToolDefinition],
+    Union[HttpApiToolDefinition, EndCallToolDefinition, TransferCallToolDefinition],
     Field(discriminator="type"),
 ]
 
@@ -89,6 +134,17 @@ class CreateToolRequest(BaseModel):
     icon_color: Optional[str] = Field(default="#3B82F6", max_length=7)
     definition: ToolDefinition
 
+    @field_validator("category")
+    @classmethod
+    def validate_category(cls, v: str) -> str:
+        """Validate that category is a valid ToolCategory value."""
+        valid_categories = [c.value for c in ToolCategory]
+        if v not in valid_categories:
+            raise ValueError(
+                f"Invalid category '{v}'. Must be one of: {', '.join(valid_categories)}"
+            )
+        return v
+
 
 class UpdateToolRequest(BaseModel):
     """Request schema for updating a tool."""

api/services/configuration/registry.py

@@ -278,7 +278,15 @@ class DograhTTSService(BaseTTSConfiguration):
 
 
 SARVAM_TTS_MODELS = ["bulbul:v2", "bulbul:v3"]
-SARVAM_V2_VOICES = ["anushka", "manisha", "vidya", "arya", "abhilash", "karun", "hitesh"]
+SARVAM_V2_VOICES = [
+    "anushka",
+    "manisha",
+    "vidya",
+    "arya",
+    "abhilash",
+    "karun",
+    "hitesh",
+]
 SARVAM_V3_VOICES = [
     "shubh",
     "aditya",

api/services/pipecat/event_handlers.py

@@ -80,6 +80,7 @@ def register_event_handlers(
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(_transport, _participant):
         call_disposed = engine.is_call_disposed()
+
         logger.debug(
             f"In on_client_disconnected callback handler. Call disposed: {call_disposed}"
         )
@@ -87,7 +88,6 @@ def register_event_handlers(
         # Stop recordings
         await audio_buffer.stop_recording()
 
-        # End the call
         await engine.end_call_with_reason(
             EndTaskReason.USER_HANGUP.value, abort_immediately=True
         )

api/services/pipecat/run_pipeline.py

@@ -552,11 +552,12 @@ 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
+    # Set the context, audio_config, and audio_buffer after creation
     engine.set_context(context)
+    engine.set_audio_config(audio_config)
 
     # Set Stasis connection for immediate transfers (if available)
     if stasis_connection:

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,46 @@ 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,
+        transfer_id: str,
+        conference_name: 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)
+            transfer_id: Unique identifier for tracking this transfer
+            conference_name: Name of the conference to join the destination into
+            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
+
+        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/call_transfer_manager.py

@@ -0,0 +1,200 @@
+"""Redis-based transfer event coordination service
+
+Handles transfer event publishing, subscription, and context storage
+"""
+
+import asyncio
+import time
+from typing import Dict, Optional
+
+import redis.asyncio as aioredis
+from loguru import logger
+
+from api.constants import REDIS_URL
+from api.services.telephony.transfer_event_protocol import (
+    TransferContext,
+    TransferEvent,
+    TransferEventType,
+    TransferRedisChannels,
+)
+
+
+class CallTransferManager:
+    """Manages call 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.transfer_id)
+            await redis.setex(key, ttl, context.to_json())
+            logger.debug(f"Stored transfer context for {context.transfer_id}")
+        except Exception as e:
+            logger.error(f"Failed to store transfer context: {e}")
+
+    async def get_transfer_context(self, transfer_id: str) -> Optional[TransferContext]:
+        """Retrieve transfer context from Redis.
+
+        Args:
+            transfer_id: Transfer identifier
+
+        Returns:
+            Transfer context if found, None otherwise
+        """
+        try:
+            redis = await self._get_redis()
+            key = TransferRedisChannels.transfer_context_key(transfer_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, transfer_id: str) -> None:
+        """Remove transfer context from Redis.
+
+        Args:
+            transfer_id: Transfer identifier
+        """
+        try:
+            redis = await self._get_redis()
+            key = TransferRedisChannels.transfer_context_key(transfer_id)
+            await redis.delete(key)
+            logger.debug(f"Removed transfer context for {transfer_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.transfer_id)
+            await redis.publish(channel, event.to_json())
+            logger.info(f"Published {event.type} event for {event.transfer_id}")
+        except Exception as e:
+            logger.error(f"Failed to publish transfer event: {e}")
+
+    async def wait_for_transfer_completion(
+        self, transfer_id: str, timeout_seconds: float = 30.0
+    ) -> Optional[TransferEvent]:
+        """Wait for transfer completion event using Redis pub/sub.
+
+        Args:
+            transfer_id: Transfer identifier to wait for
+            timeout_seconds: Maximum time to wait
+
+        Returns:
+            Transfer completion event if received, None on timeout
+        """
+        channel = TransferRedisChannels.transfer_events(transfer_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 {transfer_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.debug(f"Transfer completion wait timed out for {transfer_id}")
+            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 call transfer manager instance
+_call_transfer_manager: Optional[CallTransferManager] = None
+
+
+async def get_call_transfer_manager() -> CallTransferManager:
+    """Get or create the global call transfer manager instance."""
+    global _call_transfer_manager
+    if not _call_transfer_manager:
+        _call_transfer_manager = CallTransferManager()
+    return _call_transfer_manager

api/services/telephony/providers/cloudonix_provider.py

@@ -680,3 +680,30 @@ class CloudonixProvider(TelephonyProvider):
 </Response>"""
 
         return Response(content=twiml, media_type="application/xml"), "application/xml"
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        transfer_id: str,
+        conference_name: str,
+        timeout: int = 30,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Cloudonix provider does not support call transfers.
+
+        Raises:
+            NotImplementedError: Cloudonix call transfers are yet to be implemented
+        """
+        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

@@ -72,6 +72,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 +173,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]:
@@ -459,3 +461,129 @@ class TwilioProvider(TelephonyProvider):
 </Response>"""
 
         return Response(content=twiml_content, media_type="application/xml")
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        transfer_id: str,
+        conference_name: str,
+        timeout: int = 30,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Initiate a call transfer via Twilio.
+
+        Uses inline TwiML to put the destination into a conference when they answer,
+        and a status callback to track the transfer outcome.
+
+        Args:
+            destination: The destination phone number (E.164 format)
+            transfer_id: Unique identifier for tracking this transfer
+            conference_name: Name of the conference to join the destination into
+            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()
+
+        status_callback_url = (
+            f"{backend_endpoint}/api/v1/telephony/transfer-result/{transfer_id}"
+        )
+
+        # Inline TwiML: when the destination answers, put them into the conference
+        twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
+<Response>
+    <Say>You have answered a transfer call. Connecting you now.</Say>
+    <Dial>
+        <Conference endConferenceOnExit="true">{conference_name}</Conference>
+    </Dial>
+</Response>"""
+
+        # Prepare Twilio API call data
+        endpoint = f"{self.base_url}/Calls.json"
+        data = {
+            "To": destination,
+            "From": from_number,
+            "Timeout": timeout,
+            "Twiml": twiml,
+            "StatusCallback": status_callback_url,
+            "StatusCallbackEvent": [
+                "answered",
+                "no-answer",
+                "busy",
+                "failed",
+                "completed",
+            ],
+            "StatusCallbackMethod": "POST",
+        }
+
+        # Add any additional kwargs
+        data.update(kwargs)
+
+        try:
+            logger.debug(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.debug(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,
+                                "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,30 @@ class VobizProvider(TelephonyProvider):
 </Response>"""
 
         return Response(content=vobiz_xml_content, media_type="application/xml")
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        transfer_id: str,
+        conference_name: str,
+        timeout: int = 30,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Vobiz provider does not support call transfers.
+
+        Raises:
+            NotImplementedError: Vobiz call transfers are yet to be implemented
+        """
+        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,30 @@ class VonageProvider(TelephonyProvider):
         ]
 
         return Response(content=json.dumps(error_ncco), media_type="application/json")
+
+    # ======== CALL TRANSFER METHODS ========
+
+    async def transfer_call(
+        self,
+        destination: str,
+        transfer_id: str,
+        conference_name: str,
+        timeout: int = 30,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Vonage provider does not support call transfers.
+
+        Raises:
+            NotImplementedError: call transfers are yet to be implemented
+        """
+        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_event_protocol.py

@@ -0,0 +1,102 @@
+"""Redis communication protocol for call 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
+    transfer_id: str
+    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,
+            "end_call": self.end_call,
+            "reason": self.reason,
+        }
+        return result
+
+
+@dataclass
+class TransferContext:
+    """Transfer context data stored in Redis."""
+
+    transfer_id: str
+    call_sid: Optional[str]
+    target_number: str
+    tool_uuid: str
+    original_call_sid: str
+    conference_name: str
+    initiated_at: float
+
+    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(transfer_id: str) -> str:
+        """Channel for transfer events for a specific transfer."""
+        return f"transfer:events:{transfer_id}"
+
+    @staticmethod
+    def transfer_context_key(transfer_id: str) -> str:
+        """Redis key for transfer context storage."""
+        return f"transfer:context:{transfer_id}"

api/services/workflow/pipecat_engine.py

@@ -15,7 +15,6 @@ from pipecat.frames.frames import (
 from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.llm_service import FunctionCallParams
-from pipecat.transports.base_transport import BaseTransport
 from pipecat.utils.enums import EndTaskReason
 
 if TYPE_CHECKING:
@@ -61,7 +60,6 @@ class PipecatEngine:
         task: Optional[PipelineTask] = None,
         llm: Optional["LLMService"] = None,
         context: Optional[LLMContext] = None,
-        transport: Optional[BaseTransport] = None,
         workflow: WorkflowGraph,
         call_context_vars: dict,
         workflow_run_id: Optional[int] = None,
@@ -75,7 +73,6 @@ class PipecatEngine:
         self.task = task
         self.llm = llm
         self.context = context
-        self.transport = transport
         self.workflow = workflow
         self._call_context_vars = call_context_vars
         self._workflow_run_id = workflow_run_id
@@ -113,6 +110,9 @@ class PipecatEngine:
         self._embeddings_model: Optional[str] = embeddings_model
         self._embeddings_base_url: Optional[str] = embeddings_base_url
 
+        # Audio configuration (set via set_audio_config from _run_pipeline)
+        self._audio_config = None
+
     async def _get_organization_id(self) -> Optional[int]:
         """Get and cache the organization ID from workflow run."""
         if self._custom_tool_manager:
@@ -207,14 +207,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 +245,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 +278,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 +294,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 +305,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 +336,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 +537,9 @@ 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 +712,22 @@ class PipecatEngine:
                 f"Stasis connection set for immediate transfers: {connection.channel_id}"
             )
 
+    def set_audio_config(self, audio_config) -> None:
+        """Set the audio configuration for the pipeline."""
+        self._audio_config = audio_config
+
+    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,12 +6,20 @@ during workflow execution.
 
 from __future__ import annotations
 
+import asyncio
+import re
+import time
+import uuid
 from typing import TYPE_CHECKING, Any, Optional
 
 from loguru import logger
 
+from api.constants import APP_ROOT_DIR
 from api.db import db_client
 from api.enums import ToolCategory
+from api.services.telephony.call_transfer_manager import get_call_transfer_manager
+from api.services.telephony.factory import get_telephony_provider
+from api.services.telephony.transfer_event_protocol import TransferContext
 from api.services.workflow.disposition_mapper import (
     get_organization_id_from_workflow_run,
 )
@@ -20,8 +28,13 @@ from api.services.workflow.tools.custom_tool import (
     execute_http_tool,
     tool_to_function_schema,
 )
+from api.utils.hold_audio import load_hold_audio
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.frames.frames import FunctionCallResultProperties, TTSSpeakFrame
+from pipecat.frames.frames import (
+    FunctionCallResultProperties,
+    OutputAudioRawFrame,
+    TTSSpeakFrame,
+)
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.utils.enums import EndTaskReason
 
@@ -115,8 +128,15 @@ class CustomToolManager:
                 function_name = schema["function"]["name"]
 
                 # Create and register the handler
-                handler = self._create_handler(tool, function_name)
-                self._engine.llm.register_function(function_name, handler)
+                handler, disable_timeout, cancel_on_interruption = self._create_handler(
+                    tool, function_name
+                )
+                self._engine.llm.register_function(
+                    function_name,
+                    handler,
+                    cancel_on_interruption=cancel_on_interruption,
+                    disable_timeout=disable_timeout,
+                )
 
                 logger.debug(
                     f"Registered custom tool handler: {function_name} "
@@ -136,10 +156,21 @@ class CustomToolManager:
         Returns:
             Async handler function for the tool
         """
-        if tool.category == ToolCategory.END_CALL.value:
-            return self._create_end_call_handler(tool, function_name)
+        # Whether to disable function call timeout
+        disable_timeout = False
+        cancel_on_interruption = True
 
-        return self._create_http_tool_handler(tool, function_name)
+        if tool.category == ToolCategory.END_CALL.value:
+            cancel_on_interruption = False
+            handler = self._create_end_call_handler(tool, function_name)
+        elif tool.category == ToolCategory.TRANSFER_CALL.value:
+            disable_timeout = True
+            cancel_on_interruption = False
+            handler = self._create_transfer_call_handler(tool, function_name)
+        else:
+            handler = self._create_http_tool_handler(tool, function_name)
+
+        return handler, disable_timeout, cancel_on_interruption
 
     def _create_http_tool_handler(self, tool: Any, function_name: str):
         """Create a handler function for an HTTP API tool.
@@ -230,3 +261,321 @@ 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
+        """
+
+        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_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
+
+                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 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, 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 = {
+                        "status": "failed",
+                        "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
+                    )
+                    return
+
+                # Get original callSID from gathered_context
+                workflow_run = await db_client.get_workflow_run_by_id(
+                    self._engine._workflow_run_id
+                )
+                original_call_sid = workflow_run.gathered_context.get("call_id")
+
+                # Generate a unique transfer ID for tracking this transfer
+                transfer_id = str(uuid.uuid4())
+
+                # Compute conference name from original call SID
+                conference_name = f"transfer-{original_call_sid}"
+
+                # Mute the pipeline
+                self._engine.set_mute_pipeline(True)
+
+                # Initiate transfer via provider with inline TwiML
+                transfer_result = await provider.transfer_call(
+                    destination=destination,
+                    transfer_id=transfer_id,
+                    conference_name=conference_name,
+                    timeout=timeout_seconds,
+                )
+
+                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(),
+                )
+                await call_transfer_manager.store_transfer_context(transfer_context)
+
+                # Wait for status callback completion using Redis pub/sub
+                logger.info(
+                    f"Transfer call initiated for {destination} (transfer_id={transfer_id}), waiting for completion..."
+                )
+
+                # Start hold music during transfer waiting period
+                hold_music_stop_event = asyncio.Event()
+                hold_music_task = None
+
+                try:
+                    # Use audio config for sample rate (set during pipeline setup)
+                    sample_rate = (
+                        self._engine._audio_config.transport_out_sample_rate
+                        if self._engine._audio_config
+                        else 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 call_transfer_manager.wait_for_transfer_completion(
+                            transfer_id, timeout_seconds
+                        )
+                    )
+
+                except Exception as e:
+                    logger.error(f"Error during transfer wait: {e}")
+                    transfer_event = None
+
+                finally:
+                    # Single cleanup point: stop hold music, unmute pipeline, remove context
+                    logger.info(
+                        "Transfer wait ended, cleaning up hold music, pipeline state, and transfer context"
+                    )
+                    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:
+                    final_result = transfer_event.to_result_dict()
+                    await self._handle_transfer_result(
+                        final_result, function_call_params, properties
+                    )
+                else:
+                    logger.error(
+                        f"Transfer call timed out or failed after {timeout_seconds} seconds"
+                    )
+                    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"Transfer call tool '{function_name}' execution failed: {e}"
+                )
+                self._engine.set_mute_pipeline(False)
+
+                # 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", "")
+
+        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}"
+            )
+
+            # Inform LLM of success and end the call with Transfer call reason
+            response_properties = FunctionCallResultProperties(run_llm=False)
+            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")
+
+            await function_call_params.result_callback(
+                {
+                    "status": "transfer_failed",
+                    "reason": reason,
+                    "message": "Transfer failed",
+                }
+            )
+        else:
+            # Unknown action, treat as generic success
+            logger.warning(f"Unknown transfer action: {action}, treating as success")
+            await function_call_params.result_callback(result)
+
+    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:
+            # Path to hold music file based on sample rate
+            hold_music_file = (
+                APP_ROOT_DIR / "assets" / f"transfer_hold_ring_{sample_rate}.wav"
+            )
+            hold_audio_data = load_hold_audio(hold_music_file, sample_rate)
+            num_samples = len(hold_audio_data) // 2
+            duration = int(num_samples / sample_rate)
+
+            logger.info(f"Starting hold music loop with file: {hold_music_file}")
+
+            while not stop_event.is_set():
+                # Queue the hold audio frame
+                frame = OutputAudioRawFrame(
+                    audio=hold_audio_data,
+                    sample_rate=sample_rate,
+                    num_channels=1,
+                )
+                await self._engine.task.queue_frame(frame)
+
+                # Wait for the audio to play or until stopped
+                try:
+                    await asyncio.wait_for(stop_event.wait(), timeout=duration + 1.5)
+                    break  # Stop event was set
+                except asyncio.TimeoutError:
+                    pass  # Continue looping
+
+            logger.info("Hold music loop stopped")
+
+        except Exception as e:
+            logger.error(f"Error in hold music loop: {e}")

api/utils/hold_audio.py

@@ -0,0 +1,94 @@
+"""
+Hold audio utility for loading and caching hold music files.
+
+This module provides functionality to load hold music audio files at specific sample rates
+with caching to improve performance during multiple calls.
+"""
+
+from typing import Dict, Optional, Tuple
+
+import numpy as np
+from loguru import logger
+
+try:
+    import soundfile as sf
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use hold audio, you need to `pip install soundfile`.")
+    raise Exception(f"Missing module: {e}")
+
+
+# Global cache for loaded hold music data
+_hold_audio_cache: Dict[Tuple[str, int], np.ndarray] = {}
+
+
+def load_hold_audio(file_path: str, sample_rate: int) -> Optional[bytes]:
+    """Load hold music audio file at the specified sample rate with caching.
+
+    Args:
+        file_path: Path to the hold music audio file
+        sample_rate: Target sample rate (8000 or 16000 Hz supported)
+
+    Returns:
+        Audio data as bytes (PCM16) or None if loading failed
+    """
+    cache_key = (file_path, sample_rate)
+
+    # Check cache first
+    if cache_key in _hold_audio_cache:
+        logger.debug(f"Using cached hold audio for {file_path} at {sample_rate}Hz")
+        audio_data = _hold_audio_cache[cache_key]
+        return audio_data.tobytes()
+
+    try:
+        logger.info(f"Loading hold audio from {file_path} at {sample_rate}Hz")
+
+        # Load audio file
+        sound, file_sample_rate = sf.read(file_path, dtype="int16")
+        logger.info(
+            f"Audio file loaded - file sample_rate: {file_sample_rate}, target: {sample_rate}"
+        )
+
+        # Ensure mono audio (take first channel if stereo)
+        if len(sound.shape) > 1:
+            sound = sound[:, 0]
+
+        # Resample if needed
+        if file_sample_rate != sample_rate:
+            logger.warning(
+                f"Hold music file has sample rate {file_sample_rate}, expected {sample_rate}"
+            )
+            # For now, we'll use the audio as-is and let the transport handle resampling
+            # In a production system, you might want to use librosa or scipy for proper resampling
+
+        # Convert to int16 and cache
+        audio_data = sound.astype(np.int16)
+        _hold_audio_cache[cache_key] = audio_data
+
+        logger.info(
+            f"Hold audio loaded successfully: {len(audio_data)} samples at {sample_rate}Hz"
+        )
+        return audio_data.tobytes()
+
+    except Exception as e:
+        logger.error(f"Failed to load hold audio file {file_path}: {e}")
+        return None
+
+
+def clear_hold_audio_cache():
+    """Clear the hold audio cache to free memory."""
+    global _hold_audio_cache
+    _hold_audio_cache.clear()
+    logger.info("Hold audio cache cleared")
+
+
+def get_cache_info() -> Dict[str, int]:
+    """Get information about the current cache state.
+
+    Returns:
+        Dictionary with cache statistics
+    """
+    return {
+        "cached_files": len(_hold_audio_cache),
+        "total_cache_size": sum(len(data) for data in _hold_audio_cache.values()),
+    }