Audio conversion utilities in Python

An audio utility library

To build a robust Voice Agent app, you need utility functions that convert audio between formats. This lesson collects the essential Python tools for real-time audio processing.

PCM16 to base64 and back

These are the two functions you’ll use most: encoding captured audio in base64 to send it, and decoding received base64 audio to play it.

import base64
import numpy as np

def audio_to_base64(audio_data: np.ndarray) -> str:
    """Converts a float32 numpy array in [-1, 1] to base64 PCM16."""
    audio_int16 = (audio_data * 32767).astype(np.int16)
    return base64.b64encode(audio_int16.tobytes()).decode("utf-8")

def base64_to_audio(base64_audio: str) -> np.ndarray:
    """Converts a base64 PCM16 string to a float32 numpy array."""
    audio_bytes = base64.b64decode(base64_audio)
    audio_int16 = np.frombuffer(audio_bytes, dtype=np.int16)
    return audio_int16.astype(np.float32) / 32768.0

Using them in the Voice Agent flow

import pyaudio
import json

# Mic capture → send
def capture_and_send(ws, stream, chunk_size=4096):
    raw_data = stream.read(chunk_size, exception_on_overflow=False)
    audio_array = np.frombuffer(raw_data, dtype=np.int16)
    audio_float = audio_array.astype(np.float32) / 32768.0
    b64 = audio_to_base64(audio_float)

    ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": b64
    }))

# Receive → play
def play_received(player_stream, base64_audio):
    audio_float = base64_to_audio(base64_audio)
    audio_int16 = (audio_float * 32767).astype(np.int16)
    player_stream.write(audio_int16.tobytes())

Resampling

If your microphone captures at a different sample rate than the API expects, you need to resample:

from scipy.signal import resample

def resample_audio(
    audio: np.ndarray,
    original_rate: int,
    target_rate: int
) -> np.ndarray:
    """Resamples audio from one sample rate to another."""
    if original_rate == target_rate:
        return audio
    num_samples = int(len(audio) * target_rate / original_rate)
    return resample(audio, num_samples)

# Example: mic at 44100 Hz → API at 24000 Hz
audio_44100 = capture_microphone()
audio_24000 = resample_audio(audio_44100, 44100, 24000)
b64 = audio_to_base64(audio_24000)

Converting between G.711 and PCM

For phone integrations, you’ll need to convert between G.711 and linear PCM. Watch your Python version: audioop was removed from the standard library in 3.13. On a recent Python, install the audioop-lts package, which provides the same interface:

import audioop

def pcm_to_mulaw(pcm_data: bytes) -> bytes:
    """Converts PCM16 to G.711 mu-law."""
    return audioop.lin2ulaw(pcm_data, 2)

def mulaw_to_pcm(mulaw_data: bytes) -> bytes:
    """Converts G.711 mu-law to PCM16."""
    return audioop.ulaw2lin(mulaw_data, 2)

def pcm_to_alaw(pcm_data: bytes) -> bytes:
    """Converts PCM16 to G.711 A-law."""
    return audioop.lin2alaw(pcm_data, 2)

def alaw_to_pcm(alaw_data: bytes) -> bytes:
    """Converts G.711 A-law to PCM16."""
    return audioop.alaw2lin(alaw_data, 2)

A complete phone relay

Here’s an example of a relay between a G.711 SIP stream and the Voice Agent API:

async def relay_sip_to_voice_agent(sip_stream, ws):
    """Relays audio from a SIP call to the Voice Agent API."""
    async for mulaw_chunk in sip_stream:
        # Convert mu-law → PCM → base64
        pcm_data = mulaw_to_pcm(mulaw_chunk)
        b64 = base64.b64encode(pcm_data).decode("utf-8")

        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": b64
        }))

async def relay_voice_agent_to_sip(ws, sip_stream):
    """Relays the API's audio response to the SIP stream."""
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            pcm_data = base64.b64decode(event["delta"])
            mulaw_data = pcm_to_mulaw(pcm_data)
            await sip_stream.send(mulaw_data)

Client-side silence detection

If you handle push-to-talk yourself, or want local silence detection before sending, this function measures the signal’s energy:

def is_silence(
    audio: np.ndarray,
    threshold: float = 0.01
) -> bool:
    """Detects whether an audio chunk is silence."""
    rms = np.sqrt(np.mean(audio ** 2))
    return rms < threshold

# Usage
audio_chunk = capture_microphone()
if not is_silence(audio_chunk, threshold=0.02):
    b64 = audio_to_base64(audio_chunk)
    ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": b64
    }))

Saving and loading WAV files

For debugging and testing, you can save and load WAV files:

import wave

def save_wav(
    filename: str,
    audio: np.ndarray,
    sample_rate: int = 24000
):
    """Saves a numpy array as a WAV file."""
    audio_int16 = (audio * 32767).astype(np.int16)
    with wave.open(filename, "wb") as wf:
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(sample_rate)
        wf.writeframes(audio_int16.tobytes())

def load_wav(filename: str) -> tuple[np.ndarray, int]:
    """Loads a WAV file as a float32 numpy array."""
    with wave.open(filename, "rb") as wf:
        sample_rate = wf.getframerate()
        frames = wf.readframes(wf.getnframes())
        audio_int16 = np.frombuffer(frames, dtype=np.int16)
        return audio_int16.astype(np.float32) / 32768.0, sample_rate

Key takeaways

• audio_to_base64 and base64_to_audio are the two essential functions for the Voice Agent flow
• Resampling is needed when your microphone’s sample rate doesn’t match the API’s
• The audioop module handles G.711 mu-law and A-law conversions; it was removed from the standard library in Python 3.13, so install audioop-lts from PyPI on recent versions
• Client-side silence detection saves bandwidth by sending only useful audio
• WAV utilities make debugging easier by letting you save and replay audio streams

Check your understanding

WebSocket, VAD, audio: the voice agent from end to end.

1. How do you set up a Voice Agent session?

Answer: With a real-time WebSocket connection, authenticated with ephemeral tokens (never the main key on the client), then configured through session.update and confirmed by session.created and session.updated.

2. What does VAD handle?

Answer: Automatic speech detection: it works out when the user starts and stops speaking, so the agent knows when to listen and when to answer, with no button.

3. How does audio flow?

Answer: Input goes out through input_audio_buffer.append (with buffer commit and clear), and the response arrives as audio events to play, in PCM or G.711 depending on the channel.

4. What do tools and MCP make possible in a voice conversation?

Answer: Function calling and built-in tools (web, X, collections) during the spoken exchange: the agent looks things up and takes action while it talks, MCP events included.

5. And for telephony?

Answer: SIP and LiveKit with G.711: the voice agent plugs into the phone network, and the Python audio conversion utilities bridge the formats.

A secure session, VAD, well-managed buffers, tools in voice: the complete phone assistant is within reach, and this course has given you every piece.