Introduction

Quick Start

View as Markdown

Minimal programs for inbound voice, WhatsApp reply, and outbound dial using agentduet 1.0.0.

Prerequisites

  • Python 3.12+ and pip install agentduet==1.0.0
  • AGENTDUET_API_KEY and AGENTDUET_CONNECTOR_UUID from agentduet.com

Incoming calls and WhatsApp messages arrive at the connector. Register handlers on SessionManager. Each notification includes addressing (subscriber, participant). Open a short-lived Session, then act:

  • Call → session.process_call(noti) returns a ready Call
  • Message → session.send_message(SendWAMessage(...)) sends the reply

session_id is any unique string. Reuse it to continue a conversation; create a new one with new_session_id() to start fresh.

Step 1: Answer a call (echo)

The following program answers every inbound call and echoes caller audio. Use it to verify connector and media connectivity before attaching a model.

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() -> 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:
        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
            except CallClosedError:
                pass  # normal end
        await sm.run_forever()
if __name__ == "__main__":
    asyncio.run(main())

Step 2: Reply to a WhatsApp message

Register @sm.on_incoming_message on the same SessionManager. Open a session for msg.subscriber and send. WhatsApp must be configured on the connector.

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. Inspect it to decide how to reply. Full channel notes: WhatsApp Messaging.

Step 3: Authentication refresher

API key (dev / most setups):

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

mTLS:

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

Full mTLS echo program: Installation - mTLS.

Step 4: Place a call

Outbound audio in is on callee, not caller. Set AGENTDUET_SUBSCRIBER (your connector calling identity) and DESTINATION_NUMBER.

import asyncio
import logging
import os
from agentduet import (
    Address,
    CallAudioConfig,
    CallClosedError,
    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),
    )
    subscriber = os.environ["AGENTDUET_SUBSCRIBER"]
    destination = os.environ["DESTINATION_NUMBER"]
    async with SessionManager(config) as sm:
        session = await sm.open_session(new_session_id(), subscriber)
        call = await session.make_call(Address.telco(destination))
        if not await call.dial(ring_time_seconds=30):
            logger.error("Dial was not answered")
            return
        try:
            async for chunk in call.callee.audio_stream():
                await call.send_audio(chunk)  # echo callee → outbound
        except CallClosedError:
            pass
        await call.close()
if __name__ == "__main__":
    asyncio.run(main())

Step 5: Attach a realtime model

Replace the echo loop with a bidirectional bridge to a speech model. Integration tutorials:

On barge-in, call await call.clear_send_audio_buffer(). Do not use private interrupt helpers.

Next Step

Architecture

Understand how SessionManager, subscriber Sessions, and Calls coordinate telephony transport, WhatsApp messaging, and audio buffers.