Azure AI Voicelive Py
MicrosoftOfficialCreated by Microsoft
- Skill
- Tool integration
- Software development
- MIT
- Low risk
AutoGineer summary
Written by AutoGineer from the skill’s own files, not by the author
The azure-ai-voicelive-py skill provides a Python SDK to facilitate bidirectional WebSocket communication with Azure AI models. It enables developers to integrate voice capabilities into applications, supporting features such as voice assistants, chatbots, speech-to-speech translation, and voice-driven avatars. The SDK includes tools for managing audio buffers, configuring session parameters, handling function calls, and processing real-time events.
What it does
- Establishes bidirectional WebSocket connections for audio streaming.
- Supports Server VAD and manual turn-based conversation control.
- Enables function calling and integration with MCP tools.
- Provides mechanisms for real-time transcription and audio output.
- Offers configuration for various voice models and audio formats.
- Facilitates integration with voice-driven avatars.
When to use it
- When building Python applications that require real-time voice interaction.
- When implementing voice-enabled chatbots or virtual assistants.
- When developing systems requiring speech-to-speech translation.
- When integrating voice-driven avatars into software projects.
Advantages
- Native support for asynchronous Python patterns.
- Flexible authentication options including Entra ID integration.
- Comprehensive event handling for session and audio management.
- Granular control over audio input and output buffers.
Usage examples
- “How do I initialize a connection to the Azure AI Voice Live service?”
- The user receives a code snippet demonstrating the use of the connect() function within an async context manager, including the required endpoint and credential configuration.
- “Show me how to configure a voice assistant with a custom function tool.”
- The user receives an example of using the session.update() method to define a FunctionTool, including its name, description, and parameter schema.
Install
Download the skill as a ZIP:
https://autogineer.com/agent-skills/microsoft/azure-ai-voicelive-py/downloadUnzip it into the cross-client skills folder of your project:
mkdir -p .agents/skills && unzip azure-ai-voicelive-py.zip -d .agents/skills
Documentation by Microsoft
Azure AI Voice Live SDK
Build real-time voice AI applications with bidirectional WebSocket communication.
Installation
pip install azure-ai-voicelive aiohttp azure-identityEnvironment Variables
AZURE_COGNITIVE_SERVICES_ENDPOINT=https://<region>.api.cognitive.microsoft.com # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
AZURE_COGNITIVE_SERVICES_KEY=<api-key> # Only required for the legacy API-key auth path belowAuthentication & Lifecycle
🔑 Two rules apply to every code sample below:
Prefer
DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
Local dev:
DefaultAzureCredentialworks as-is.Production: set
AZURE_TOKEN_CREDENTIALS=prod(orAZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:
Sync:
with <Client>(...) as client:Async:
async with <Client>(...) as client:andasync with DefaultAzureCredential() as credential:(fromazure.identity.aio)Snippets may abbreviate this setup, but production code should always follow both rules.
import os
from azure.ai.voicelive.aio import connect
from azure.identity.aio import DefaultAzureCredential, ManagedIdentityCredential
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
async with DefaultAzureCredential(require_envvar=True) as credential:
async with connect(
endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
credential=credential,
model="gpt-4o-realtime-preview",
credential_scopes=["https://cognitiveservices.azure.com/.default"]
) as conn:
...Legacy: API Key (existing keyed deployments)
New code should use DefaultAzureCredential above. Use AzureKeyCredential only if you have an existing keyed deployment that hasn't been migrated to Entra ID yet — for example, regulated environments still completing their Entra rollout.
import os
from azure.core.credentials import AzureKeyCredential
from azure.ai.voicelive.aio import connect
async with connect(
endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
credential=AzureKeyCredential(os.environ["AZURE_COGNITIVE_SERVICES_KEY"]),
model="gpt-4o-realtime-preview",
) as conn:
...Quick Start
import asyncio
import os
from azure.ai.voicelive.aio import connect
from azure.identity.aio import DefaultAzureCredential
async def main():
async with connect(
endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
credential=DefaultAzureCredential(),
model="gpt-4o-realtime-preview",
credential_scopes=["https://cognitiveservices.azure.com/.default"]
) as conn:
# Update session with instructions
await conn.session.update(session={
"instructions": "You are a helpful assistant.",
"modalities": ["text", "audio"],
"voice": "alloy"
})
# Listen for events
async for event in conn:
print(f"Event: {event.type}")
if event.type == "response.audio_transcript.done":
print(f"Transcript: {event.transcript}")
elif event.type == "response.done":
break
asyncio.run(main())Core Architecture
Connection Resources
The VoiceLiveConnection exposes these resources:
| Resource | Purpose | Key Methods |
|---|---|---|
conn.session | Session configuration | update(session=...) |
conn.response | Model responses | create(), cancel() |
conn.input_audio_buffer | Audio input | append(), commit(), clear() |
conn.output_audio_buffer | Audio output | clear() |
conn.conversation | Conversation state | item.create(), item.delete(), item.truncate() |
conn.transcription_session | Transcription config | update(session=...) |
Session Configuration
from azure.ai.voicelive.models import RequestSession, FunctionTool
await conn.session.update(session=RequestSession(
instructions="You are a helpful voice assistant.",
modalities=["text", "audio"],
voice="alloy", # or "echo", "shimmer", "sage", etc.
input_audio_format="pcm16",
output_audio_format="pcm16",
turn_detection={
"type": "server_vad",
"threshold": 0.5,
"prefix_padding_ms": 300,
"silence_duration_ms": 500
},
tools=[
FunctionTool(
type="function",
name="get_weather",
description="Get current weather",
parameters={
"type": "object",
"properties": {
"location": {"type": "string"}
},
"required": ["location"]
}
)
]
))Audio Streaming
Send Audio (Base64 PCM16)
import base64
# Read audio chunk (16-bit PCM, 24kHz mono)
audio_chunk = await read_audio_from_microphone()
b64_audio = base64.b64encode(audio_chunk).decode()
await conn.input_audio_buffer.append(audio=b64_audio)Receive Audio
async for event in conn:
if event.type == "response.audio.delta":
audio_bytes = base64.b64decode(event.delta)
await play_audio(audio_bytes)
elif event.type == "response.audio.done":
print("Audio complete")Event Handling
async for event in conn:
match event.type:
# Session events
case "session.created":
print(f"Session: {event.session}")
case "session.updated":
print("Session updated")
# Audio input events
case "input_audio_buffer.speech_started":
print(f"Speech started at {event.audio_start_ms}ms")
case "input_audio_buffer.speech_stopped":
print(f"Speech stopped at {event.audio_end_ms}ms")
# Transcription events
case "conversation.item.input_audio_transcription.completed":
print(f"User said: {event.transcript}")
case "conversation.item.input_audio_transcription.delta":
print(f"Partial: {event.delta}")
# Response events
case "response.created":
print(f"Response started: {event.response.id}")
case "response.audio_transcript.delta":
print(event.delta, end="", flush=True)
case "response.audio.delta":
audio = base64.b64decode(event.delta)
case "response.done":
print(f"Response complete: {event.response.status}")
# Function calls
case "response.function_call_arguments.done":
result = handle_function(event.name, event.arguments)
await conn.conversation.item.create(item={
"type": "function_call_output",
"call_id": event.call_id,
"output": json.dumps(result)
})
await conn.response.create()
# Errors
case "error":
print(f"Error: {event.error.message}")Common Patterns
Manual Turn Mode (No VAD)
await conn.session.update(session={"turn_detection": None})
# Manually control turns
await conn.input_audio_buffer.append(audio=b64_audio)
await conn.input_audio_buffer.commit() # End of user turn
await conn.response.create() # Trigger responseInterrupt Handling
async for event in conn:
if event.type == "input_audio_buffer.speech_started":
# User interrupted - cancel current response
await conn.response.cancel()
await conn.output_audio_buffer.clear()Conversation History
# Add system message
await conn.conversation.item.create(item={
"type": "message",
"role": "system",
"content": [{"type": "input_text", "text": "Be concise."}]
})
# Add user message
await conn.conversation.item.create(item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "Hello!"}]
})
await conn.response.create()Voice Options
| Voice | Description |
|---|---|
alloy | Neutral, balanced |
echo | Warm, conversational |
shimmer | Clear, professional |
sage | Calm, authoritative |
coral | Friendly, upbeat |
ash | Deep, measured |
ballad | Expressive |
verse | Storytelling |
Azure voices: Use AzureStandardVoice, AzureCustomVoice, or AzurePersonalVoice models.
Audio Formats
| Format | Sample Rate | Use Case |
|---|---|---|
pcm16 | 24kHz | Default, high quality |
pcm16-8000hz | 8kHz | Telephony |
pcm16-16000hz | 16kHz | Voice assistants |
g711_ulaw | 8kHz | Telephony (US) |
g711_alaw | 8kHz | Telephony (EU) |
Turn Detection Options
# Server VAD (default)
{"type": "server_vad", "threshold": 0.5, "silence_duration_ms": 500}
# Azure Semantic VAD (smarter detection)
{"type": "azure_semantic_vad"}
{"type": "azure_semantic_vad_en"} # English optimized
{"type": "azure_semantic_vad_multilingual"}Error Handling
from azure.ai.voicelive.aio import ConnectionError, ConnectionClosed
try:
async with connect(...) as conn:
async for event in conn:
if event.type == "error":
print(f"API Error: {event.error.code} - {event.error.message}")
except ConnectionClosed as e:
print(f"Connection closed: {e.code} - {e.reason}")
except ConnectionError as e:
print(f"Connection error: {e}")Best Practices
This SDK is async-only; use the
.aionamespace throughout. Do not try to pair it with sync clients from other Azure SDKs in the same call path — keep the whole request path async.Always use context managers for clients and async credentials. Wrap every connection in
async with connect(...) as conn:. For asyncDefaultAzureCredentialfromazure.identity.aio, also useasync with credential:so tokens and transports are cleaned up.
References
Detailed API Reference: See references/api-reference.md
Complete Examples: See references/examples.md
All Models & Types: See references/models.md
Files
- ATTRIBUTION.mdAdded by AutoGineer974 B
- LICENSELicence file added by AutoGineer1.1 KiB
- SKILL.md11.3 KiB
- references/api-reference.md8.7 KiB
- references/examples.md16.9 KiB
- references/models.md18.0 KiB
references/examples.md
Azure AI Voice Live SDK - Examples
Table of Contents
Basic Voice Assistant
Function Calling
Manual Turn Control
Audio File Processing
Interrupt Handling
Multi-modal (Text + Audio)
Azure Voice Integration
Avatar Integration
Transcription Only
Basic Voice Assistant
Complete voice assistant with Server VAD.
import asyncio
import base64
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential
async def voice_assistant():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
# Configure session
await conn.session.update(session={
"instructions": "You are a helpful voice assistant. Be concise.",
"modalities": ["text", "audio"],
"voice": "alloy",
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 500
},
"input_audio_transcription": {
"model": "whisper-1"
}
})
# Start microphone input (pseudo-code)
mic_task = asyncio.create_task(stream_microphone(conn))
# Process events
async for event in conn:
match event.type:
case "session.created":
print("Session ready")
case "input_audio_buffer.speech_started":
print("🎤 Listening...")
case "conversation.item.input_audio_transcription.completed":
print(f"You: {event.transcript}")
case "response.audio.delta":
audio = base64.b64decode(event.delta)
await play_audio(audio)
case "response.audio_transcript.done":
print(f"Assistant: {event.transcript}")
case "error":
print(f"Error: {event.error.message}")
break
async def stream_microphone(conn):
"""Stream microphone audio to the connection."""
async for chunk in read_microphone(): # Your audio capture
b64 = base64.b64encode(chunk).decode()
await conn.input_audio_buffer.append(audio=b64)
asyncio.run(voice_assistant())Function Calling
Voice assistant with tool use.
import asyncio
import json
import base64
from azure.ai.voicelive.aio import connect
from azure.ai.voicelive.models import FunctionTool
from azure.core.credentials import AzureKeyCredential
# Define tools
TOOLS = [
FunctionTool(
type="function",
name="get_weather",
description="Get current weather for a location",
parameters={
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and state, e.g. 'San Francisco, CA'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "fahrenheit"
}
},
"required": ["location"]
}
),
FunctionTool(
type="function",
name="set_reminder",
description="Set a reminder for the user",
parameters={
"type": "object",
"properties": {
"message": {"type": "string"},
"time": {"type": "string", "description": "ISO 8601 datetime"}
},
"required": ["message", "time"]
}
)
]
def handle_function_call(name: str, arguments: str) -> dict:
"""Execute function and return result."""
args = json.loads(arguments)
if name == "get_weather":
# Mock weather API
return {
"location": args["location"],
"temperature": 72,
"unit": args.get("unit", "fahrenheit"),
"conditions": "sunny"
}
elif name == "set_reminder":
# Mock reminder service
return {"status": "success", "reminder_id": "123"}
else:
return {"error": f"Unknown function: {name}"}
async def function_calling_assistant():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
await conn.session.update(session={
"instructions": "You can check weather and set reminders.",
"modalities": ["text", "audio"],
"voice": "alloy",
"tools": TOOLS,
"tool_choice": "auto"
})
async for event in conn:
match event.type:
case "response.function_call_arguments.done":
# Execute the function
result = handle_function_call(event.name, event.arguments)
# Send result back
await conn.conversation.item.create(item={
"type": "function_call_output",
"call_id": event.call_id,
"output": json.dumps(result)
})
# Continue the conversation
await conn.response.create()
case "response.audio.delta":
audio = base64.b64decode(event.delta)
await play_audio(audio)
case "response.done":
if event.response.status == "completed":
print("Response complete")
asyncio.run(function_calling_assistant())Manual Turn Control
Push-to-talk style without VAD.
import asyncio
import base64
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential
async def push_to_talk():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
# Disable VAD for manual control
await conn.session.update(session={
"instructions": "You are a helpful assistant.",
"modalities": ["text", "audio"],
"voice": "alloy",
"turn_detection": None # Disable VAD
})
# Simulate push-to-talk
while True:
input("Press Enter to start recording...")
# Record audio (simulate with chunks)
chunks = await record_audio_until_release()
# Send all audio
for chunk in chunks:
b64 = base64.b64encode(chunk).decode()
await conn.input_audio_buffer.append(audio=b64)
# Commit and request response
await conn.input_audio_buffer.commit()
await conn.response.create()
# Wait for response
async for event in conn:
if event.type == "response.audio.delta":
audio = base64.b64decode(event.delta)
await play_audio(audio)
elif event.type == "response.done":
break
asyncio.run(push_to_talk())Audio File Processing
Process an audio file and get a response.
import asyncio
import base64
from pathlib import Path
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential
async def process_audio_file(audio_path: str):
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
await conn.session.update(session={
"instructions": "Respond to the audio message.",
"modalities": ["text", "audio"],
"voice": "alloy",
"turn_detection": None,
"input_audio_transcription": {"model": "whisper-1"}
})
# Read and send audio file
audio_data = Path(audio_path).read_bytes()
# Send in chunks (24kHz * 2 bytes * 0.1s = 4800 bytes)
chunk_size = 4800
for i in range(0, len(audio_data), chunk_size):
chunk = audio_data[i:i + chunk_size]
b64 = base64.b64encode(chunk).decode()
await conn.input_audio_buffer.append(audio=b64)
# Commit and request response
await conn.input_audio_buffer.commit()
await conn.response.create()
# Collect response
response_audio = bytearray()
response_text = ""
user_transcript = ""
async for event in conn:
match event.type:
case "conversation.item.input_audio_transcription.completed":
user_transcript = event.transcript
case "response.audio.delta":
response_audio.extend(base64.b64decode(event.delta))
case "response.audio_transcript.done":
response_text = event.transcript
case "response.done":
break
return {
"user_said": user_transcript,
"assistant_said": response_text,
"audio": bytes(response_audio)
}
result = asyncio.run(process_audio_file("input.pcm"))
print(f"User: {result['user_said']}")
print(f"Assistant: {result['assistant_said']}")
Path("output.pcm").write_bytes(result['audio'])Interrupt Handling
Handle user interruptions gracefully.
import asyncio
import base64
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential
async def interruptible_assistant():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
await conn.session.update(session={
"instructions": "You are a helpful assistant.",
"modalities": ["text", "audio"],
"voice": "alloy",
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 500
}
})
is_responding = False
async for event in conn:
match event.type:
case "response.created":
is_responding = True
case "response.done":
is_responding = False
case "input_audio_buffer.speech_started":
if is_responding:
# User interrupted - stop current response
print("🛑 Interrupt detected!")
await conn.response.cancel()
await conn.output_audio_buffer.clear()
case "response.audio.delta":
if is_responding:
audio = base64.b64decode(event.delta)
await play_audio(audio)
asyncio.run(interruptible_assistant())Multi-modal (Text + Audio)
Send text context, receive audio response.
import asyncio
import base64
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential
async def multimodal_assistant():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
await conn.session.update(session={
"instructions": "You are a helpful assistant.",
"modalities": ["text", "audio"],
"voice": "alloy",
"turn_detection": None
})
# Add context via text
await conn.conversation.item.create(item={
"type": "message",
"role": "system",
"content": [{"type": "input_text", "text": "The user's name is Alice."}]
})
# Add user message as text
await conn.conversation.item.create(item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "What's my name?"}]
})
# Request audio response
await conn.response.create()
async for event in conn:
if event.type == "response.audio.delta":
audio = base64.b64decode(event.delta)
await play_audio(audio)
elif event.type == "response.done":
break
asyncio.run(multimodal_assistant())Azure Voice Integration
Use Azure Text-to-Speech voices.
import asyncio
from azure.ai.voicelive.aio import connect
from azure.ai.voicelive.models import AzureStandardVoice, AzureCustomVoice
from azure.core.credentials import AzureKeyCredential
async def azure_voice_assistant():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
# Use Azure standard voice
await conn.session.update(session={
"instructions": "You are a helpful assistant.",
"modalities": ["text", "audio"],
"voice": AzureStandardVoice(
type="azure-standard",
name="en-US-JennyNeural"
)
})
# Or use custom voice
# await conn.session.update(session={
# "voice": AzureCustomVoice(
# type="azure-custom",
# endpoint_id="your-custom-voice-endpoint",
# name="YourCustomVoice"
# )
# })
async for event in conn:
# ... handle events
pass
asyncio.run(azure_voice_assistant())Avatar Integration
Connect to Azure Avatar for visual output.
import asyncio
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential
async def avatar_assistant():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
await conn.session.update(session={
"instructions": "You are a helpful assistant.",
"modalities": ["text", "audio", "avatar"],
"voice": "alloy",
"avatar": {
"type": "video-avatar",
"character": "lisa",
"output_protocol": "webrtc"
}
})
# Connect avatar
await conn.send({
"type": "session.avatar.connect"
})
async for event in conn:
match event.type:
case "session.avatar.connecting":
ice_servers = event.ice_servers
# Use ice_servers for WebRTC connection
print(f"Avatar connecting with {len(ice_servers)} ICE servers")
case "response.audio.delta":
# Audio is streamed via WebRTC, not this event
pass
asyncio.run(avatar_assistant())Transcription Only
Speech-to-text without AI response.
import asyncio
import base64
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential
async def transcription_only():
async with connect(
endpoint="https://eastus.api.cognitive.microsoft.com",
credential=AzureKeyCredential("YOUR_KEY"),
model="gpt-4o-realtime-preview"
) as conn:
await conn.session.update(session={
"modalities": ["text"], # No audio output
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 1000
},
"input_audio_transcription": {
"model": "whisper-1"
}
})
# Stream microphone
mic_task = asyncio.create_task(stream_microphone(conn))
transcripts = []
async for event in conn:
match event.type:
case "conversation.item.input_audio_transcription.delta":
print(event.delta, end="", flush=True)
case "conversation.item.input_audio_transcription.completed":
print() # Newline
transcripts.append(event.transcript)
return transcripts
asyncio.run(transcription_only())Requirements and compatibility
Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta
“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.
Needs: Needs an API key, Needs Python
Security
What our scanner noticed· Low risk
- Instructions that use the network
A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.
Attribution and licence
- Created by
- Microsoft
- Publisher
- Microsoft (Official)
- Source
- microsoft/skills · .github/plugins/azure-sdk-python/skills/azure-ai-voicelive-py · commit 0696600
- Licence
- MIT License (redistribution allowed)
- LICENSE (added by AutoGineer from LICENSE)
- ATTRIBUTION.md (written by AutoGineer)
- Synced
AutoGineer is not affiliated with or endorsed by Microsoft or Microsoft.
Download ZIP
The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download azure-ai-voicelive-py.zip