agentduet 1.0.0b4

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

AgentDuet Python SDK

Build AI agents that talk to people over real phone and WhatsApp calls. Answer an incoming call, stream the caller's audio into your AI model, stream the model's voice back, and handle interruptions the instant they happen. The same session also sends and receives WhatsApp messages, so one agent can speak and chat on the same channel.

Features

• Voice and text on one session. A Session is your live channel to a contact. Stream real-time audio through a Call, and send or receive WhatsApp messages on the same session (SMS coming).
• 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.
• Works across channels. Phone (TELCO) and WhatsApp (WA) today, with room for new platforms under the same interface.
• Pick your auth. API key, or mTLS client certificates.
• Stays connected. Heartbeat, connection monitoring, and automatic reconnect with exponential backoff.
• Race-free startup. Nothing is delivered until you call ready(), so you never miss an event while wiring up handlers.
• 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 connector.

Installation

pip install agentduet

The package is imported as agentduet. Requires Python 3.11+.

Quick Start

The SDK centers on the Session, your live channel to one contact. A session carries two modalities:

• Voice arrives as a Call object: you answer it, then stream audio in and out.
• Text you handle directly on the session: @session.on_incoming_message to receive, session.send_message() to send.

Set AGENTDUET_API_KEY and AGENTDUET_CONNECTOR_UUID in your environment, then pick a modality below.

Voice

This example handles an inbound call: the server notifies you of a session, you open_session() to claim it, register your handlers, then call ready(). It answers each call and echoes the caller's audio back.

import asyncio
import logging
import os
from agentduet import SessionManager, SessionManagerConfig, SessionNotification, Call, CallAudioConfig

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=24000),
    )
    async with SessionManager(config) as sm:
        logger.info("Connected. Waiting for sessions...")

        @sm.on_session_notification
        async def handle_session(noti: SessionNotification):
            session = await sm.open_session(noti.session_id)

            @session.on_incoming_call
            async def on_call(call: Call):
                logger.info("Incoming call %s from %s", call.id, call.caller_number)

                @call.on_terminated
                def on_terminated():
                    logger.info("Call %s terminated", call.id)

                if not await call.answer():
                    logger.error("Answer failed")
                    return

                async for audio_chunk in call.audio_stream():
                    await call.send_audio(audio_chunk)   # echo back

                await call.close()

            await session.ready()      # register handlers first, then start delivery

        await sm.run_forever()

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

Note: the server delivers nothing until you call ready(), so register your handlers first. This is what makes startup race-free.

To read and send audio in separate tasks instead of one loop, use asyncio.TaskGroup with a queue. The AI integration examples further down do exactly this.

Text (WhatsApp)

A chat-only agent uses the exact same setup. Swap the call handler for a message handler: register @session.on_incoming_message and reply with session.send_message().

msg.content 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; the example below just acknowledges every message.

from agentduet import IncomingMessage
from agentduet.messages import SendWAMessage

@session.on_incoming_message
async def on_message(msg: IncomingMessage):
    logger.info("Message from %s: %s", msg.sender, msg.content)

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

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, or a prebuilt ssl_context):

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 it for the defaults.

Field	Default	Notes
sample_rate	16000	Hz. One of 8000, 16000, or 24000. Match it to your AI model (the integration examples use 24000 for Gemini).
audio_mode	AudioMode.MIXED	MIXED delivers all call legs combined into a single stream. ISOLATED keeps each leg as a separate channel, which you read with call.audio_stream(channel_id=...).
buffer_size	1 MB	Outgoing ring-buffer size in bytes; must be a power of two.

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 runs on three connection layers 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 notifies you of new sessions. You hold a single SessionManager for the life of your process.
2. Session. Your channel to one contact. Text operations (send_message(), @session.on_incoming_message) act on the session directly. Voice calls arrive as a Call on @session.on_incoming_call.
3. Voice session. 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(), connect(), send_audio(), and so on) and closes it with the call; you never manage it directly, you just drive the Call.

How an inbound session reaches your code

1. The server sends a session.notify. The SDK delivers it to your @sm.on_session_notification handler as a SessionNotification carrying session_id, remote_address, local_address, channel, activity, and created_at.
2. You call await sm.open_session(noti.session_id) to claim it, register the session's handlers, then call await session.ready().
3. Once you are ready, the server starts delivering events. A voice call arrives as a Call on your @session.on_incoming_call handler. Answering it opens the media connection for audio.

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, to stop receiving inbound calls, or to enable outbound call and message events. Build the conditions with TriggerConditionsBuilder and send them once after connecting.

from agentduet import InboundCallMode, TriggerConditionsBuilder

trigger_config = (
    TriggerConditionsBuilder()
    .inbound_call(InboundCallMode.ALL)   # deliver all incoming calls
    .outbound_call(True)                 # also deliver outbound call events
    .inbound_message(True)               # deliver incoming WhatsApp messages
    .outbound_message(True)              # also deliver outbound message events
    .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.

Two things to keep in mind:

• The builder starts from everything off. Any flow you do not enable in the config will be turned off when you send it. Set every flow you want, not just the one you are changing.
• 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. 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.

@session.on_incoming_call
async def handle_call(call: Call):
    if not await call.answer():
        logger.error("Failed to answer")
        return
    async for chunk in call.audio_stream():
        response = await ai_model.generate(chunk)
        await call.send_audio(response)
    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(), you can switch the agent's audio between three modes:

• whisper(): only the number subscriber hears the agent (the callee on an incoming call, the caller on an outgoing one).
• barge(): both parties hear the agent.
• spy(): the agent hears both parties, neither hears the agent.

@session.on_incoming_call
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.

@session.on_incoming_call
async def forward_call(call: Call):
    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")
    return

Event Handling

The SDK gives you two ways to receive things from a call, depending on whether it is a one-time event or a continuous stream.

Decorators for discrete events

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

@session.on_incoming_call
async def handle_call(call: Call):
    await call.answer()

    @call.on_terminated
    def on_terminated():
        print("Call ended")

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

on_terminated is shorthand for on_call_event(CallEvent.CALL_TERMINATED). A CALL_ERROR handler receives a dict with error_code and an optional error_message.

Async iterators for continuous streams

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

async for audio_chunk in call.audio_stream():
    processed = await process_audio(audio_chunk)
    await call.send_audio(processed)

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 @session.on_incoming_call handler in place of the echo loop from the Quick Start. The SessionManager setup is identical, so it is omitted here.

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

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.audio_stream():
                await session.send_realtime_input(
                    audio=types.Blob(data=chunk, mime_type="audio/pcm;rate=24000")
                )
        async def from_gemini():
            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)
        await asyncio.gather(to_gemini(), from_gemini())

Wire it in:

@session.on_incoming_call
async def on_call(call: Call):
    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

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.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"])
        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)

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

Wire it in the same way, calling start_adk_session(call) from your @session.on_incoming_call handler.

API Reference

SessionManager

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

Method	Description
@on_session_notification	Decorator. Receives a SessionNotification for every new inbound session.
await open_session(session_id)	Claim a session by id. Returns a Session. Idempotent.
await create_session(channel, remote_address)	Start an outbound (SDK-initiated) session. Returns a Session.
await list_sessions(channel=None, remote_address=None)	List OPEN sessions for this connector.
await setup_trigger_conditions(config)	Set which call and message flows the server routes to you. See Trigger Conditions.
await run_forever()	Run until interrupted or the context manager exits.

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

SessionNotification

Delivered to your @sm.on_session_notification handler.

Field	Description
session_id	Pass this to open_session().
remote_address	Remote party (the caller, for inbound calls).
local_address	Local party (the callee, for inbound calls).
channel	Channel type, for example TELCO or WA.
activity	A SessionActivityType: NEW_INCOMING_CALL, NEW_INCOMING_MESSAGE, or RESUME.
created_at	Notify emit time, unix epoch seconds.

Session

Your channel to one contact. Register handlers, then call ready().

Method	Description
@on_incoming_call	Decorator. Receives the Call for a voice session.
@on_incoming_message	Decorator. Receives an IncomingMessage (WA channel).
@on_closed	Decorator. Runs when the session closes.
await ready()	Start event delivery. Idempotent; no-op on a closed session.
await send_message(...)	Send an outbound message (WA channel).
await wait_closed()	Wait until the session closes.
await close(...)	Close the session and release resources.

Properties: id, channel, remote_address, local_address, opened_at (unix epoch seconds), is_closed.

IncomingMessage

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

Field	Description
sender	Sender address.
content	Raw message payload as a dict (text, type, and so on).

Call

A voice call with a media connection. Command methods (answer, 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 the call.
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 number 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.
audio_stream(channel_id=0)	Async iterator of incoming audio chunks.
@on_terminated	Decorator. Shorthand for on_call_event(CallEvent.CALL_TERMINATED).
@on_call_event(event)	Decorator for call event handlers.
to_json() / from_json(data)	Serialize and reconstruct a Call for cross-node handoff.

Properties: id, caller_number, callee_number, state (CallState), audio_config (CallAudioConfig).

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.

Error Handling

Operational command failures (server success=false, timeouts) come back as a CommandResult, 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.
• ConnectionError: connection problems (session manager or media).
• AuthenticationError: auth failures (API key, token, or mTLS).
• CallError: base for call errors.
  • CallClosedError: operation attempted on a closed call.
• SessionError: base for session problems.
  • SessionAlreadyExistsError: an outbound session for this address is already open. Carries session_id, which you can pass to open_session() to claim it.
  • SessionBusyError: the session is already bound to another live connection. Carries bound_sm_ws_id.
  • SessionClosedError: the session is closed on the server.
  • SessionNotFoundError: the session id does not exist on the server.
• WrongChannelError: operation does not fit the session's channel (for example, messaging on a TELCO session).
• BufferFullError: outgoing audio buffer is full.
• BufferClosedError: write attempted on a closed buffer (for example, after the call ended).

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
INVALID_PAYLOAD	Message payload format or structure is invalid.
SESSION_NOT_FOUND	The session id does not exist on the server.
SESSION_NOT_OPEN	The session has already closed.
INVALID_CHANNEL	The session channel does not support text messaging.
OVERFLOW	Message sending limit exceeded for this connector.
INTERNAL_ERROR	Server error while processing the message.
REMOTE_ERROR	The provider (for example, Meta/WhatsApp) rejected the message.

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.

Requirements

• Python 3.11+
• websockets>=15.0
• httpx>=0.27
• abxbus>=2.4