---

name: voice-live description: Implement real-time voice agent capabilities using Azure Voice Live API. Use when building conversational AI voice agents, speech-to-speech applications, real-time audio processing, voice-enabled bots, or when users mention Voice Live, real-time voice, speech synthesis with avatars, or WebSocket-based voice interactions. Covers WebSocket setup, session configuration, audio streaming, noise suppression, echo cancellation, turn detection, function calling (VoiceRAG), avatar integration, and custom voice support.

Voice Live API Skill

Build real-time voice agents using Azure Voice Live API with WebSocket connections, Python backend, and advanced conversational features.

Overview

The Voice Live API enables low-latency, high-quality speech-to-speech interactions for voice agents. It integrates speech recognition, generative AI, and text-to-speech into a unified WebSocket interface.

Key Capabilities

• Broad locale coverage: 140+ locales for STT, 600+ voices across 150+ locales for TTS
• Customizable input/output: Phrase lists, custom speech models, custom voices
• Flexible AI models: GPT-5, GPT-4.1, GPT-4o, Phi, and more
• Advanced conversational features:
  • Noise suppression (azure_deep_noise_suppression)
  • Echo cancellation (server_echo_cancellation)
  • Robust interruption detection
  • Advanced end-of-turn detection
• Avatar integration: Standard or customizable avatars synchronized with audio
• Function calling: VoiceRAG pattern for external actions and grounded responses

Quick Start

Prerequisites

1. Azure subscription
2. Microsoft Foundry resource or Azure Speech Services resource
3. Python 3.9+ with websockets and azure-identity packages

Installation

Copypip install websockets azure-identity pyaudio

Basic Connection

Copyimport asyncio
import websockets
import json
import base64
from azure.identity import DefaultAzureCredential

# Configuration
RESOURCE_NAME = "<your-ai-foundry-resource-name>"
MODEL = "gpt-4o"  # or gpt-realtime, gpt-5, phi4-mini, etc.
API_VERSION = "2025-10-01"

async def connect_voice_live():
    # Get auth token
    credential = DefaultAzureCredential()
    token = credential.get_token("https://ai.azure.com/.default").token
    
    # Build WebSocket URL
    ws_url = f"wss://{RESOURCE_NAME}.services.ai.azure.com/voice-live/realtime?api-version={API_VERSION}&model={MODEL}"
    
    headers = {"Authorization": f"Bearer {token}"}
    
    async with websockets.connect(ws_url, additional_headers=headers) as ws:
        # Wait for session.created
        response = await ws.recv()
        session = json.loads(response)
        print(f"Session created: {session['session']['id']}")
        
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "You are a helpful AI assistant.",
                "voice": {
                    "name": "en-US-Ava:DragonHDLatestNeural",
                    "type": "azure-standard"
                },
                "turn_detection": {
                    "type": "azure_semantic_vad",
                    "threshold": 0.3,
                    "prefix_padding_ms": 200,
                    "silence_duration_ms": 200,
                    "end_of_utterance_detection": {
                        "model": "semantic_detection_v1",
                        "threshold_level": "default",
                        "timeout_ms": 1000
                    }
                },
                "input_audio_noise_reduction": {"type": "azure_deep_noise_suppression"},
                "input_audio_echo_cancellation": {"type": "server_echo_cancellation"}
            }
        }))
        
        # Handle messages
        async for message in ws:
            event = json.loads(message)
            await handle_event(event)

async def handle_event(event):
    event_type = event.get("type")
    
    if event_type == "session.updated":
        print("Session configured successfully")
    elif event_type == "response.audio.delta":
        # Decode and play audio
        audio_data = base64.b64decode(event["delta"])
        # Play audio_data through speakers
    elif event_type == "response.text.delta":
        print(event["delta"], end="", flush=True)
    elif event_type == "response.done":
        print("\n[Response complete]")
    elif event_type == "error":
        print(f"Error: {event['error']['message']}")

asyncio.run(connect_voice_live())

Session Configuration

Voice Options

Copy# Azure Standard Voice
"voice": {
    "name": "en-US-AvaNeural",
    "type": "azure-standard"
}

# Azure HD Voice (with expressiveness)
"voice": {
    "name": "en-US-Ava:DragonHDLatestNeural",
    "type": "azure-standard",
    "temperature": 0.8,
    "rate": "1.0"
}

# Azure Custom Voice
"voice": {
    "name": "my-custom-voice",
    "type": "azure-custom",
    "endpoint_id": "12345678-1234-1234-1234-123456789012"
}

# OpenAI Voice (only with gpt-realtime models)
"voice": {
    "type": "openai",
    "name": "alloy"  # alloy, ash, ballad, coral, echo, sage, shimmer, verse
}

Turn Detection Options

Copy# Azure Semantic VAD (recommended)
"turn_detection": {
    "type": "azure_semantic_vad",
    "threshold": 0.3,
    "prefix_padding_ms": 200,
    "silence_duration_ms": 500,
    "remove_filler_words": True,
    "end_of_utterance_detection": {
        "model": "semantic_detection_v1",
        "threshold_level": "default",
        "timeout_ms": 1000
    }
}

# Multilingual Semantic VAD
"turn_detection": {
    "type": "azure_semantic_vad_multilingual",
    "languages": ["en", "es", "fr", "de", "ja", "zh"]
}

# Server VAD (basic)
"turn_detection": {
    "type": "server_vad",
    "threshold": 0.5,
    "prefix_padding_ms": 300,
    "silence_duration_ms": 500
}

Audio Configuration

Copy"input_audio_format": "pcm16",  # pcm16, g711_ulaw, g711_alaw
"output_audio_format": "pcm16",
"input_audio_sampling_rate": 24000,  # 16000 or 24000
"input_audio_noise_reduction": {"type": "azure_deep_noise_suppression"},
"input_audio_echo_cancellation": {"type": "server_echo_cancellation"}

Sending Audio

Copyimport pyaudio

async def stream_audio(ws):
    p = pyaudio.PyAudio()
    stream = p.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        input=True,
        frames_per_buffer=4800  # 200ms chunks
    )
    
    try:
        while True:
            audio_data = stream.read(4800)
            audio_b64 = base64.b64encode(audio_data).decode()
            
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": audio_b64
            }))
            await asyncio.sleep(0.2)
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()

Function Calling (VoiceRAG Pattern)

Enable external actions and grounded responses:

Copy# Configure tools in session
"tools": [
    {
        "type": "function",
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    },
    {
        "type": "function",
        "name": "search_knowledge_base",
        "description": "Search internal knowledge base",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"}
            },
            "required": ["query"]
        }
    }
]

Handle function calls:

Copyasync def handle_event(event, ws):
    if event["type"] == "response.function_call_arguments.done":
        call_id = event["call_id"]
        name = event["name"] if "name" in event else None
        arguments = json.loads(event["arguments"])
        
        # Execute the function
        result = await execute_function(name, arguments)
        
        # Send function result back
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": json.dumps(result)
            }
        }))
        
        # Trigger response generation
        await ws.send(json.dumps({"type": "response.create"}))

async def execute_function(name, args):
    if name == "get_weather":
        # Call weather API
        return {"temperature": 22, "condition": "sunny"}
    elif name == "search_knowledge_base":
        # Search your RAG system
        return {"results": ["Document 1", "Document 2"]}

Avatar Integration

Copy"modalities": ["text", "audio", "avatar"],
"avatar": {
    "character": "lisa",  # or custom avatar ID
    "style": "casual-sitting",
    "customized": False,
    "video": {
        "resolution": {"width": 1920, "height": 1080},
        "bitrate": 2000000,
        "codec": "h264"
    }
}

After session.updated, establish WebRTC connection:

Copy# Send client SDP
await ws.send(json.dumps({
    "type": "session.avatar.connect",
    "client_sdp": "<your-client-sdp>"
}))

# Receive server SDP in session.avatar.connecting event

Custom Speech Input

Phrase List (Just-in-time customization)

Copy"input_audio_transcription": {
    "model": "azure-speech",
    "phrase_list": ["Neo QLED TV", "TUF Gaming", "AutoQuote Explorer"]
}

Custom Speech Model

Copy"input_audio_transcription": {
    "model": "azure-speech",
    "language": "en",
    "custom_speech": {
        "zh-CN": "847cb03d-7f22-4b11-444-e1be1d77bf17"
    }
}

Supported Models

WebSocket Event Reference

See references/events.md for complete event documentation.

Key Client Events

• session.update - Configure session
• input_audio_buffer.append - Send audio
• input_audio_buffer.commit - Commit audio buffer
• conversation.item.create - Add items (messages, function outputs)
• response.create - Request response
• response.cancel - Cancel response

Key Server Events

• session.created - Session established
• session.updated - Configuration confirmed
• response.audio.delta - Streaming audio
• response.text.delta - Streaming text
• response.function_call_arguments.done - Function call ready
• input_audio_buffer.speech_started - Speech detected
• input_audio_buffer.speech_stopped - Speech ended

Best Practices

1. Audio quality: Use 24kHz sample rate for best quality
2. Noise suppression: Always enable for better accuracy
3. Echo cancellation: Enable when playing audio through speakers
4. Turn detection: Use azure_semantic_vad for natural conversations
5. Buffer size: Send audio in 200ms chunks for low latency
6. Error handling: Monitor error events and implement reconnection
7. Function calling: Keep functions focused and well-documented

Complete Example

See scripts/voice_live_client.py for a full-featured implementation.