Concepts

Audio Streaming

View as Markdown

Call media is raw PCM: mono, 16-bit little-endian. There is no container or codec layer in the SDK. Configure rate and buffers with CallAudioConfig, and match the sample rate to the attached speech model.

Configuration

Configure PCM with CallAudioConfig on SessionManagerConfig.create(...).

FieldDefaultNotes
sample_rate16000 when call_audio= omittedOne of 8000, 16000, 24000. Required when you construct CallAudioConfig. Match your model (Gemini Live / Grok / OpenAI Realtime examples use 24000).
buffer_size1 MBOutgoing ring buffer in bytes; power of two.
inbound_queue_maxsize1000Max buffered inbound chunks per track. When full, oldest chunk is dropped. 0 = unbounded (only if you always consume in real time).
from agentduet import CallAudioConfig, SessionManagerConfig
import os
config = SessionManagerConfig.create(
    api_key=os.getenv("AGENTDUET_API_KEY"),
    connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
    call_audio=CallAudioConfig(sample_rate=24000, buffer_size=1024 * 1024),
)

Per-party receive

async for chunk in call.caller.audio_stream():
    ...  # only the caller
async for chunk in call.callee.audio_stream():
    ...  # only the callee (outbound dialed party, or staff after connect)

The agent is the sender (send_audio), not a receivable track.

Send, backpressure, interrupt

try:
    await call.send_audio(pcm_bytes)
except BufferFullError:
    # Throttle or drop - outbound ring buffer is full
    pass
await call.clear_send_audio_buffer()  # barge-in: drop queued outbound audio
size = await call.get_send_audio_buffer_size()

send_audio does not write straight to the wire. Bytes land in an internal ring buffer; the SDK drains it as the server pulls. That is why barge-in needs an explicit clear: otherwise the caller still hears the old sentence for a beat.

After hangup, send_audio / clear_send_audio_buffer raise CallClosedError. Treat that as the normal stop signal.

Concurrent bridge (complete pattern)

import asyncio
from agentduet import Call, CallClosedError, BufferFullError
async def bridge(call: Call, model) -> None:
    """model must expose send_pcm(chunk) and an async iterator of (pcm, interrupted)."""
    async def to_model():
        async for chunk in call.caller.audio_stream():
            await model.send_pcm(chunk)
    async def from_model():
        try:
            async for pcm, interrupted in model.recv_pcm():
                if interrupted:
                    await call.clear_send_audio_buffer()
                    continue
                try:
                    await call.send_audio(pcm)
                except BufferFullError:
                    pass
        except CallClosedError:
            pass
    await asyncio.gather(to_model(), from_model())

Wire hangup to cancel model work:

@call.on_hangup
def on_hangup(evt):
    asyncio.create_task(model.close())

MIME / rate tips for models

Model familyTypical AgentDuet sample_rateNotes
Gemini Live24000audio/pcm;rate=24000 blobs
OpenAI Realtime24000pcm16 in/out
Grok Voice24000match session audio.format.rate
Qwen Omni24000 on AgentDuetresample to 16 kHz before send to Qwen

Next Step

Event Handling

Handle asynchronous call lifecycle callbacks, hangup notifications, competing-consumer concurrency, and error recovery.