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 messages on WhatsApp and DDUET.
• 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, mTLS client certificates, or short-lived access tokens your app hands over on demand.
• 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 agentduet

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 "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]". The adapters package is still a pre-release, so installing it by name needs --pre; the agentduet[adapters] extra does not, because its requirement already names a pre-release version. 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: a subscriber_id (the principal the event belongs to) plus from_address and to_address — whichever of the two is not you is the external party — and a session_id naming the session of yours the server bound it to, or None while that party is anonymous. To act on one, you open a short-lived Session for that subscriber with sm.open_session(session_id, subscriber_id):

• 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. Each channel addresses the recipient its own way: WhatsApp reads it from the payload to, while DDUET takes it as participant on the message object.

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.from_address)

            session = await sm.open_session(new_session_id(), noti.subscriber_id)
            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())

Claim promptly. Each call.notify gives you 10 seconds to claim the call (process_call), so claim first and do slow work afterwards. A call nobody claims does not just lapse: the server applies your connector's call_fallback policy (see Trigger Conditions), connecting it without the agent or hanging it up, and a late or second claim raises CallNotFoundError. A notification can arrive up to 15 seconds after the call started, when your connector was at its session cap or the caller already had a live call on the session it was bound to. Its 10 seconds still start when it arrives.

Reply to a WhatsApp message

Register @sm.on_incoming_message. Each IncomingMessage carries the subscriber_id (your business identity), the from_address (the customer to reply to), the to_address (you), the conversation it belongs to, and the raw webhook payload. To reply, open a session for your subscriber and call send_message(). Passing the message's conversation back is optional; it must be echoed, never built: on WhatsApp it is an opaque id the server derives per customer, stable forever.

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.from_address, msg.payload)

    session = await sm.open_session(new_session_id(), msg.subscriber_id)
    result = await session.send_message(
        SendWAMessage(
            api_version="v23.0",
            data={
                "messaging_product": "whatsapp",
                "type": "text",
                "to": msg.from_address.value,
                "text": {"body": "Thanks, we got your message!"},
            },
        ),
        conversation=msg.conversation,  # echo it back; never build one yourself
    )
    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.

Reply to a chat (DDUET)

DDUET is a plaintext text/media chat relayed between a Nexus BaChat session and your agent. Inbound messages arrive the same way as WhatsApp — via @sm.on_incoming_message — and are distinguished by msg.network is Network.DDUET. Two things set DDUET apart:

• Identities are Nexus account uids, not emails or phone numbers. msg.from_address is the conversation's other party (a stable identity, not the conversation key); correlate a conversation on msg.dduet.session_uid instead, so two concurrent sessions of the same party stay distinct.
• msg.payload is the Nexus relay body. Read it through msg.dduet — a typed view of the fields you route on (session_uid, ba_uid, user_uid, parts, …), with .raw for anything this SDK does not model yet.
• To reply, open a session for your subscriber — on DDUET that is the acting BA, msg.subscriber_id (equal to msg.dduet.ba_uid), not the connector uuid — and send a SendDduetMessage carrying session_uid from msg.dduet plus participant from msg.from_address.value. The server routes the reply back to Nexus for you. The acting BA is the session's subscriber_id, never a field on the send: opening the session for msg.subscriber_id is what names it, so a connector backing several BAs neither has to pass it twice nor can pass two contradictory values.

from agentduet import IncomingMessage, Network, SendDduetMessage, new_session_id


@sm.on_incoming_message
async def on_message(msg: IncomingMessage):
    if msg.network is not Network.DDUET:
        return  # WhatsApp / phone handled elsewhere

    d = msg.dduet
    session = await sm.open_session(new_session_id(), msg.subscriber_id)
    result = await session.send_message(
        SendDduetMessage.text(
            "Thanks, we got your message!",
            participant=msg.from_address.value,
            session_uid=d.session_uid,
        )
    )
    if not result.success:
        logger.error("Send failed: %s (%s)", result.error_code, result.error_content)

msg.dduet.parts is the Nexus MessageContent parts list (text/media items, proto3-JSON); text-first, inspect it to handle richer content. The DDUET message relay is text/media only; a DDUET voice call goes through the ordinary call path. When your agent dials one (session.make_call with a DDUET address), the server writes its call-log row (a call part) into the conversation once the call ends. Don't post one yourself.

Who wrote a message is msg.dduet.user_uid vs msg.dduet.ba_uid, never the sender role string — every BA member relays as AGENT. ba_uid is always the BA you act for, so user_uid != ba_uid means the other party wrote it and user_uid == ba_uid means a human on your own side did (a takeover). from_address stays the conversation's other party either way, so a bot that should not talk over its colleagues skips those (see examples/dduet_echo_bot.py).

You can also start a conversation rather than only replying to one. Omit session_uid and pass the target's account uid as participant; Nexus mints a BaChat session and the server hands its uid back on the ack:

result = await session.send_message(
    SendDduetMessage.text(
        "Your order has shipped.", participant="acct-of-the-other-party"
    )
)
session_uid = result.dduet_session_uid  # carry this on later sends of this conversation

If your connector backs several BAs, the one you are acting for is the session's subscriber_id — open the session for it. A send with no session_uid still mints; on later sends pass session_uid=session_uid.

Three result.error_code values are worth handling on this channel:

Code	What it means	What to do
SESSION_EXPIRED	The BaChat session passed its 24h idle expiry.	Re-send with no session_uid (mint a fresh session) under a new client_msg_id.
AMBIGUOUS_BA	The connector backs several BAs and the send named none.	Open the session for the acting BA — usually a sign it was opened for something other than the inbound msg.subscriber_id.
FORBIDDEN	No open member row on that session, or a BA this connector does not back.	Not retryable; check the connector's BA bindings.

Hand a conversation to your staff

When the agent should step back, hand the conversation to the BA's human staff. Say goodbye first, then hand over: the server sends the two to Nexus in that order, so there is no need to wait between them. The optional summary is for staff only; the customer never sees it.

@sm.on_incoming_message
async def on_message(msg: IncomingMessage):
    if msg.network is not Network.DDUET or not wants_a_human(msg):
        return
    session = await sm.open_session(new_session_id(), msg.subscriber_id)
    sent = await session.send_message(
        SendDduetMessage.text(
            "I'm passing you to a colleague, who will reply here shortly.",
            participant=msg.from_address.value,
        ),
        conversation=msg.conversation,
    )
    if not sent.success:
        # Your call: retry the goodbye, or hand over without one (the handover can't be undone).
        logger.warning("Goodbye failed: %s", sent.error_code)
    result = await session.hand_over_conversation(
        msg.from_address,
        msg.conversation,
        summary="Customer wants a refund for order #42; the payment cleared on 3 Oct.",
    )
    if not result:
        logger.error("Handover failed: %s (%s)", result.error_code, result.error_content)

A handover cannot be undone; only a staff member can hand the conversation back. After it:

• The conversation's messages stop reaching your agent, and a send_message into it returns FORBIDDEN. Your agent's own A2UI function calls still arrive.
• A DDUET call in the conversation still rings your agent until Nexus routes such calls to staff. Claim it, or leave it unclaimed for your call_fallback.

On REMOTE_ERROR (or a timeout) the outcome may be unknown, and retrying is safe. If the first attempt worked, the retry returns FORBIDDEN: the conversation is already with your staff. The SDK does not retry for you. Handover is DDUET only; a WhatsApp or phone participant raises ValueError and nothing is sent.

Outbound retry (application-owned)

The SDK does not auto-retry outbound sends. Delivery is at-least-once and retry is the application's responsibility:

• Pass a stable client_msg_id and reuse the same value on every re-send of one logical message. The server dedups by it: a completed send replays its original result (no duplicate), an interrupted send re-POSTs.
• On SendMessageResult(success=False, error_code=SEND_IN_PROGRESS), back off and retry with the same client_msg_id.
• On TransportError (connection lost mid-send), retry after the connection re-establishes, again with the same client_msg_id.
• A re-send after an ambiguous failure may duplicate downstream — size retries accordingly.

Downloading DDUET media

A media message's payload.content parts carry the attachment metadata but no bytes. Open a session and hand the message's attachments straight to download_media_all:

@sm.on_incoming_message
async def on_message(msg):
    dduet = msg.dduet
    if dduet is None or not dduet.attachments:
        return

    session = await sm.open_session(new_session_id(), msg.subscriber_id)
    for got in await session.download_media_all(dduet.attachments):
        if got.ok:
            print(got.attachment.media_uid, got.attachment.mime_type, len(got.data))
        else:
            print(got.attachment.media_uid, "failed:", got.error)

DduetMessage.attachments already stamps every attachment with its message's session_uid and ba_uid, so nothing has to be re-typed. For a single attachment, await session.download_media(attachment) returns the bytes and raises on failure.

Per-item failures never fail the batch. Results come back in input order, one per input, and ok discriminates. A MediaNotFoundError (404) is the expected outcome for a uid that has been reaped by the 7-day GC or belongs to another session — under the previous protocol those uids were silently omitted from the response, and now they are reported per item instead. A 429 from Nexus's own door is the one batch-fatal case: the remaining items are marked with the same QuotaExceededError, whose retry_after carries the server's hint.

The credential never reaches your code. Each download is a URL plus a Bearer token that lives at most 60 seconds, and the SDK mints them a wave at a time — concurrency tokens, spent immediately, then the next wave — so nothing is ever minted into a queue. That costs one mint frame per wave against the connector's 60-per-minute budget, so a caller who wants fewer frames raises concurrency rather than pre-minting. max_total_bytes (64 MiB by default) is checked against the attachments' declared sizes before any frame is sent, because this call returns bodies, not URLs.

Previews. A media part embeds no thumbnail bytes, so the zero-fetch preview is the attachment's blurhash. Where an attachment carries a thumbnail_media_uid, that is a second media uid — attachment.thumbnail is that sibling, already carrying the same routing, so it rides the same batch and you get both sizes in one pass. (The inline WebP you may have read about lives on a reply snapshot, which is why a reply renders without fetching anything.)

Sending DDUET media

Upload the bytes, then reference the returned uid in a message. The upload is a separate step because a DDUET message is a list of parts: text plus several attachments needs every media_uid in hand before the message can be built.

from agentduet import SendDduetMessage, UploadDduetMedia


async def reply_with_chart(msg, png_bytes: bytes):
    dduet = msg.dduet
    session = await sm.open_session(new_session_id(), msg.subscriber_id)
    uploaded = await session.upload_media(
        UploadDduetMedia(data=png_bytes, filename="chart.png", mime_type="image/png")
    )
    await session.send_message(
        SendDduetMessage.compose(
            text="Here is this month's chart",
            media=[uploaded],
            participant=msg.from_address.value,
            session_uid=dduet.session_uid,
        )
    )

Uploading does not attach anything: the msg.out part that references the uid is what attaches it. Nexus enforces its own size caps ahead of the token — 5 MB for an image, 16 MB for audio/video, 100 MB for a document — and refuses an oversized body with MediaTooLargeError.

Answer a function call (DDUET)

When a DDUET user taps a button in the conversation, such as "reveal my API key" or "pay now", your agent answers it directly with a confidential value that is never a chat message, never model context, and never logged by the SDK. Register one async handler per function name. Refuse a call with fn.reject(status), or open a session, claim the call onto it, do the work, and answer:

from agentduet import IncomingFunctionCall, new_session_id

_conversation_sessions: dict[tuple[str, str], str] = {}


def conversation_session_id(fn: IncomingFunctionCall) -> str:
    key = (fn.subscriber_id, fn.from_address.value)
    return _conversation_sessions.setdefault(key, new_session_id())


@sm.on_function_call("submitPayment")
async def submit_payment(fn: IncomingFunctionCall):
    if fn.user_uid not in allowed_payers:
        await fn.reject(403)                                  # error answer, no session
        return
    # The conversation's session: reuse it, don't mint one per tap.
    session = await sm.open_session(conversation_session_id(fn), fn.subscriber_id)
    await session.process_function_call(fn)                   # claim first; raises on refusal
    receipt = await psp.charge(fn.args["amount"], key=fn.idempotency_uid)
    outcome = await session.send_function_result(fn, status=200, result={"id": receipt.id})
    if not outcome.success:                                   # log ids, never the result
        logger.warning("payment %s not delivered: %s", fn.id, outcome.error_code)

What happens	The user gets
send_function_result(fn, status=..., result=...)	your status and result
fn.reject(status)	that status (400..599), no result
no handler registered for the name	501
the handler raised, or returned without answering	500
the claim finds the call gone (FunctionCallNotFoundError)	nothing more is sent; the user already has their 504
the handler was too slow for fn.remaining()	504, and your late answer returns FN_CALL_NOT_FOUND

Claim before the work. Every session-side refusal (from open_session or the claim) comes before anything has run, so a payment never goes through without a way to answer it; once claimed, the session is held until the call's deadline however long the work takes. Reuse the conversation's session (the one your agent already talks to this peer on) rather than minting one per tap: one session carries many claimed calls, while each new session holds a slot of the connector's session cap until it idles out, so a burst of taps would exhaust the cap and block your messages and calls. The SDK never cancels a slow handler, so check fn.remaining() before starting long work. fn.idempotency_uid is one per gesture and the server stores nothing, so a function that must run once (a payment, not a read) deduplicates a double tap on it. Each call runs in its own task, so a slow function never delays messages or phone calls. See examples/dduet_function_call.py.

When your connector is at its session cap, the server holds a function call for up to 30 seconds waiting for a slot. If none frees, it answers the user itself (503) and your handler never sees the call; one that waited arrives with fn.timeout_ms already reduced to the time left, so fn.remaining() stays right. fn.conversation is the BaChat conversation the tap came from.

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("+15551234567"))

    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 three modes when you build the config. All of them 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",
)

Access token (a token_provider callback, for an agent that signs in as a user):

Hand the SDK a zero-arg callable, sync or async, that returns the current access token. The SDK calls it fresh before every connect attempt (the first one and every automatic reconnect) and before every authenticated HTTP request, then sends the result as a bearer token. It never caches it, and the only check it makes is that you returned a non-blank string, so your app stays in charge of storage and refresh: a token that is still valid is simply handed back again. No connector UUID here: the token already says which connector it belongs to.

import asyncio
import time

from agentduet import SessionManagerConfig

class TokenStore:
    """Owns the access token: caches it, refreshes it, waits out a sign-out."""

    def __init__(self) -> None:
        self._token = ""
        self._expires_at = 0.0
        self._signed_in = asyncio.Event()
        self._lock = asyncio.Lock()

    async def get_access_token(self) -> str:
        # One refresh at a time: a reconnect and a list_sessions() can ask at the
        # same moment, and two concurrent refreshes race your rotating refresh token.
        async with self._lock:
            # Still good for another minute? Hand back what we already have.
            if self._token and time.monotonic() < self._expires_at - 60:
                return self._token
            try:
                self._token, lifetime = await refresh_access_token()   # your OAuth call
                self._expires_at = time.monotonic() + lifetime
            except InvalidGrant:
                # The refresh token is gone, so only the user can fix this. Wait for
                # them rather than raising, and the SDK just waits to reconnect.
                self._signed_in.clear()
                show_signin_screen()          # your UI calls signed_in() when done
                await self._signed_in.wait()
            if not self._token:
                # A blank token is a sign-out to the SDK, so fail loudly instead:
                # this means the sign-in screen finished without storing anything.
                raise RuntimeError("signed in but no access token was stored")
            return self._token

    def signed_in(self, token: str, lifetime: float) -> None:
        """Call this from your sign-in UI once the user is back."""
        self._token = token
        self._expires_at = time.monotonic() + lifetime
        self._signed_in.set()

store = TokenStore()          # keep the reference: your sign-in UI needs it
config = SessionManagerConfig.create(token_provider=store.get_access_token)

refresh_access_token, InvalidGrant, and show_signin_screen are your app's own OAuth pieces. The SDK never sees your refresh token, and the only thing it checks about an access token is that you returned a non-blank string. For the app side spelled out end to end (browser PKCE flow, a rotating refresh token stored on disk, and this provider wired into a running agent), see examples/oauth_signin_example.py.

Blocking like that is the point: the SDK puts no timeout on your provider, so waiting for the user to sign back in keeps run_forever() genuinely running forever and the agent resumes taking calls the moment a token arrives. Use an async def provider for anything that can block for long, because a synchronous one runs on a worker thread the SDK cannot interrupt, and a long block there delays your process exit. One consequence at startup: async with SessionManager(config) is no longer bounded by the socket-open and setup timeouts in this mode, since it waits for your provider first.

Raising from the provider is the fail-fast alternative. It is your "signed out" signal, and where the SDK reports it depends on which operation asked for the token. In all three cases you get TokenProviderError (a subclass of AuthenticationError) with your exception as its cause:

Provider raises during	You catch it on	What the SDK does
the first connect	async with SessionManager(config) (that is, __aenter__)	No socket was opened and no reconnect loop exists yet, so there is nothing to reconnect. Wrap the async with itself if you want to catch a sign-out at startup.
an automatic reconnect	await sm.run_forever()	The terminal path. The SDK stops reconnecting, shuts the manager down, and re-raises here. Drive your own loop with start() instead and you will not see this (see below).
an HTTP call such as list_sessions()	that call, and nowhere else	Nothing. The session-manager link is untouched and keeps running, so this is a normal per-call failure you can retry.

Returning an unusable token raises the same error, with no cause attached: anything that is not a string, and an empty or whitespace-only string too.

An expired token is not a sign-out, so a rejected handshake keeps retrying with backoff; the next attempt asks your provider for a fresh token.

Driving your own loop. The sign-out raise is a run_forever() guarantee. If you call start() and run your own loop instead (a desktop app whose UI framework already owns the loop, say), nothing raises it for you: your next SDK operation just fails on a connection the SDK has already closed, which looks the same as a transient drop. Read sm.fatal_error to tell the two apart. It is None while the SDK is healthy or merely reconnecting, and holds the TokenProviderError once the sign-out is terminal:

sm.start()               # then hand control to your own loop

if sm.fatal_error is not None:
    show_signin_screen() # signed out for good, not a network blip

Endpoint. The SDK talks to the production host by default. To point it elsewhere (a local server, say), pass base_url= to create() or set AGENTDUET_BASE_URL; the precedence is explicit base_url=, then the environment variable, then production.

config = SessionManagerConfig.create(
    api_key=os.getenv("AGENTDUET_API_KEY"),
    connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
    base_url="ws://localhost:8080",
)

API keys and bearer tokens both travel in plaintext, so the SDK logs a warning if you pair either with a non-loopback ws:// URL. mTLS is exempt: its credential never leaves the TLS layer.

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_id). 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_id, 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_id 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_id.

call.origin says who placed the call, which caller cannot: a call your agent placed and a call the subscriber placed from a desk phone both have caller == subscriber_id, yet the first still needs dialling and the second is already ringing. It is the dial-vs-answer decision:

call.origin	Built by	What to do
CallOrigin.AGENT	session.make_call(dest)	await call.dial() — place the SIP leg
CallOrigin.SUBSCRIBER	@on_outgoing_call → process_call(noti)	await call.answer() — join the subscriber's leg
CallOrigin.PARTICIPANT	@on_incoming_call → process_call(noti)	await call.answer()

from agentduet import CallOrigin

if call.origin is CallOrigin.AGENT:
    await call.dial()
else:
    await call.answer()

dial() raises CallStateError on any other origin, so a wrong guess fails fast rather than hanging on a leg that is already up.

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_id party (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, to stop receiving inbound calls, or to stop receiving incoming messages. 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 messages on every channel
    .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.

inbound_message controls incoming messages (WhatsApp and DDUET) the same way: True delivers them to your @on_incoming_message handlers, False stops delivery. Call and message routing are independent, so you can take calls only, messages only, or both.

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 toggle asks the server to notify you when a subscriber places a call outside the SDK (from a desk phone, say), so your agent can step in before it reaches the destination — see Handle a call your agent didn't place. The outbound_message toggle belongs to an upcoming feature and has 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.

Calls nobody claims: call_fallback

A notified call goes through a fallback when no agent claims it: your connector is offline, the notification could not be sent, nobody claimed it within its 10-second claim window, or no session slot freed during the 15 seconds the server waits when your connector is at its session cap. Declare what happens with call_fallback:

from agentduet import CallFallback, InboundCallMode, TriggerConditionsBuilder

trigger_config = (
    TriggerConditionsBuilder()
    .inbound_call(InboundCallMode.ALL)
    .call_fallback(CallFallback.CONNECT_TO_DESTINATION)  # or CallFallback.DROP
    .build()
)

Connect sends the call on to its destination without the agent; drop hangs it up. Leaving call_fallback unset sends nothing (undeclared):

Call	inbound_call	undeclared	CONNECT_TO_DESTINATION	DROP
incoming	ALL	connect	connect	drop
incoming	MISSED_ONLY or NO	drop	drop	drop
outgoing	(any)	connect	connect	drop

An incoming call connects only on an ALL connector: on MISSED_ONLY the subscriber already rang and did not answer, so connecting would ring them twice. Behaviour change: an unclaimed incoming call on a connector that is not ALL is now hung up, where it used to be left ringing until the voice network gave up.

A message before the drop: drop_message

With call_fallback(CallFallback.DROP), the server can speak a short message to the caller before it hangs up:

trigger_config = (
    TriggerConditionsBuilder()
    .call_fallback(CallFallback.DROP)
    .drop_message("Sorry, no one can take your call right now. Please try again later.")
    .build()
)

The text is spoken in one fixed English voice; it is trimmed and must be 1 to 500 characters. It plays only when you declared DROP: build() refuses it with any other call_fallback, and an incoming call dropped because inbound_call is not ALL hangs up without it. A later setup without drop_message clears it.

Recording calls the fallback connects: fallback_recording

Opt in, and the server records every call its fallback connects without the agent (a dropped call has nothing to record). The parties hear no announcement: by opting in, you take responsibility for consent.

trigger_config = (
    TriggerConditionsBuilder()
    .inbound_call(InboundCallMode.ALL)
    .call_fallback(CallFallback.CONNECT_TO_DESTINATION)
    .fallback_recording(True)
    .build()
)

Every setup_trigger_conditions replaces all the toggles, so a later setup without fallback_recording turns it off again.

Receive the recordings with one handler. The SDK lists them when the server announces one, when delivery starts, after a reconnect and every 10 minutes:

from agentduet import CallRecording

@sm.on_call_recording
async def on_recording(rec: CallRecording):
    audio = await sm.download_recording(rec.recording_id)   # Ogg Opus, stereo
    side = rec.audio_channels                                # left/right: "subscriber" or "participant"
    await store(rec.call_id, rec.participant, audio)

When the handler returns, the SDK acknowledges the recording and the server deletes it. Returning without downloading discards it. If the handler raises, the recording stays on the server and is offered again on the next listing; one never acknowledged expires after 30 days. Without a handler, the SDK lists and acknowledges nothing.

To decide yourself when a recording is deleted, register the handler with auto_ack=False and acknowledge once the file is safely stored, from the handler or later (a queue, another task):

@sm.on_call_recording(auto_ack=False)
async def on_recording(rec: CallRecording):
    await queue.put(rec)          # offered once per process; never acknowledged by the SDK

async def worker():
    while True:
        rec = await queue.get()
        await store(rec, await sm.download_recording(rec.recording_id))
        await sm.ack_call_recordings([rec.recording_id])   # deletes it on the server

A recording handed over but never acknowledged is not offered again by the same process; it comes back after a restart and expires after 30 days.

Or skip the handler and page through the recordings yourself, oldest ended_at first. Listing never acknowledges:

cursor = None
while True:
    page = await sm.list_call_recordings(limit=100, cursor=cursor)
    done = []
    for rec in page.recordings:
        await store(rec, await sm.download_recording(rec.recording_id))
        done.append(rec.recording_id)
    await sm.ack_call_recordings(done)    # up to any number of ids; idempotent
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

The cursor walks one listing and is not a bookmark: a recording can become ready behind a cursor already handed out, so start every listing from the first page. Without a handler the SDK does not list on its own, so poll as often as you need.

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. connect() leaves the agent in spy mode: it hears both parties but its audio is not injected until you switch modes. From there, move 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

Handle a call your agent didn't place

When a subscriber dials out from a desk phone (or any other app on the same line), the server can notify you the same way it does for incoming calls — enable outbound_call in your trigger conditions first. You get an OutgoingCallNotification whose to_address is the number being dialled, and you handle it exactly like an incoming call: answer() to speak to the subscriber first, then connect() to ring the destination, or connect() straight away.

from agentduet import OutgoingCallNotification, new_session_id

@sm.on_outgoing_call
async def on_outgoing(noti: OutgoingCallNotification):
    logger.info("%s is calling %s", noti.from_address, noti.to_address)

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

    await call.answer()          # talk to your own subscriber first (optional)
    await call.connect()         # then ring the destination

The resulting Call behaves like any other, except call.caller is your subscriber and call.callee the destination. Its call.origin is CallOrigin.SUBSCRIBER — the call is already placed, so dial() is rejected; answer() and connect() are the verbs.

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 "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_id=...), 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_id)
    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_outgoing_call	Decorator. Receives an OutgoingCallNotification for every call the subscriber places outside the SDK. Requires outbound_call in your trigger conditions.
@on_incoming_message	Decorator. Receives an IncomingMessage for every incoming message on any channel (WA, DDUET).
await open_session(session_id, subscriber_id, *, participant=None)	Open a handle to an ephemeral session for subscriber_id. session_id is any unique string you supply. A plain open stores nothing for a new id (the first send or claim creates the session) and refreshes a live one. Pass participant (an Address, e.g. msg.from_address) to adopt that party's slot into the session now. Returns a Session.
await list_sessions(subscriber_id=None, offset=0, limit=50)	List active sessions for this connector (observability). Returns list[SessionInfo]. With token_provider auth this resolves a fresh token per request, so it can raise TokenProviderError; that failure belongs to this call alone and leaves the session-manager link running.
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. With token_provider auth, note that only run_forever() raises a terminal sign-out: your own loop sees it as the next SDK operation failing on a closed connection.
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 rather than dropped, so the usual register-then-start window loses nothing. The buffer holds 1000 frames and then drops the oldest with a warning, which only bites if you connect and never start delivery.

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, *, conversation=None)	Create an outbound Call to dest (an Address), in CallState.NEW. Call dial() to ring. conversation, echoed from an inbound message or notification, names the conversation the call belongs to; on DDUET it is where the server logs the call row.
await send_message(msg, *, client_msg_id=None, conversation=None)	Send an outbound message (a BaseSendMessage subclass: SendWAMessage, SendDduetMessage). Returns a SendMessageResult. client_msg_id is the idempotency key: reuse the same value on every re-send of one logical message so the server dedups, and use a fresh one for a genuinely new message. Omit it and the SDK mints one. conversation is optional and echoed from the inbound message (never built); on DDUET it must agree with the message's session_uid, which is checked before sending.
async with keep_alive(interval=20.0)	Re-open the session every interval seconds while the block runs, so a slot adopted with open_session(..., participant=) survives a turn longer than the server's idle TTL.
await download_media(attachment)	Fetch one inbound attachment's bytes. Mints a 60-second token and spends it at once; raises on failure.
await download_media_all(attachments, *, max_total_bytes=64 MiB, concurrency=4)	Fetch a batch, returning one DownloadedMedia per input in input order. Per-item failures are reported, not raised. concurrency is also the mint-wave size: that many tokens are minted and spent immediately, then the next wave — so nothing is ever minted into a queue. At most Session.MAX_MEDIA_UIDS (100) attachments; max_total_bytes is pre-flighted against their declared sizes.
await upload_media(media)	Upload one file (an UploadDduetMedia) and get back an UploadedMedia carrying its media_uid. Uploading attaches nothing — the msg.out part referencing the uid is what does.

Properties: id, subscriber_id.

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.
Address.dduet(value)	The other party on a Nexus BaChat conversation, by account_uid, on the DDUET 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_id	The principal the call runs on behalf of; pass it to open_session().
from_address	The external caller, as an Address.
to_address	You — the same value as subscriber_id, as an Address.
network	Shorthand for participant.network.
created_at	Notify emit time, unix epoch seconds.
session_id	Your session the server bound the call to, or None while the caller is anonymous.
conversation	The call's conversation (DDUET convo-id, or an opaque per-person id on TELCO/WA), or None when a DDUET caller named none. Echo it on make_call(..., conversation=).

OutgoingCallNotification

Delivered to your @sm.on_outgoing_call handler. Same shape as IncomingCallNotification; the type tells you the direction, so neither carries a direction field.

Field	Description
call_id	The call's id (also call.id after process_call).
subscriber_id	Your own number — the line that placed the call; pass it to open_session().
from_address	You — the same value as subscriber_id, as an Address.
to_address	The external destination, as an Address.
created_at	Notify emit time, unix epoch seconds.
session_id	As on IncomingCallNotification.
conversation	As on IncomingCallNotification.

IncomingMessage

Delivered to your @sm.on_incoming_message handler (WA and DDUET channels).

Field	Description
id	Message id.
subscriber_id	Your business identity; pass it to open_session().
from_address	The sender, as an Address (your reply target).
to_address	You — the same value as subscriber_id, as an Address.
network	The Network of from_address (WA, TELCO, or DDUET). A channel this SDK version does not know parses to Network.UNKNOWN rather than raising, so branch with an explicit else.
payload	The channel's own inbound message as a dict, as that channel defines it: the WhatsApp webhook content node, or the Nexus relay body for DDUET. A channel's routing fields live in here — there is no sibling address object.
dduet	DDUET only: payload typed as a DduetMessage (below); None on other channels.
session_id	Your session the server bound the message to, or None while the sender is anonymous.
conversation	The conversation the message belongs to: Nexus's convoId on DDUET (same as dduet.session_uid), an opaque id derived per sender on WA. None only for a DDUET message that named none. Echo it on send_message(..., conversation=).

DduetMessage

msg.dduet — a typed view of an inbound DDUET relay body. Missing fields degrade to empty rather than raising, so a frame this SDK cannot fully parse still reaches your handler.

Field	Description
session_uid	The Nexus BaChat session — the conversation key. Pass it back as SendDduetMessage.session_uid. Nexus is renaming it convoId: read from convoId first, sessionUid second.
convo_id	The same value as session_uid, under Nexus's new name.
msg_uid	Nexus's id for this message.
ba_uid	The BA you act for, and what msg.subscriber_id equals on this channel. Open the reply's session for it; it is not a field on the send.
user_uid	The account_uid that authored this message. != ba_uid is the other party, == ba_uid is a human on your own side (a takeover).
content	The Nexus MessageContent, protobuf decoded to proto3-JSON.
parts	Shorthand for content["parts"] — the text/media items. [] when absent.
attachments	Every media attachment across parts, in part order, each already stamped with this message's session_uid and ba_uid — so the whole list is one argument to session.download_media_all(...). [] on a text-only message.
user_metadata	Display hint for user_uid: {"email", "name"}, either key possibly absent (email is always absent on a BA-authored relay). Never correlate on it; correlate on user_uid.
sender	Nexus's role string. Informational only — never use it to decide authorship: every BA member relays as AGENT.
raw	The whole payload dict, for fields Nexus adds before this SDK models them.

Call

A voice call with a media connection. Command methods (answer, dial, connect, whisper, barge, spy, disconnect, close, clear_send_audio_buffer) 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. The agent drops to a silent monitor (spy mode) until you call whisper() or barge().
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). Returns a CommandResult whose payload is the number of buffered bytes discarded.
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_id, caller (CallParty), callee (CallParty), origin (CallOrigin), 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.
dduet_session_uid	DDUET only: sugar for response_content["sessionUid"] on a successful send — the session Nexus just minted, or the one you echoed back. None for other channels and for failed sends. This is how you learn a minted session's uid; carry it on later sends of that conversation. Read from convoId first, sessionUid second.
dduet_convo_id	The same value as dduet_session_uid, under Nexus's new name.

SessionInfo

One active session, as returned by sm.list_sessions() (concurrency observability).

Field	Description
session_id	The session id.
subscriber_id	The principal the session runs for.
participants	The external parties attached to it, as a list[Address].
created_at	When the session was created, in unix epoch milliseconds.
last_activity_at	Last activity, in unix epoch milliseconds.

Both are passed through from the server unconverted.

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, AGENTDUET_CONNECTOR_UUID, and the optional AGENTDUET_BASE_URL. 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, access token, or mTLS).
  • TokenProviderError: your token_provider raised (the provider's own exception is the __cause__), or returned something the SDK cannot send: a non-string, or a blank one. Where it surfaces depends on the operation that asked for the token: async with entry on the first connect, run_forever() on a reconnect (the terminal path, where the SDK also stops reconnecting and shuts down), or the failing HTTP call itself. See Authentication.
• 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. QuotaExceededError is also what Nexus's own media door answers with on a 429, and there it carries retry_after (seconds, or None when the server sent no usable hint).
• Media transfer, from Nexus's own door rather than the platform's frame codes. These are reported per item by download_media_all and raised by download_media / upload_media:
  • MediaDownloadError: a fetch answered non-2xx; field status_code. Not a TransportError — the link was fine.
    • MediaNotFoundError: 404, the uid is not readable in this session. Expect this one: it is the ordinary outcome for a uid reaped by the 7-day GC or belonging to another conversation, and it is deliberately distinct from the uniform 401 a bad token gets.
  • MediaUploadError: the upload POST failed, or answered 2xx with no media_uid; field status_code.
  • MediaTooLargeError: 413, over Nexus's per-type cap (image 5 MB, audio/video 16 MB, document 100 MB), refused before the token is even read.
• 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)

Message errors

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

MessageErrorCode	Meaning
QUOTA_EXCEEDED	The connector's session cap is full and this send needed a new slot.
CHANNEL_NOT_CONFIGURED	Legacy; no current server build sends it.
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, or Nexus) rejected the message.
SEND_IN_PROGRESS	A send with that client_msg_id is still in flight; back off and re-send it.
OUTBOUND_MSG_OVERFLOW	The server's outbound queue for this subscriber is full; back off and re-send.
SERVER_DRAINING	The node is shutting down; reconnect and re-send.
SESSION_EXPIRED	DDUET: the BaChat session passed its 24h idle expiry — mint a new one.
FORBIDDEN	DDUET: no open member row on that session, or a BA this connector does not back.
AMBIGUOUS_BA	DDUET: the connector backs several BAs and the request named none. No SDK call can hit this: sends and media mints alike carry the session's subscriber as baUid. Seeing it means the server read none — report it.
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_id, from_address) yourself; there is no per-session ordering guarantee.
• Session idle eviction. Server sessions have a sliding idle TTL, a server setting that defaults to 60 seconds. 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 full-jitter exponential backoff (first retry within 1 s, capped at 8 s), and a silently dead link is detected within 20 s. 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 support@agentduet.com. To get an API key and connector UUID, visit agentduet.com.