or run

tessl search

Version

Workspace: tessl
Visibility: Public
Created: 7 days ago
Last updated: 1 day ago
Describes: pkg:pypi/pipecat-ai@0.0.x

tessl/pypi-pipecat-ai

tessl install tessl/pypi-pipecat-ai@0.0.0

An open source framework for building real-time voice and multimodal conversational AI agents with support for speech-to-text, text-to-speech, LLMs, and multiple transport protocols

DTMF (Dual-Tone Multi-Frequency)

DTMF support enables telephone keypad integration for phone system interactions. Pipecat provides DTMF frame types, tone generation, and utilities for handling phone keypad input/output.

Overview

DTMF (Dual-Tone Multi-Frequency) is the signaling system used by telephone keypads. Each key press generates two simultaneous tones at specific frequencies, allowing automated phone systems to detect which key was pressed.

Pipecat provides complete DTMF support for:

Sending DTMF tones through transports
Receiving DTMF input from users
IVR (Interactive Voice Response) navigation
Phone menu interaction
Call control signaling

DTMF Types

KeypadEntry

{ .api }
from pipecat.audio.dtmf.types import KeypadEntry

class KeypadEntry(str, Enum):
    """DTMF keypad entries for phone system integration.

    Represents telephone keypad keys (0-9, *, #) for dual-tone
    multi-frequency signaling.

    Attributes:
        ONE: Number key 1
        TWO: Number key 2
        THREE: Number key 3
        FOUR: Number key 4
        FIVE: Number key 5
        SIX: Number key 6
        SEVEN: Number key 7
        EIGHT: Number key 8
        NINE: Number key 9
        ZERO: Number key 0
        POUND: Pound/hash key (#)
        STAR: Star/asterisk key (*)

    Example:
        from pipecat.audio.dtmf.types import KeypadEntry

        # Press '5' on keypad
        button = KeypadEntry.FIVE

        # Press '#' (pound)
        button = KeypadEntry.POUND

        # Press '*' (star)
        button = KeypadEntry.STAR
    """
    ONE = "1"
    TWO = "2"
    THREE = "3"
    FOUR = "4"
    FIVE = "5"
    SIX = "6"
    SEVEN = "7"
    EIGHT = "8"
    NINE = "9"
    ZERO = "0"
    POUND = "#"
    STAR = "*"

DTMF Utilities

load_dtmf_audio

{ .api }
from pipecat.audio.dtmf.utils import load_dtmf_audio
from pipecat.audio.dtmf.types import KeypadEntry

async def load_dtmf_audio(
    button: KeypadEntry,
    *,
    sample_rate: int = 8000
) -> bytes:
    """Load audio for DTMF tones associated with the given button.

    Loads pre-recorded DTMF audio files and resamples them to the
    requested sample rate. Audio is cached in-memory after first load
    for improved performance.

    Args:
        button: The keypad button for which to load DTMF audio
        sample_rate: The sample rate for the audio (default: 8000 Hz)

    Returns:
        The audio data for the DTMF tone as bytes (PCM format)

    Example:
        from pipecat.audio.dtmf.utils import load_dtmf_audio
        from pipecat.audio.dtmf.types import KeypadEntry

        # Load audio for '5' key at 8kHz
        audio = await load_dtmf_audio(KeypadEntry.FIVE)

        # Load audio for '#' key at 16kHz
        audio = await load_dtmf_audio(
            KeypadEntry.POUND,
            sample_rate=16000
        )
    """
    pass

DTMF Frames

OutputDTMFFrame

{ .api }
from pipecat.frames.frames import OutputDTMFFrame
from pipecat.audio.dtmf.types import KeypadEntry

class OutputDTMFFrame(DTMFFrame, DataFrame):
    """DTMF keypress output frame.

    Sends a DTMF tone through the transport. The tone will be queued
    and sent in order with other data frames.

    Attributes:
        button (KeypadEntry): The keypad button to press

    Example:
        # Press '5' on keypad
        frame = OutputDTMFFrame(button=KeypadEntry.FIVE)
        await task.queue_frame(frame)

        # Press '*' to access menu
        frame = OutputDTMFFrame(button=KeypadEntry.STAR)
        await task.queue_frame(frame)
    """
    pass

InputDTMFFrame

{ .api }
from pipecat.frames.frames import InputDTMFFrame

class InputDTMFFrame(DTMFFrame, SystemFrame):
    """DTMF keypress input frame.

    Received when user presses a key on their telephone keypad.
    SystemFrame for immediate processing.

    Example:
        # Handle DTMF input
        async def process_frame(self, frame, direction):
            if isinstance(frame, InputDTMFFrame):
                if frame.button == KeypadEntry.FIVE:
                    print("User pressed 5")
                elif frame.button == KeypadEntry.STAR:
                    print("User pressed *")

            await self.push_frame(frame, direction)
    """
    pass

OutputDTMFUrgentFrame

{ .api }
from pipecat.frames.frames import OutputDTMFUrgentFrame

class OutputDTMFUrgentFrame(DTMFFrame, SystemFrame):
    """DTMF keypress urgent output frame.

    Sends a DTMF tone immediately, bypassing the normal queue.
    SystemFrame for time-sensitive DTMF signaling.

    Example:
        # Send urgent interrupt code
        frame = OutputDTMFUrgentFrame(button=KeypadEntry.POUND)
        await task.queue_frame(frame)
    """
    pass

Usage Patterns

Sending DTMF Tones

{ .api }
from pipecat.frames.frames import OutputDTMFFrame
from pipecat.audio.dtmf.types import KeypadEntry
from pipecat.processors.frame_processor import FrameProcessor

class DTMFSender(FrameProcessor):
    """Send DTMF tones through transport."""

    async def send_dtmf_sequence(self, digits: str):
        """Send a sequence of DTMF digits.

        Args:
            digits: String of digits to send (e.g., "12345")
        """
        for digit in digits:
            # Map digit to KeypadEntry
            if digit.isdigit():
                button = KeypadEntry(digit)
            elif digit == "#":
                button = KeypadEntry.POUND
            elif digit == "*":
                button = KeypadEntry.STAR
            else:
                continue

            # Send DTMF frame
            frame = OutputDTMFFrame(button=button)
            await self.push_frame(frame)

            # Add small delay between tones
            await asyncio.sleep(0.1)

# Usage
sender = DTMFSender()
await sender.send_dtmf_sequence("12345")  # Send "12345"
await sender.send_dtmf_sequence("*123#")  # Send "*123#"

Receiving DTMF Input

{ .api }
from pipecat.frames.frames import InputDTMFFrame
from pipecat.audio.dtmf.types import KeypadEntry
from pipecat.processors.frame_processor import FrameProcessor

class DTMFReceiver(FrameProcessor):
    """Receive and process DTMF input."""

    def __init__(self):
        super().__init__()
        self._dtmf_buffer = ""

    async def process_frame(self, frame, direction):
        if isinstance(frame, InputDTMFFrame):
            # Add to buffer
            self._dtmf_buffer += frame.button.value
            print(f"Received DTMF: {frame.button.value}")

            # Check for complete sequence
            if frame.button == KeypadEntry.POUND:
                # End of input (# pressed)
                print(f"Complete input: {self._dtmf_buffer}")
                await self._process_dtmf_input(self._dtmf_buffer)
                self._dtmf_buffer = ""

        await self.push_frame(frame, direction)

    async def _process_dtmf_input(self, input: str):
        """Process complete DTMF input."""
        if input == "1":
            print("Option 1 selected")
        elif input == "2":
            print("Option 2 selected")
        # ... handle other options

IVR Menu Navigation

{ .api }
from pipecat.frames.frames import OutputDTMFFrame, InputDTMFFrame
from pipecat.audio.dtmf.types import KeypadEntry
from pipecat.processors.frame_processor import FrameProcessor

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

    def __init__(self, menu_sequence: str):
        super().__init__()
        self._sequence = menu_sequence
        self._current_step = 0
        self._awaiting_menu = True

    async def process_frame(self, frame, direction):
        # Wait for audio indicating menu is ready
        if isinstance(frame, TTSAudioRawFrame) and self._awaiting_menu:
            # Menu audio received, send next digit
            if self._current_step < len(self._sequence):
                digit = self._sequence[self._current_step]
                button = self._map_digit_to_button(digit)

                # Send DTMF tone
                await self.push_frame(
                    OutputDTMFFrame(button=button),
                    direction
                )

                self._current_step += 1
                self._awaiting_menu = False

        await self.push_frame(frame, direction)

    def _map_digit_to_button(self, digit: str) -> KeypadEntry:
        """Map string digit to KeypadEntry."""
        if digit.isdigit():
            return KeypadEntry(digit)
        elif digit == "#":
            return KeypadEntry.POUND
        elif digit == "*":
            return KeypadEntry.STAR
        raise ValueError(f"Invalid digit: {digit}")

# Usage: Navigate to option 2, then option 3, then confirm
navigator = IVRNavigator(menu_sequence="23#")

DTMF Aggregator

{ .api }
from pipecat.processors.aggregators.dtmf import DTMFAggregator

class DTMFAggregator(FrameProcessor):
    """Aggregate DTMF inputs into complete entries.

    Collects DTMF keypress frames and aggregates them into
    complete phone numbers or menu selections.

    Args:
        terminator: Keypad entry that ends input (default: POUND)
        timeout: Seconds to wait before auto-submitting (default: 5.0)
        max_length: Maximum number of digits (default: 20)

    Example:
        # Aggregate DTMF with # as terminator
        aggregator = DTMFAggregator(
            terminator=KeypadEntry.POUND,
            timeout=5.0
        )

        # Pipeline processes individual keypresses
        # Aggregator emits complete input when # pressed
    """

    def __init__(
        self,
        terminator: KeypadEntry = KeypadEntry.POUND,
        timeout: float = 5.0,
        max_length: int = 20
    ):
        super().__init__()
        self._terminator = terminator
        self._timeout = timeout
        self._max_length = max_length
        self._buffer = ""
        self._last_input_time = 0

    async def process_frame(self, frame, direction):
        if isinstance(frame, InputDTMFFrame):
            self._last_input_time = time.time()

            if frame.button == self._terminator:
                # Submit complete input
                await self._emit_complete_input()
                self._buffer = ""
            else:
                # Add to buffer
                self._buffer += frame.button.value

                # Check max length
                if len(self._buffer) >= self._max_length:
                    await self._emit_complete_input()
                    self._buffer = ""

        await self.push_frame(frame, direction)

    async def _emit_complete_input(self):
        """Emit aggregated DTMF input."""
        if self._buffer:
            # Emit as transcription or custom frame
            frame = TranscriptionFrame(
                text=self._buffer,
                user_id="dtmf",
                timestamp=datetime.now().isoformat()
            )
            await self.push_frame(frame)

Combining Audio and DTMF

{ .api }
from pipecat.frames.frames import OutputDTMFFrame, TTSSpeakFrame
from pipecat.audio.dtmf.types import KeypadEntry

async def send_menu_selection(task, option: int):
    """Send menu selection with confirmation.

    Args:
        task: Pipeline task
        option: Menu option number (1-9)
    """
    # Speak selection
    await task.queue_frame(
        TTSSpeakFrame(text=f"Selecting option {option}")
    )

    # Send DTMF tone
    button = KeypadEntry(str(option))
    await task.queue_frame(OutputDTMFFrame(button=button))

    # Confirm with pound
    await task.queue_frame(
        OutputDTMFFrame(button=KeypadEntry.POUND)
    )

DTMF with Delays

{ .api }
import asyncio
from pipecat.frames.frames import OutputDTMFFrame
from pipecat.audio.dtmf.types import KeypadEntry

async def send_dtmf_with_delays(task, digits: str, delay_ms: int = 100):
    """Send DTMF digits with delays between them.

    Args:
        task: Pipeline task
        digits: String of digits to send
        delay_ms: Delay between digits in milliseconds
    """
    for digit in digits:
        # Map digit
        if digit.isdigit():
            button = KeypadEntry(digit)
        elif digit == "#":
            button = KeypadEntry.POUND
        elif digit == "*":
            button = KeypadEntry.STAR
        else:
            continue

        # Send DTMF
        await task.queue_frame(OutputDTMFFrame(button=button))

        # Wait between tones
        await asyncio.sleep(delay_ms / 1000.0)

# Example: Send phone number with delays
await send_dtmf_with_delays(task, "5551234567#", delay_ms=150)

Best Practices

Use Appropriate Delays

{ .api }
# Good: Add delays between DTMF tones
await task.queue_frame(OutputDTMFFrame(button=KeypadEntry.ONE))
await asyncio.sleep(0.1)  # 100ms delay
await task.queue_frame(OutputDTMFFrame(button=KeypadEntry.TWO))

# Bad: Sending too quickly may not be recognized
await task.queue_frame(OutputDTMFFrame(button=KeypadEntry.ONE))
await task.queue_frame(OutputDTMFFrame(button=KeypadEntry.TWO))

Handle Input Timeouts

{ .api }
class DTMFInputHandler(FrameProcessor):
    """Handle DTMF input with timeout."""

    def __init__(self, timeout: float = 5.0):
        super().__init__()
        self._timeout = timeout
        self._buffer = ""
        self._last_input = 0

    async def process_frame(self, frame, direction):
        if isinstance(frame, InputDTMFFrame):
            current_time = time.time()

            # Check timeout
            if self._last_input and (current_time - self._last_input > self._timeout):
                # Timeout - clear buffer
                self._buffer = ""

            self._buffer += frame.button.value
            self._last_input = current_time

        await self.push_frame(frame, direction)

Validate DTMF Input

{ .api }
def validate_dtmf_input(input: str) -> bool:
    """Validate DTMF input is valid.

    Args:
        input: DTMF input string

    Returns:
        True if valid, False otherwise
    """
    valid_chars = set("0123456789*#")
    return all(c in valid_chars for c in input)

# Usage
if validate_dtmf_input(dtmf_buffer):
    await process_input(dtmf_buffer)
else:
    print("Invalid DTMF input")

Use Urgent Frames for Critical Signaling

{ .api }
# Use urgent frame for emergency interrupt
async def emergency_interrupt(task):
    """Send emergency interrupt sequence."""
    # Send ** as urgent interrupt
    await task.queue_frame(
        OutputDTMFUrgentFrame(button=KeypadEntry.STAR)
    )
    await task.queue_frame(
        OutputDTMFUrgentFrame(button=KeypadEntry.STAR)
    )

Complete Example

{ .api }
from pipecat.frames.frames import (
    OutputDTMFFrame,
    InputDTMFFrame,
    TTSSpeakFrame
)
from pipecat.audio.dtmf.types import KeypadEntry
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.pipeline.pipeline import Pipeline
import asyncio

class DTMFPhoneSystem(FrameProcessor):
    """Complete DTMF phone system integration."""

    def __init__(self):
        super().__init__()
        self._input_buffer = ""
        self._state = "idle"

    async def process_frame(self, frame, direction):
        # Handle incoming DTMF
        if isinstance(frame, InputDTMFFrame):
            await self._handle_dtmf_input(frame.button)

        await self.push_frame(frame, direction)

    async def _handle_dtmf_input(self, button: KeypadEntry):
        """Process DTMF input from user."""
        if button == KeypadEntry.POUND:
            # End of input
            if self._input_buffer == "1":
                await self._route_to_sales()
            elif self._input_buffer == "2":
                await self._route_to_support()
            elif self._input_buffer == "0":
                await self._route_to_operator()
            self._input_buffer = ""
        else:
            self._input_buffer += button.value

    async def _route_to_sales(self):
        """Route call to sales."""
        await self.push_frame(
            TTSSpeakFrame(text="Connecting you to sales")
        )
        # Send transfer DTMF sequence
        await self._send_transfer_sequence("1001")

    async def _send_transfer_sequence(self, extension: str):
        """Send DTMF transfer sequence."""
        # Star-star-extension-pound for transfer
        await self.push_frame(
            OutputDTMFFrame(button=KeypadEntry.STAR)
        )
        await asyncio.sleep(0.1)
        await self.push_frame(
            OutputDTMFFrame(button=KeypadEntry.STAR)
        )

        for digit in extension:
            await asyncio.sleep(0.1)
            button = KeypadEntry(digit)
            await self.push_frame(OutputDTMFFrame(button=button))

        await asyncio.sleep(0.1)
        await self.push_frame(
            OutputDTMFFrame(button=KeypadEntry.POUND)
        )

# Pipeline setup
phone_system = DTMFPhoneSystem()
pipeline = Pipeline([
    transport.input(),
    phone_system,
    transport.output()
])

Version

tessl/pypi-pipecat-ai

dtmf.mddocs/audio/

DTMF (Dual-Tone Multi-Frequency)

Overview

DTMF Types

KeypadEntry

DTMF Utilities

load_dtmf_audio

DTMF Frames

OutputDTMFFrame

InputDTMFFrame

OutputDTMFUrgentFrame

Usage Patterns

Sending DTMF Tones

Receiving DTMF Input

IVR Menu Navigation

DTMF Aggregator

Combining Audio and DTMF

DTMF with Delays

Best Practices

Use Appropriate Delays

Handle Input Timeouts

Validate DTMF Input

Use Urgent Frames for Critical Signaling

Complete Example

See Also

Version

tessl/pypi-pipecat-ai

dtmf.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/audio/

DTMF (Dual-Tone Multi-Frequency)

Overview

DTMF Types

KeypadEntry

DTMF Utilities

load_dtmf_audio

DTMF Frames

OutputDTMFFrame

InputDTMFFrame

OutputDTMFUrgentFrame

Usage Patterns

Sending DTMF Tones

Receiving DTMF Input

IVR Menu Navigation

DTMF Aggregator

Combining Audio and DTMF

DTMF with Delays

Best Practices

Use Appropriate Delays

Handle Input Timeouts

Validate DTMF Input

Use Urgent Frames for Critical Signaling

Complete Example

See Also

dtmf.mddocs/audio/