Extensions

{ .api }
from pipecat.extensions.ivr.ivr_navigator import IVRNavigator, IVRMenu, IVRStatus

class IVRNavigator(FrameProcessor):
    """Navigate IVR phone menus automatically.

    Automates navigation through phone tree menus using:
    - DTMF tone sequencing
    - Menu state machine
    - Verbal response handling
    - Timeout management

    Use Cases:
    - Automated phone tree navigation
    - IVR system testing
    - Call routing automation
    - Menu option selection

    Args:
        menu: IVR menu structure definition
        timeout_seconds: Timeout for menu responses
        retry_attempts: Number of retry attempts per menu

    Example:
        # Define menu structure
        menu = IVRMenu(
            id="main",
            prompt_keywords=["main menu", "options"],
            options=[
                IVROption(digit="1", next_menu_id="sales"),
                IVROption(digit="2", next_menu_id="support")
            ]
        )

        # Create navigator
        navigator = IVRNavigator(
            menu=menu,
            timeout_seconds=10
        )

        # Add to pipeline
        pipeline = Pipeline([
            transport.input(),
            stt,
            navigator,
            # ... rest of pipeline
        ])
    """

    def __init__(
        self,
        menu: IVRMenu,
        timeout_seconds: int = 10,
        retry_attempts: int = 3
    ):
        pass

    async def navigate(self, target_menu_id: str):
        """Navigate to a specific menu.

        Args:
            target_menu_id: ID of target menu
        """
        pass

    async def select_option(self, digit: str):
        """Select a menu option by digit.

        Args:
            digit: DTMF digit (0-9, *, #)
        """
        pass

    def get_current_status(self) -> IVRStatus:
        """Get current navigation status.

        Returns:
            Current IVR status
        """
        pass

{ .api }
from pipecat.extensions.ivr.ivr_navigator import IVRMenu, IVROption, IVRStatus

class IVRMenu:
    """IVR menu definition.

    Attributes:
        id (str): Unique menu identifier
        prompt_keywords (List[str]): Keywords to detect this menu
        options (List[IVROption]): Available menu options
        timeout_seconds (int): Menu timeout
        parent_menu_id (Optional[str]): Parent menu for back navigation
    """

    def __init__(
        self,
        id: str,
        prompt_keywords: List[str],
        options: List[IVROption],
        timeout_seconds: int = 10,
        parent_menu_id: Optional[str] = None
    ):
        pass


class IVROption:
    """IVR menu option.

    Attributes:
        digit (str): DTMF digit for this option (0-9, *, #)
        description (str): Option description
        next_menu_id (Optional[str]): Next menu to navigate to
        action (Optional[Callable]): Action to execute
    """

    def __init__(
        self,
        digit: str,
        description: str = "",
        next_menu_id: Optional[str] = None,
        action: Optional[Callable] = None
    ):
        pass


class IVRStatus(Enum):
    """IVR navigation status.

    Values:
        IDLE: Not navigating
        LISTENING: Listening for menu prompt
        NAVIGATING: Actively navigating
        COMPLETED: Navigation completed
        FAILED: Navigation failed
        TIMEOUT: Navigation timed out
    """
    IDLE = "idle"
    LISTENING = "listening"
    NAVIGATING = "navigating"
    COMPLETED = "completed"
    FAILED = "failed"
    TIMEOUT = "timeout"

{ .api }
from pipecat.extensions.voicemail.voicemail_detector import VoicemailDetector, VoicemailDetectionResult

class VoicemailDetector(FrameProcessor):
    """Detect voicemail and answering machines.

    Detects voicemail systems using:
    - Beep detection
    - Machine/human discrimination
    - Pattern recognition
    - Duration analysis

    Use Cases:
    - Skip voicemail in outbound calls
    - Leave automated messages
    - Call disposition handling
    - Call completion testing

    Args:
        beep_threshold: Threshold for beep detection
        min_silence_duration_ms: Minimum silence for voicemail
        max_greeting_duration_ms: Maximum greeting length
        on_voicemail_detected: Callback for voicemail detection

    Example:
        # Define callback
        async def handle_voicemail(result: VoicemailDetectionResult):
            if result.is_voicemail:
                print("Voicemail detected, leaving message")
                await task.queue_frame(
                    TTSSpeakFrame("Please call us back")
                )

        # Create detector
        detector = VoicemailDetector(
            beep_threshold=0.7,
            on_voicemail_detected=handle_voicemail
        )

        # Add to pipeline
        pipeline = Pipeline([
            transport.input(),
            stt,
            detector,
            # ... rest of pipeline
        ])
    """

    def __init__(
        self,
        beep_threshold: float = 0.7,
        min_silence_duration_ms: int = 1000,
        max_greeting_duration_ms: int = 30000,
        on_voicemail_detected: Optional[Callable] = None
    ):
        pass

    async def start_detection(self):
        """Start voicemail detection."""
        pass

    async def stop_detection(self):
        """Stop voicemail detection."""
        pass

    def is_detecting(self) -> bool:
        """Check if actively detecting.

        Returns:
            True if detection is active
        """
        pass

    def get_detection_result(self) -> Optional[VoicemailDetectionResult]:
        """Get current detection result.

        Returns:
            Detection result or None
        """
        pass

{ .api }
from pipecat.extensions.voicemail.voicemail_detector import VoicemailDetectionResult

class VoicemailDetectionResult:
    """Voicemail detection result.

    Attributes:
        is_voicemail (bool): True if voicemail detected
        confidence (float): Detection confidence (0.0-1.0)
        beep_detected (bool): True if beep was detected
        greeting_duration_ms (float): Greeting duration
        detection_time_ms (float): Time to detection
        method (str): Detection method used
    """

    def __init__(
        self,
        is_voicemail: bool,
        confidence: float,
        beep_detected: bool = False,
        greeting_duration_ms: float = 0.0,
        detection_time_ms: float = 0.0,
        method: str = "pattern"
    ):
        pass

{ .api }
from pipecat.extensions.ivr.ivr_navigator import IVRNavigator, IVRMenu, IVROption

# Define multi-level menu structure
main_menu = IVRMenu(
    id="main",
    prompt_keywords=["main menu", "options", "press"],
    options=[
        IVROption(digit="1", description="Sales", next_menu_id="sales"),
        IVROption(digit="2", description="Support", next_menu_id="support"),
        IVROption(digit="0", description="Operator", action=transfer_to_operator)
    ]
)

sales_menu = IVRMenu(
    id="sales",
    prompt_keywords=["sales", "department"],
    options=[
        IVROption(digit="1", description="New customer"),
        IVROption(digit="2", description="Existing customer"),
        IVROption(digit="9", description="Return", next_menu_id="main")
    ],
    parent_menu_id="main"
)

# Create navigator with menu tree
navigator = IVRNavigator(
    menu=main_menu,
    timeout_seconds=15,
    retry_attempts=2
)

# Navigate to specific option
await navigator.navigate("sales")
await navigator.select_option("1")  # Select new customer

# Check status
status = navigator.get_current_status()
if status == IVRStatus.COMPLETED:
    print("Navigation successful")

{ .api }
from pipecat.extensions.voicemail.voicemail_detector import VoicemailDetector

# Define voicemail callback
async def on_voicemail(result: VoicemailDetectionResult):
    """Handle voicemail detection."""
    print(f"Voicemail detected (confidence: {result.confidence})")

    if result.beep_detected:
        # Leave message after beep
        await task.queue_frames([
            TTSSpeakFrame(
                "Hello, this is an automated message. "
                "Please call us back at 555-1234. Thank you."
            ),
            EndFrame()  # End call after message
        ])
    else:
        # Hang up if no beep
        await task.queue_frame(EndFrame())

# Create detector
detector = VoicemailDetector(
    beep_threshold=0.75,
    min_silence_duration_ms=1500,
    max_greeting_duration_ms=25000,
    on_voicemail_detected=on_voicemail
)

# Add to outbound call pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    detector,
    user_agg,
    llm,
    tts,
    transport.output()
])

{ .api }
from pipecat.extensions.ivr.ivr_navigator import IVRNavigator
from pipecat.extensions.voicemail.voicemail_detector import VoicemailDetector

async def outbound_call_pipeline():
    """Complete outbound call handling pipeline."""

    # Configure voicemail detection
    async def handle_voicemail(result):
        if result.is_voicemail:
            print("Voicemail detected, leaving message")
            await task.queue_frame(
                TTSSpeakFrame("This is an important message...")
            )
            await task.cancel()  # End after message

    detector = VoicemailDetector(
        on_voicemail_detected=handle_voicemail
    )

    # Configure IVR navigation
    menu = IVRMenu(
        id="main",
        prompt_keywords=["main menu"],
        options=[
            IVROption(digit="2", description="Support")
        ]
    )

    navigator = IVRNavigator(menu=menu)

    # Build pipeline
    pipeline = Pipeline([
        transport.input(),
        stt,
        detector,      # Detect voicemail first
        navigator,     # Navigate IVR if human answers
        user_agg,
        llm,
        tts,
        transport.output()
    ])

    task = PipelineTask(pipeline)
    await task.run()

{ .api }
# Good: Define complete menu structure
menu = IVRMenu(
    id="main",
    prompt_keywords=["menu", "options", "press"],
    options=[
        IVROption("1", "Sales", next_menu_id="sales"),
        IVROption("2", "Support", next_menu_id="support")
    ],
    timeout_seconds=10
)

# Good: Handle navigation failures
@navigator.event_handler("on_navigation_failed")
async def handle_failure(reason: str):
    logger.error(f"IVR navigation failed: {reason}")
    # Fallback to operator
    await transfer_to_operator()

# Bad: No error handling
navigator = IVRNavigator(menu=menu)
# What happens if navigation fails?

{ .api }
# Good: Conservative thresholds
detector = VoicemailDetector(
    beep_threshold=0.75,          # High confidence
    min_silence_duration_ms=1500, # Allow for pauses
    max_greeting_duration_ms=30000 # Standard greeting length
)

# Good: Handle both outcomes
async def handle_detection(result):
    if result.is_voicemail:
        if result.confidence > 0.8:
            # High confidence - leave message
            await leave_message()
        else:
            # Low confidence - hang up
            await end_call()
    else:
        # Human answered - continue conversation
        pass

# Bad: Aggressive thresholds
detector = VoicemailDetector(
    beep_threshold=0.3,  # Too sensitive
    min_silence_duration_ms=500  # Too short
)
# Will false-positive on human pauses

{ .api }
# Good: Clean up resources
async def run_outbound_call():
    detector = VoicemailDetector()
    navigator = IVRNavigator(menu=menu)

    try:
        pipeline = Pipeline([
            transport.input(),
            detector,
            navigator,
            # ... rest
        ])

        task = PipelineTask(pipeline)
        await task.run()

    finally:
        # Clean up
        await detector.stop_detection()
        await navigator.cancel()