Call Commands | AgentDuet SDK

Call commands control telephony on a live Call: answer, connect a human party, whisper, barge, spy, close, and disconnect. Each command method returns a CommandResult (truthy on success). Treat a falsy result as an operational failure; connection loss and programmer errors still raise exceptions.

result = await call.answer()
if not result:
    logger.error("%s (%s)", result.error_message, result.error_code)
    return

Full method list: API Reference - Call. This page is about when to use each command, with complete programs you can run.

Two inbound shapes

A. Agent is the voice (answer)

Use when your bot speaks for the subscriber (IVR, receptionist, booking agent).

if not await call.answer():
    return
async for chunk in call.caller.audio_stream():
    await call.send_audio(chunk)  # or model PCM

Later escalate with connect() if a human should join, then whisper() a brief and close() so humans continue.

B. Ambient agent (connect first)

Use when two humans talk directly and your code listens or coaches. On inbound, call bare connect() without answer():

result = await call.connect(ring_time_seconds=30)
if not result:
    if result.error_code == "CALL_UNANSWERED":
        await call.disconnect()
    return
await call.spy()      # listen only
# ... later ...
await call.whisper()  # heard only by subscriber
await call.send_audio(coaching_pcm)
await call.close()    # humans stay connected

Tutorial: answer + respond (echo)

Agent is the voice. Answer inbound, echo caller audio until hangup.

import asyncio
import logging
import os
from agentduet import (
    CallAudioConfig,
    CallClosedError,
    IncomingCallNotification,
    SessionManager,
    SessionManagerConfig,
    new_session_id,
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def main() -> None:
    config = SessionManagerConfig.create(
        api_key=os.getenv("AGENTDUET_API_KEY"),
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
        call_audio=CallAudioConfig(sample_rate=16000),
    )
    async with SessionManager(config) as sm:
        @sm.on_incoming_call
        async def on_call(noti: IncomingCallNotification):
            session = await sm.open_session(new_session_id(), noti.subscriber)
            call = await session.process_call(noti)
            result = await call.answer()
            if not result:
                logger.error(
                    "answer failed: %s (%s)",
                    result.error_message,
                    result.error_code,
                )
                return
            try:
                async for chunk in call.caller.audio_stream():
                    await call.send_audio(chunk)
            except CallClosedError:
                logger.info("Call %s closed", call.id)
        await sm.run_forever()
if __name__ == "__main__":
    asyncio.run(main())

Tutorial: ambient connect → spy → whisper

Humans talk directly. Your agent connects ambiently, listens in spy mode, then whispers a short tone burst to the subscriber only, then leaves with close().

import asyncio
import logging
import os
import struct
from agentduet import (
    Call,
    CallAudioConfig,
    CallClosedError,
    IncomingCallNotification,
    SessionManager,
    SessionManagerConfig,
    new_session_id,
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
SAMPLE_RATE = 16000
def tone_pcm(
    duration_s: float = 0.4,
    freq_hz: float = 880.0,
    sample_rate: int = SAMPLE_RATE,
    amplitude: int = 4000,
) -> bytes:
    """Generate a short mono 16-bit LE sine tone (stand-in for coaching TTS)."""
    import math
    n = int(duration_s * sample_rate)
    samples = [
        int(amplitude * math.sin(2 * math.pi * freq_hz * i / sample_rate))
        for i in range(n)
    ]
    return struct.pack(f"<{n}h", *samples)
async def ambient_coach(call: Call) -> None:
    result = await call.connect(ring_time_seconds=30)
    if not result:
        logger.error(
            "connect failed: %s (%s)",
            result.error_message,
            result.error_code,
        )
        if result.error_code == "CALL_UNANSWERED":
            await call.disconnect()
        return
    if not await call.spy():
        logger.error("spy() failed for %s", call.id)
        return
    # Listen briefly on the caller track (customer on inbound).
    heard = 0
    try:
        async for chunk in call.caller.audio_stream():
            heard += len(chunk)
            if heard >= SAMPLE_RATE * 2:  # ~1s of 16-bit mono ≈ 2 * rate bytes
                break
    except CallClosedError:
        return
    # Private tip to subscriber (staff), then leave the media path.
    if await call.whisper():
        tip = tone_pcm()
        chunk_bytes = int(SAMPLE_RATE * 0.04) * 2  # 40 ms
        for i in range(0, len(tip), chunk_bytes):
            await call.send_audio(tip[i : i + chunk_bytes])
    await call.close()  # humans stay connected
async def main() -> None:
    config = SessionManagerConfig.create(
        api_key=os.getenv("AGENTDUET_API_KEY"),
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
        call_audio=CallAudioConfig(sample_rate=SAMPLE_RATE),
    )
    async with SessionManager(config) as sm:
        @sm.on_incoming_call
        async def on_call(noti: IncomingCallNotification):
            session = await sm.open_session(new_session_id(), noti.subscriber)
            call = await session.process_call(noti)
            await ambient_coach(call)
        await sm.run_forever()
if __name__ == "__main__":
    asyncio.run(main())

To record both parties, also consume call.callee.audio_stream() in a parallel task - audio is always isolated per party.

Ambient modes (after `connect()`)

Command	Who hears the agent
`spy()`	Nobody (monitor only)
`whisper()`	Subscriber only
`barge()`	Both parties

Who is the subscriber? On inbound: your line (callee). On outbound: your line (caller). See Participant Model.

Outbound

call = await session.make_call(Address.telco("+15551234567"))
if not await call.dial(ring_time_seconds=30):
    # CALL_UNANSWERED / TIMEOUT - retry, dial another number, or end the call
    return
async for chunk in call.callee.audio_stream():
    ...

Audio commands

Command	Use
`send_audio(pcm)`	Queue outbound PCM
`clear_send_audio_buffer()`	Drop queued audio on barge-in (public API - prefer this)
`get_send_audio_buffer_size()`	Inspect buffer fill

Do not call private helpers such as call._interrupt(). Use clear_send_audio_buffer().

Next Step

Sessions and Handoff

Serialize NEW calls to JSON and hand off media processing across distributed worker processes or microservices.