/azure-ai-voicelive-py

Azure AI Voice Live SDK - API Reference

Table of Contents

• connect() Function
• VoiceLiveConnection
• SessionResource
• ResponseResource
• InputAudioBufferResource
• OutputAudioBufferResource
• ConversationResource
• TranscriptionSessionResource
• WebsocketConnectionOptions
• Exceptions

connect() Function

Creates an async context manager for WebSocket connections.

from azure.ai.voicelive.aio import connect

async with connect(
    credential: Union[AzureKeyCredential, AsyncTokenCredential],
    endpoint: str,
    api_version: str = "2025-10-01",
    model: Optional[str] = None,
    query: Optional[Mapping[str, Any]] = None,
    headers: Optional[Mapping[str, Any]] = None,
    connection_options: Optional[WebsocketConnectionOptions] = None,
    credential_scopes: Optional[List[str]] = None,
    **kwargs
) -> VoiceLiveConnection:
    ...

Parameters

VoiceLiveConnection

Main connection class with resource accessors.

Properties

Methods

async def recv() -> ServerEvent:
    """Receive and parse the next typed server event."""

async def recv_bytes() -> bytes:
    """Receive raw bytes from the connection."""

async def send(event: Union[Mapping[str, Any], ClientEvent]) -> None:
    """Send an event to the server."""

async def close(*, code: int = 1000, reason: str = "") -> None:
    """Close the WebSocket connection."""

async def __aiter__() -> AsyncIterator[ServerEvent]:
    """Iterate over server events until connection closes."""

SessionResource

Manage session configuration.

Methods

async def update(
    *,
    session: Union[Mapping[str, Any], RequestSession],
    event_id: Optional[str] = None
) -> None:
    """Update session configuration."""

RequestSession Fields

ResponseResource

Manage model responses.

Methods

async def create(
    *,
    response: Optional[Union[ResponseCreateParams, Mapping[str, Any]]] = None,
    event_id: Optional[str] = None,
    additional_instructions: Optional[str] = None
) -> None:
    """Create a response (trigger inference)."""

async def cancel(
    *,
    response_id: Optional[str] = None,
    event_id: Optional[str] = None
) -> None:
    """Cancel an in-progress response."""

ResponseCreateParams Fields

InputAudioBufferResource

Manage audio input buffer.

Methods

async def append(
    *,
    audio: str,  # Base64-encoded audio
    event_id: Optional[str] = None
) -> None:
    """Append audio data to the input buffer."""

async def commit(
    *,
    event_id: Optional[str] = None
) -> None:
    """Commit the buffer as a user message."""

async def clear(
    *,
    event_id: Optional[str] = None
) -> None:
    """Clear the input buffer without committing."""

OutputAudioBufferResource

Manage audio output buffer.

Methods

async def clear(
    *,
    event_id: Optional[str] = None
) -> None:
    """Clear pending audio output (for interrupts)."""

ConversationResource

Manage conversation state.

Properties

ConversationItemResource

CRUD operations on conversation items.

Methods

async def create(
    *,
    item: Union[ConversationRequestItem, Mapping[str, Any]],
    previous_item_id: Optional[str] = None,
    event_id: Optional[str] = None
) -> None:
    """Create a new conversation item."""

async def delete(
    *,
    item_id: str,
    event_id: Optional[str] = None
) -> None:
    """Delete a conversation item."""

async def retrieve(
    *,
    item_id: str,
    event_id: Optional[str] = None
) -> None:
    """Retrieve item details (server responds with event)."""

async def truncate(
    *,
    item_id: str,
    audio_end_ms: int,
    content_index: int,
    event_id: Optional[str] = None
) -> None:
    """Truncate audio at specified time."""

ConversationRequestItem Types

# User/Assistant/System message
{
    "type": "message",
    "role": "user" | "assistant" | "system",
    "content": [
        {"type": "input_text", "text": "..."},
        {"type": "input_audio", "audio": "base64..."},
        {"type": "text", "text": "..."},
        {"type": "audio", "audio": "base64...", "transcript": "..."}
    ]
}

# Function call (from model)
{
    "type": "function_call",
    "call_id": "...",
    "name": "function_name",
    "arguments": "{...}"
}

# Function output (from client)
{
    "type": "function_call_output",
    "call_id": "...",
    "output": "{...}"
}

TranscriptionSessionResource

Configure input transcription.

Methods

async def update(
    *,
    session: Mapping[str, Any],
    event_id: Optional[str] = None
) -> None:
    """Update transcription session configuration."""

WebsocketConnectionOptions

Transport configuration for the WebSocket connection.

Exceptions

from azure.ai.voicelive.aio import ConnectionError, ConnectionClosed

class ConnectionError(AzureError):
    """Base WebSocket connection error."""

class ConnectionClosed(ConnectionError):
    """WebSocket connection was closed."""
    code: int      # Close code
    reason: str    # Close reason

Common Close Codes