agentduet 1.0.0b10

Build AI agents for phone and WhatsApp calls with real-time audio and messaging

AgentDuet Python SDK

Build AI agents that meet your customers where they already are: on the phone or on WhatsApp. Answer an incoming call, stream the caller's audio into your AI model and the model's voice back, and handle interruptions the instant they happen. Follow up by text in the same conversation, so one agent can speak and chat.

Features

• Meet customers where they are. Phone and WhatsApp, voice and text, in one SDK. A single Session can carry a call and messages with the same customer, so your agent speaks and chats in one conversation.
• Real-time audio built for AI. Bidirectional low-latency PCM streaming with pull-based flow control, plus instant buffer clearing so your agent stops talking the moment the caller cuts in.
• Bring any AI model. Audio is plain PCM in and out, so any real-time model plugs in: Gemini Live, OpenAI Realtime, Amazon Nova Sonic, and more.
• Inbound and outbound. Answer incoming calls, place outbound calls, and send and receive WhatsApp messages.
• Assist live calls. Your agent can join a call between two people: listen in, speak to one side or both, or sit between the callers and translate live.
• Pick your auth. API key, or mTLS client certificates.
• Stays connected. Heartbeat, connection monitoring, and automatic reconnect with exponential backoff.
• Scale across nodes. Serialize a call to JSON and rebuild it on another server to move media processing wherever you want.
• Route what you want. Push runtime rules to control which calls and messages reach your agent.

Installation

pip install --pre agentduet

The package is currently published as 1.0.0bN pre-releases, so --pre is required until 1.0.0 final ships. The package is imported as agentduet. Requires Python 3.12+.

To get a voice agent talking to an AI model in one command, add the adapters extra:

pip install --pre "agentduet[adapters]"

That brings in agentduet-adapters (Apache-2.0) with every model adapter. For a production install, name the one provider you call instead: pip install --pre "agentduet-adapters[gemini]". See VoiceAgent.

Quick Start

Get your API key and connector UUID at agentduet.com and set them as AGENTDUET_API_KEY and AGENTDUET_CONNECTOR_UUID in your environment.

Incoming calls and messages are delivered at the connector level; the connector is your agent's connection point to AgentDuet, the thing your credentials identify. You register @sm.on_incoming_call and @sm.on_incoming_message on the SessionManager. Each notification carries addressing only: a subscriber (the principal the event belongs to) and the external participant. To act on one, you open a short-lived Session for that subscriber with sm.open_session(session_id, subscriber):

• For a call, session.process_call(noti) attaches the session to the call and returns a ready-to-use Call, media access already set up.
• For a message, session.send_message(...) sends your reply; the server infers the recipient from the payload.

The session_id is any unique string you choose; reuse the same id to continue a conversation, or use a new one to start fresh. Pick the flow you want below.

Answer a call

Answer each incoming call and echo the caller's audio back.

import asyncio
import logging
import os

from agentduet import (
    CallAudioConfig,
    CallClosedError,
    IncomingCallNotification,
    SessionManager,
    SessionManagerConfig,
    new_session_id,
)

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)


async def main():
    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:
        logger.info("Connected. Waiting for calls...")

        @sm.on_incoming_call
        async def on_call(noti: IncomingCallNotification):
            logger.info("Incoming call %s from %s", noti.call_id, noti.participant)

            session = await sm.open_session(new_session_id(), noti.subscriber)
            call = await session.process_call(noti)

            @call.on_hangup
            def on_hangup(evt):
                logger.info("Call %s hung up", call.id)

            if not await call.answer():
                logger.error("Answer failed for call %s", call.id)
                return

            try:
                async for chunk in call.caller.audio_stream():
                    await call.send_audio(chunk)  # echo back
            except CallClosedError:
                pass  # caller hung up mid-send; normal end of call

        await sm.run_forever()


if __name__ == "__main__":
    asyncio.run(main())

Reply to a WhatsApp message

Register @sm.on_incoming_message. Each IncomingMessage carries the subscriber (your business identity), the participant (the customer to reply to), and the raw webhook payload. To reply, open a session for your subscriber and call send_message().

from agentduet import IncomingMessage, SendWAMessage, new_session_id


@sm.on_incoming_message
async def on_message(msg: IncomingMessage):
    logger.info("Message from %s: %s", msg.participant, msg.payload)

    session = await sm.open_session(new_session_id(), msg.subscriber)
    result = await session.send_message(
        SendWAMessage(
            api_version="v23.0",
            data={
                "messaging_product": "whatsapp",
                "type": "text",
                "to": msg.participant.value,
                "text": {"body": "Thanks, we got your message!"},
            },
        )
    )
    if not result.success:
        logger.error("Send failed: %s (%s)", result.error_code, result.error_content)

msg.payload is the raw WhatsApp webhook payload, so its shape depends on the message type (text, button, image, and so on); inspect it to decide how to reply.

Place a call

Open a session for the calling subscriber, create a call to the destination Address, then dial().

from agentduet import Address, CallClosedError, new_session_id


async def place_call(sm):
    session = await sm.open_session(new_session_id(), "your-subscriber-id")
    call = await session.make_call(Address.telco("[phone]"))

    if await call.dial(ring_time_seconds=30):
        logger.info("Outbound call %s answered", call.id)
        try:
            async for chunk in call.callee.audio_stream():  # the dialed party's audio
                await call.send_audio(chunk)
        except CallClosedError:
            pass  # callee hung up mid-send; normal end of call
    else:
        logger.error("Dial was not answered")

Authentication

Pick one of two modes when you build the config. Both go through SessionManagerConfig.create().

API key (with your connector UUID):

config = SessionManagerConfig.create(
    api_key=os.getenv("AGENTDUET_API_KEY"),
    connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
)

mTLS (client certificate and key):

config = SessionManagerConfig.create(
    cert_path="/etc/certs/client.pem",
    key_path="/etc/certs/client.key",
)

Audio Configuration

Audio settings are optional and live in a CallAudioConfig, passed as call_audio= to SessionManagerConfig.create(). Omit call_audio= entirely for the defaults below. Note that when you do construct a CallAudioConfig, sample_rate is a required argument; the other fields have defaults.

Field	Default	Notes
sample_rate	16000 (when call_audio= is omitted)	Hz. One of 8000, 16000, or 24000 (the SampleRate type, exported). Required when constructing CallAudioConfig yourself. Match it to your AI model (the integration examples use 24000 for Gemini; VoiceAgent.from_env() also defaults to 24000).
buffer_size	1 MB	Outgoing ring-buffer size in bytes; must be a power of two.
inbound_queue_maxsize	1000	Max buffered inbound audio chunks per track. When full, the oldest chunk is dropped so a slow consumer can't exhaust memory. 0 means unbounded; opt out only if you always consume a party's audio_stream() in real time.

Audio is always isolated: each call has two receivable tracks, read per party via call.caller.audio_stream() and call.callee.audio_stream(). The agent is the sender (call.send_audio()), not a receivable track.

from agentduet import CallAudioConfig

config = SessionManagerConfig.create(
    api_key=os.getenv("AGENTDUET_API_KEY"),
    connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
    call_audio=CallAudioConfig(sample_rate=24000),
)

Architecture

The SDK has three moving parts and manages all of them for you. You drive them through three objects: a SessionManager, a Session, and a Call.

1. Session manager connection. One persistent control connection to the server. It authenticates, keeps a heartbeat, reconnects automatically on drops, and delivers incoming call and message notifications. You hold a single SessionManager for the life of your process.
2. Session. An ephemeral, per-subscriber handle you open with sm.open_session(session_id, subscriber). It carries your calls (process_call / make_call, returning a Call) and outbound messages (send_message) for that subscriber.
3. Voice connection. The per-call media connection behind a Call, carrying low-latency PCM both ways. The SDK opens it lazily the first time a call needs audio (on answer(), dial(), connect(), send_audio(), and so on) and closes it with the call; you never manage it directly, you just drive the Call.

Participant Model

Every Call has two parties and one agent.

• The two parties are the caller (who placed the call) and the callee (who was called). One of them is always the subscriber, the principal the call runs on behalf of (your connector's line/number); the other is the external participant.
• The agent is your code. It listens to each party's audio (call.caller.audio_stream() / call.callee.audio_stream()) and speaks into the call (call.send_audio()), but it is not one of the two parties.

caller and callee describe membership, not liveness: who the call is between, not who has picked up. A party can be defined on the call but not yet connected (e.g. the callee before your agent bridges them in). Read caller / callee / subscriber for identity; watch call state and events for liveness. For inbound calls the SDK seeds these for you: caller = the external participant, callee = your subscriber.

Where does the agent sit? (the one rule)

Whether a scenario is one call or several comes down to a single question: is the agent in the audio path?

• Ambient agent → one Call. The two parties talk directly; the agent listens and, when it chooses, speaks to one side or both. Scope who hears the agent with:
  • spy(): hear the call, speak to no one (monitor only)
  • whisper(): speak only to the subscriber (the callee on an inbound call, the caller on an outbound one)
  • barge(): speak to everyone
• In-path agent → one Call per party. The parties do not hear each other directly; the agent sits between them and relays/transforms audio (e.g. a live translator). Each party is its own Call with the agent on both, and the agent moves audio across them.

Worked examples

Scenario	Calls	Shape
Voice assistant: the agent answers on the subscriber's behalf	1	caller = external party, callee = subscriber (no human on the subscriber side; the agent is the callee's voice)
Call monitor: the agent listens to a live human↔human call and warns/assists	1	caller = external party, callee = subscriber (a real human); agent ambient via spy() → whisper() / barge()
Live translator: the agent relays between two humans	2	one Call per human; agent in-path on both

Multi-party (attended transfer, conferencing, N participants) is composed the same way: several Calls coordinated by your agent. A native conference Room is planned as an additive object alongside Call (it will not change the two-party Call). Blind transfer to the subscriber is available today via connect().

Per-party audio

Each party's audio arrives on its own isolated stream: call.caller.audio_stream() carries only the caller, call.callee.audio_stream() carries only the callee. The agent can therefore attribute speech to a specific party (for example, transcribe or screen one side only) without any mixing or track bookkeeping.

Trigger Conditions

You do not need this to get started. By default, the server delivers inbound calls and inbound messages to your connector, so the Quick Start works as-is.

Configure trigger conditions when you want to change what the server routes to you, for example to receive only missed calls or to stop receiving inbound calls. Build the conditions with TriggerConditionsBuilder (its build() returns the TriggerConditions object that setup_trigger_conditions takes) and send them once after connecting.

from agentduet import InboundCallMode, TriggerConditionsBuilder

trigger_config = (
    TriggerConditionsBuilder()
    .inbound_call(InboundCallMode.ALL)   # deliver all incoming calls
    .inbound_message(True)               # deliver incoming WhatsApp messages
    .build()
)

await sm.setup_trigger_conditions(trigger_config)

InboundCallMode controls which inbound calls reach you:

• InboundCallMode.ALL routes every incoming call.
• InboundCallMode.MISSED_ONLY routes only calls no other destination answered.
• InboundCallMode.NO stops inbound call delivery.

Three things to keep in mind:

• Trigger conditions route notifications; they never gate what the SDK can do. Calls you place yourself with session.make_call() and dial() work regardless of this configuration. The builder's outbound_call / outbound_message toggles belong to an upcoming feature (notifying your agent when a subscriber places a call or sends a message, so it can step in before the call or message reaches its destination) and have no effect yet.
• The builder starts permissive, and the config you send is absolute. Defaults are inbound calls ALL and inbound messages on (outbound toggles off). Whatever you build fully replaces the server-side configuration when you send it, so set every flow you want; a flow you leave at its default (or don't touch) still takes that default's value, not your previous configuration's.
• Send it once. The server applies your configuration to all routing from then on, including after an automatic reconnect. You do not need to re-send it.

Common Call Flows

These are the patterns you build from the Call commands, given a call you obtained from session.process_call(noti) (inbound) or session.make_call(dest) plus dial() (outbound). Command methods return a CommandResult that is truthy on success, so check the return value instead of catching exceptions for operational failures.

Answer and respond

The core AI agent loop: answer the call, read the caller's audio, send your model's audio back.

async def handle(call: Call):
    if not await call.answer():
        logger.error("Failed to answer")
        return
    try:
        async for chunk in call.caller.audio_stream():
            response = await ai_model.generate(chunk)
            await call.send_audio(response)
    except CallClosedError:
        pass  # caller hung up while the model was responding
    await call.close()

Connect a third party, then whisper

Bring another person onto the call as a 3-way conference, then speak privately to one side. After connect(), switch the agent's audio between the three ambient modes from the Participant Model: whisper() (subscriber only), barge() (everyone), spy() (listen only).

async def assistant_flow(call: Call):
    if not await call.answer():
        return
    if not await call.connect(ring_time_seconds=30):
        return
    await call.whisper()
    await call.send_audio(private_guidance_pcm)
    await call.close()  # agent leaves; caller and callee stay connected

Hand the call to another node

Forward the call elsewhere for media processing. The receiving node reconstructs it and owns the audio from then on. to_json() is only valid before the media connection opens (in CallState.NEW).

# On the first node, before answering:
await message_queue.publish(call.to_json())  # your transport

# On the other node:
call = Call.from_json(received_message)
if not await call.answer():
    logger.error("Failed to answer on secondary node")

Event Handling

A call hands you two kinds of output, each consumed its own way: discrete events and continuous audio.

Decorators for discrete events

Register a handler for something that happens once, like the call ending or a server-side error.

from agentduet import CallEvent


async def handle(call: Call):
    await call.answer()

    @call.on_hangup
    def on_hangup(evt):
        print("Call ended")

    # Server-side call errors
    @call.on_call_event(CallEvent.ERROR)
    def on_error(evt):
        print(f"Call error: {evt['error_code']} {evt.get('error_message')}")

Every call-event handler takes exactly one argument, the event payload. on_hangup is shorthand for on_call_event(CallEvent.HANGUP); its payload is always None (the event carries no data). A CallEvent.ERROR handler receives a dict with error_code and an optional error_message. Sync handlers are offloaded to a worker thread, so blocking work in one will not stall audio delivery.

Async iterators for continuous streams

Audio is a stream, so you consume it per party with async for. The loop ends when the call's audio stops.

try:
    async for audio_chunk in call.caller.audio_stream():
        processed = await process_audio(audio_chunk)
        await call.send_audio(processed)
except CallClosedError:
    pass  # the call ended while a send was in flight

Once the call terminates, send_audio() (and clear_send_audio_buffer()) raise CallClosedError. A hangup can land while your producer is mid-response, so catch it as the normal stop signal, as above.

Audio Buffering and Interruption

When you call send_audio(), the data does not go straight to the network. It lands in an internal buffer, and the SDK sends it to the server only as the server asks for more. This pull-based flow control is what keeps playback smooth: the server always has a steady supply of audio, and you never flood the connection.

The payoff shows up when your agent gets interrupted. Real-time models like Gemini Live emit an interruption signal the moment the caller starts talking over the agent. At that point you have a buffer full of audio the agent was about to say, and you want it gone:

# Your model signaled the caller interrupted
await call.clear_send_audio_buffer()
# Anything queued is dropped; the next send_audio() starts the new response cleanly

Without this, the agent would keep playing its old sentence over the caller for a second or two before the new response started. Clearing the buffer makes the agent stop talking immediately.

The outgoing buffer's size is the buffer_size from your CallAudioConfig (see Audio Configuration).

AI Integration Examples

These show how to bridge a call's audio stream with a real-time AI model. Each defines a bridge function you call from your @sm.on_incoming_call handler in place of the echo loop from the Quick Start. The SessionManager setup is identical, so it is omitted here.

If you don't need that much control, VoiceAgent (below) does this bridging for you.

VoiceAgent: the same thing in three lines

VoiceAgent is a model-agnostic runner over the API above. It answers the call, opens a model session per call, streams audio both ways, handles barge-in, dispatches tool calls, and cleans up on hangup:

from agentduet import VoiceAgent
from agentduet_adapters.gemini import GeminiLive

VoiceAgent.from_env().run(
    GeminiLive(instruction="You are May, a warm phone concierge. Keep replies short.")
)

The model adapters live in a separate, open-source package, agentduet-adapters, covering Gemini Live, xAI Grok Voice, Alibaba Qwen-Omni and Amazon Nova Sonic. There are two ways to install them, and the right one depends on what you are doing:

pip install --pre "agentduet[adapters]"        # trying things out: SDK + every adapter, one command
pip install --pre "agentduet-adapters[gemini]" # production: SDK + only the provider you call

Prefer the second for anything you deploy. [adapters] pulls all four providers' SDKs, so a service that only calls Gemini would also ship the AWS and Qwen dependencies; naming the provider keeps the install to what you use. Either way, installing agentduet alone pulls only websockets, httpx, and abxbus. Writing your own adapter needs nothing from that package either, only the seam exported here:

from agentduet import AudioOut, Interrupted, ToolCall, TranscriptDelta, Usage, VoiceAgent

class MyModel:                                  # satisfies VoiceModel
    async def open(self):                       # once per call
        return MySession()

class MySession:                                # satisfies ModelSession
    async def push_audio(self, pcm: bytes): ...     # caller audio in
    def events(self):                               # yields AudioOut / Interrupted /
        ...                                         # ToolCall / TranscriptDelta / Usage
    async def send_tool_result(self, call_id, result): ...
    async def close(self): ...

VoiceAgent.from_env(tools=my_tool_handler, on_transcript=log_it).run(MyModel())

Pass tools= to answer ToolCall events, on_transcript= for live transcripts, and on_usage= for cumulative per-call token usage. To dial out instead of answering, use call_out(dest, model, subscriber=...), or place_call(...) from inside a running serve() so inbound and outbound share one connector client.

Gemini Live

Streams the caller's audio to Google's Gemini Live API and plays the model's audio back, clearing the buffer on interruption.

import asyncio, os
from google import genai
from google.genai import types
from agentduet import Call, CallClosedError

gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
MODEL = "models/gemini-3.1-flash-live-preview"
CONFIG = types.LiveConnectConfig(
    response_modalities=[types.Modality.AUDIO],
    system_instruction="You are a helpful and friendly AI assistant.",
)


async def start_gemini_session(call: Call):
    if not await call.answer():
        return
    async with gemini_client.aio.live.connect(model=MODEL, config=CONFIG) as session:
        async def to_gemini():
            async for chunk in call.caller.audio_stream():
                await session.send_realtime_input(
                    audio=types.Blob(data=chunk, mime_type="audio/pcm;rate=24000")
                )

        async def from_gemini():
            try:
                while True:
                    async for response in session.receive():
                        if content := response.server_content:
                            if content.interrupted:
                                await call.clear_send_audio_buffer()
                                break
                            elif content.model_turn:
                                for part in content.model_turn.parts:
                                    if part.inline_data:
                                        await call.send_audio(part.inline_data.data)
            except CallClosedError:
                pass  # caller hung up while the model was speaking

        await asyncio.gather(to_gemini(), from_gemini())

Wire it in:

from agentduet import IncomingCallNotification, new_session_id


@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 start_gemini_session(call)

Google ADK (Agent Development Kit)

For multi-agent orchestration, memory, and structured tools, bridge the call with the Google ADK Runner and LiveRequestQueue.

import asyncio, os
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.genai import types
from agentduet import Call, CallClosedError

APP_NAME = "agentduet_app"
USER_ID = "default_user"

agent = Agent(
    name="agentduet_agent",
    model="gemini-3.1-flash-live-preview",
    instruction="You are a helpful AI assistant talking over a phone call.",
)
session_service = InMemorySessionService()
runner = Runner(app_name=APP_NAME, agent=agent, session_service=session_service)


async def start_adk_session(call: Call):
    if not await call.answer():
        return

    # ADK requires the session to exist before run_live()
    await session_service.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=call.id
    )
    queue = LiveRequestQueue()

    async def to_adk():
        async for chunk in call.caller.audio_stream():
            queue.send_realtime(types.Blob(data=chunk, mime_type="audio/pcm;rate=24000"))

    async def from_adk():
        run_config = RunConfig(streaming_mode=StreamingMode.BIDI, response_modalities=["AUDIO"])
        try:
            async for event in runner.run_live(
                user_id=USER_ID,
                session_id=call.id,
                live_request_queue=queue,
                run_config=run_config,
            ):
                if event.interrupted:
                    await call.clear_send_audio_buffer()
                if event.content and event.content.parts[0].inline_data:
                    await call.send_audio(event.content.parts[0].inline_data.data)
        except CallClosedError:
            pass  # caller hung up while the model was speaking

    await asyncio.gather(to_adk(), from_adk())

Wire it in the same way, calling start_adk_session(call) after session.process_call(noti).

API Reference

SessionManager

The entry point. Hold one for the life of your process, inside async with SessionManager(config) as sm:.

Method	Description
@on_incoming_call	Decorator. Receives an IncomingCallNotification for every incoming call.
@on_incoming_message	Decorator. Receives an IncomingMessage for every incoming WhatsApp message.
await open_session(session_id, subscriber)	Open an ephemeral session for subscriber (get-or-create). session_id is any unique string you supply. Returns a Session.
await list_sessions(subscriber=None, offset=0, limit=50)	List active sessions for this connector (observability). Returns list[SessionInfo].
await setup_trigger_conditions(config)	Set which call and message flows the server routes to you. See Trigger Conditions.
start()	Begin delivering inbound notifications. Idempotent. run_forever() calls it for you; call it directly only if you drive your own loop.
await run_forever(install_signal_handlers=True)	Start delivery, then run until interrupted or disconnect() is called. By default it installs SIGINT/SIGTERM handlers for graceful shutdown, replacing any handlers your application already set for those signals. Pass install_signal_handlers=False if your app manages its own signals (or runs the SDK off the main thread) and call disconnect() to shut down.
await disconnect()	Disconnect and release all resources; wakes a pending run_forever(). Idempotent. async with exit calls it automatically.

Property: id (server-assigned connection id; None until connected).

Register your @on_incoming_call / @on_incoming_message handlers before calling run_forever() (or start()). Notifications that arrive between connecting and that call are buffered, not dropped, so nothing is missed while you wire up handlers.

Session

An ephemeral, per-subscriber handle from sm.open_session(). Use it to attach a call or send a message.

Method	Description
await process_call(noti)	Attach to a notified call; returns a ready-to-use Call.
await make_call(dest)	Create an outbound Call to dest (an Address), in CallState.NEW. Call dial() to ring.
await send_message(msg)	Send an outbound message (a BaseSendMessage subclass such as SendWAMessage). Returns a SendMessageResult.

Properties: id, subscriber.

Address

A remote endpoint: a network plus a value. Frozen and hashable, so it works as a dict key.

Constructor / field	Description
Address.telco(value)	A phone endpoint on the TELCO network.
Address.whatsapp(value)	A WhatsApp endpoint on the WA network.
network	The Network this address lives on.
value	The address within that network (a string).

IncomingCallNotification

Delivered to your @sm.on_incoming_call handler.

Field	Description
call_id	The call's id (also call.id after process_call).
subscriber	The principal the call runs on behalf of; pass it to open_session().
participant	The external party, as an Address.
network	Shorthand for participant.network.
created_at	Notify emit time, unix epoch seconds.

IncomingMessage

Delivered to your @sm.on_incoming_message handler (WA channel).

Field	Description
id	Message id.
subscriber	Your business identity; pass it to open_session().
participant	The sender, as an Address (your reply target).
network	Shorthand for participant.network.
payload	Raw WhatsApp webhook payload as a dict.

Call

A voice call with a media connection. Command methods (answer, dial, connect, whisper, barge, spy, disconnect, close) return a CommandResult (truthy on success) for operational failures. Connection-gone and programmer errors still raise.

Method	Description
await answer()	Answer an incoming call.
await dial(ring_time_seconds=60)	Ring an outbound call built by make_call(). Falsy with CALL_UNANSWERED / TIMEOUT if it does not answer.
await connect(ring_time_seconds=60)	Start a 3-way conference between caller, callee, and agent.
await whisper()	After connect(): agent is heard only by the subscriber.
await barge()	After connect(): agent is heard by both parties.
await spy()	After connect(): agent hears both parties, neither hears the agent.
await disconnect()	End the call for all parties.
await close()	Release the agent. After connect(), caller and callee stay connected; after answer() alone, the call ends for both.
await send_audio(audio_data)	Queue binary audio for sending.
await clear_send_audio_buffer()	Drop queued outgoing audio and stop playback (use on interruption).
await get_send_audio_buffer_size()	Current size of the outgoing buffer, in bytes.
@on_hangup	Decorator. Shorthand for on_call_event(CallEvent.HANGUP); fires once on hangup.
@on_call_event(event)	Decorator for call event handlers.
to_json() / from_json(data)	Serialize and reconstruct a Call for cross-node handoff (only in CallState.NEW).

Incoming audio is per party: iterate call.caller.audio_stream() / call.callee.audio_stream() (see CallParty below).

Properties: id, participant (Address), subscriber, caller (CallParty), callee (CallParty), state (CallState), audio_config (CallAudioConfig).

CallParty

One party of a call: call.caller or call.callee. Membership, not liveness (the call's two parties, not who is currently connected). You never construct one.

Member	Description
value	The party's address string (external participant.value on one side, the subscriber on the other).
audio_stream()	Async iterator of that party's isolated incoming audio chunks.
str(party)	Returns value; party == "some-value" compares against it.

CommandResult

Returned by Call command methods.

Field	Description
success	True on success. The object is truthy, so if await call.answer(): works.
error_code	Server error code, or "TIMEOUT" on a client-side timeout.
error_message	Human-readable detail.
payload	Server response payload when present.

SendMessageResult

Returned by session.send_message().

Field	Description
success	True on success.
response_content	Provider response payload when present.
error_code	A MessageErrorCode on failure.
error_content	Human-readable failure detail.

VoiceAgent

The high-level runner (see the walkthrough under AI Integration Examples). Construct it directly with a SessionManagerConfig, or from environment variables with from_env().

Method	Description
VoiceAgent(config, *, tools=None, on_transcript=None, on_usage=None, inbound=InboundCallMode.ALL)	tools answers ToolCall events, on_transcript receives TranscriptDeltas, on_usage receives Usage updates. For inbound, see below.
VoiceAgent.from_env(*, sample_rate=24000, tools=None, on_transcript=None, on_usage=None, inbound=InboundCallMode.ALL)	Build from AGENTDUET_API_KEY and AGENTDUET_CONNECTOR_UUID. Note the sample rate defaults to 24000 here (matching most realtime AI models), not the SDK-wide 16000.
run(model)	Blocking: connect and serve inbound calls with model until interrupted. Wraps serve().
await serve(model)	Async form of run().
call_out(dest, model, *, subscriber, ring_time_seconds=60)	Blocking: place one outbound call on its own connector client, serve it with model, then return. Wraps dial().
await dial(dest, model, *, subscriber, ring_time_seconds=60)	Async form of call_out(): a one-shot outbound call on its own client.
await place_call(dest, model, *, subscriber, ring_time_seconds=60)	Dial out on the client serve() is already running, so inbound and outbound share one connector client. Fire-and-forget: schedules the call and returns immediately. Requires is_serving.

Property: is_serving (True while serve() is running, which is when place_call() is available).

inbound= rewrites the connector's trigger conditions. On startup, serve() sends the equivalent of TriggerConditionsBuilder().inbound_call(inbound).build(), and a sent configuration is absolute (see Trigger Conditions above), so the message-flow toggles reset to their defaults. Pass inbound=None to leave the connector's existing server-side configuration untouched. If applying it fails, the error is logged and serving continues.

The adapter seam: VoiceModel and ModelSession

What a model adapter implements. Both are protocols: any object with matching methods works, no base class required.

Member	Description
await VoiceModel.open()	Returns a fresh ModelSession for one call. Called once per call.
await ModelSession.push_audio(pcm)	Receives each chunk of the caller's audio, in arrival order.
ModelSession.events()	Async iterator of ModelEvents (below).
await ModelSession.send_tool_result(call_id, result)	Receives the outcome of a dispatched ToolCall.
await ModelSession.close()	Called on hangup: end the model conversation.

Events an adapter emits (ModelEvent is their union type):

Event	Fields	How VoiceAgent reacts
AudioOut	pcm	Sends it to the call. A full send buffer drops the chunk and the stream continues.
Interrupted	(none)	Clears the send-audio buffer (barge-in).
ToolCall	id, name, args	Dispatches to your tools handler and returns the result via send_tool_result. A failing tool or a missing handler returns an error result to the model instead of stalling it.
TranscriptDelta	text, role ("user" or "agent")	Passed to on_transcript if set, else dropped.
Usage	total, input, output, detail	Cumulative per call. Passed to on_usage if set, else dropped.

Handler signatures are exported as type aliases: ToolHandler is async (name, args) -> dict, TranscriptHandler is async (TranscriptDelta) -> None, and UsageHandler is async (Usage) -> None.

Error Handling

Operational command failures (server success=false, timeouts) come back as a CommandResult or SendMessageResult, not an exception. Check the return value for those. Exceptions are raised for connection loss, programmer errors, and auth or session problems.

Catch AgentDuetError to handle any SDK error, or a subclass for finer control.

• AgentDuetError: base for all SDK errors.
• TransportError: the session-manager link failed or was lost. A lost voice connection does not raise this; it surfaces as CallClosedError on call operations (a dead media link ends the call).
• RequestTimeoutError: a session-manager request got no response in time. The link was up but the server did not answer; distinct from TransportError (link actually gone) and deliberately not a subclass of it.
• AuthenticationError: auth failures (API key, token, or mTLS).
• MessageError: an unsupported message type was passed to send_message() (server-side send failures come back as a falsy SendMessageResult instead).
• CallError: base for call errors.
  • CallClosedError: the call is over. Raised for an operation on an already-terminated call, and when the voice connection is lost mid-operation (a dead media link ends the call). Treat it as the signal to stop working on that call.
  • CallStateError: operation attempted in an incompatible call state.
  • CallCommandError / CallCommandTimeoutError: a command was rejected or timed out (usually surfaced as a CommandResult).
• SessionError: base for session problems.
  • SessionAlreadyExistsError: a session with that id already exists on the server.
  • SessionNotFoundError: the session id does not exist on the server.
  • SubscriberMismatchError: the session subscriber does not match the call's.
  • ParticipantsFullError: the session already holds the maximum participants.
• CallNotFoundError: no pending call for the given id (expired or already taken).
• ChannelNotConfiguredError: the channel the operation needs (for example WhatsApp messaging) is not configured on the server.
• QuotaExceededError / OutboundOverflowError / ForbiddenError / InvalidRequestError: request-level rejections.
• BufferFullError: outgoing audio buffer is full.

Unanswered calls with connect()

When you use call.connect(), the callee may not pick up within the ring time. The server returns CALL_UNANSWERED on the CommandResult. The call stays active, so you can retry, try another number, or end it.

result = await call.connect(ring_time_seconds=30)
if not result:
    if result.error_code == "CALL_UNANSWERED":
        await call.disconnect()
    else:
        logger.error("Connect failed: %s (%s)", result.error_message, result.error_code)

WhatsApp message errors

When session.send_message() fails, the returned SendMessageResult is falsy and its error_code holds a MessageErrorCode:

MessageErrorCode	Meaning
QUOTA_EXCEEDED	Message sending limit exceeded for this connector.
CHANNEL_NOT_CONFIGURED	The session channel is not configured for messaging.
SESSION_BUSY	The session is bound to another live connection.
SESSION_NOT_FOUND	The session id does not exist on the server.
SESSION_CLOSED	The session has already closed.
SESSION_ALREADY_EXISTS	A session with that id already exists.
SUBSCRIBER_MISMATCH	The subscriber does not match the session.
PARTICIPANTS_FULL	The session already holds the maximum participants.
CALL_NOT_FOUND	No pending call for the given id.
INVALID_REQUEST	Request payload format or structure is invalid.
REMOTE_ERROR	The provider (for example, Meta/WhatsApp) rejected the message.
UNKNOWN	An unrecognized code (forward-compatible fallback).

Failure Modes and Delivery Semantics

Things the SDK handles for you, and the parts your application owns:

• Redelivery and deduplication. Inbound delivery is at-least-once: a notification can be redelivered after a reconnect or a competing-consumer reclaim. Dedup inbound messages on IncomingMessage.id (a server-generated unique id) and calls on IncomingCallNotification.call_id.
• Ordering. Delivery is connector-wide competing-consumer: notifications may arrive concurrently and out of order. Correlate by (subscriber, participant) yourself; there is no per-session ordering guarantee.
• Session idle eviction. Server sessions have a 30-minute sliding TTL. If a session idle-evicts, the next process_call / make_call / send_message transparently re-opens it and retries once; you don't handle SESSION_NOT_FOUND yourself.
• Reconnects. The session-manager link auto-reconnects with jittered exponential backoff. Established calls are unaffected (the voice WebSocket is independent). If the voice transport drops mid-call, the call fires on_hangup and any in-flight command raises CallClosedError promptly.
• Inbound audio backpressure. Each track's receive queue is bounded (CallAudioConfig.inbound_queue_maxsize, default 1000 chunks); when full, the oldest chunk is dropped so a slow consumer can't exhaust memory. Consume a party's audio_stream() promptly, or don't start it.
• Outbound audio backpressure. send_audio() raises BufferFullError when the outgoing ring buffer is full; throttle your producer on it.
• Audio after hangup. send_audio() and clear_send_audio_buffer() raise CallClosedError once the call terminates; treat it as the stop signal (see Event Handling).
• Shutdown. Leaving the async with SessionManager(...) block (or a signal in run_forever()) cancels any still-running on_incoming_call handler tasks. Put call cleanup (e.g. await call.close()) in a try/finally inside your handler.

Logging

The SDK uses Python's standard logging and follows library best practice: it attaches a NullHandler and configures nothing itself, so your application stays in control. You see nothing until you configure logging.

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)

# Turn up detail for one area
logging.getLogger("agentduet").setLevel(logging.DEBUG)
logging.getLogger("agentduet.voice_session").setLevel(logging.DEBUG)

Loggers are hierarchical under agentduet, so setting the level on agentduet covers everything, or you can target a submodule. Levels follow the usual meaning: DEBUG for wire-level detail (commands sent, message types), INFO for lifecycle events, WARNING for reconnects and missing data, ERROR for failures.

The SDK never logs secrets: no API keys, tokens, certificates, or private keys. Only connection URLs and call ids appear in logs, for debugging.

Thread Safety and Concurrency

The SDK is built on asyncio and handles concurrency for you:

• Each incoming call runs in its own task, so multiple calls are handled concurrently.
• The heartbeat and the reconnect logic each run in their own background tasks.
• All operations are non-blocking, and the SDK manages task lifecycle and cleanup.

You write ordinary async/await code in your handlers; the SDK takes care of the rest.

Support

Questions, bug reports, and feature requests: email [e-mail]. To get an API key and connector UUID, visit agentduet.com.