agentduet 1.0.0b8

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

AgentDuet Python SDK

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

Features

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

Installation

pip install agentduet

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

Quick Start

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 only addressing: a subscriber (the principal the event belongs to) and the external participant, no media credentials. To act on one, you open a short-lived Session for that subscriber:

• For a call, session.inbound_call(noti) attaches the session to the call and returns a ready-to-use Call (media URL and token filled in).
• For a message, session.send_message(...) sends your reply; the server infers the recipient from the payload.

A session_id is any unique string you choose; reuse the same id to continue a conversation, or use a new one to start fresh. Get your API key and connector UUID at agentduet.com, set them as AGENTDUET_API_KEY and AGENTDUET_CONNECTOR_UUID in your environment, then 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
import uuid

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

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


async def main():
    config = SessionManagerConfig.create(
        api_key=os.getenv("AGENTDUET_API_KEY"),
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
        call_audio=CallAudioConfig(sample_rate=16000),
    )

    async with SessionManager(config) as sm:
        logger.info("Connected. Waiting for calls...")

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

            session = await sm.open_session(str(uuid.uuid4()), noti.subscriber)
            call = await session.inbound_call(noti)

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

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

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

        await sm.run_forever()


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

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

Reply to a WhatsApp message

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

import uuid

from agentduet import IncomingMessage, SendWAMessage


@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(str(uuid.uuid4()), msg.subscriber)
    result = await session.send_message(
        SendWAMessage(
            api_version="v23.0",
            data={
                "messaging_product": "whatsapp",
                "type": "text",
                "to": msg.participant.value,
                "text": {"body": "Thanks, we got your message!"},
            },
        )
    )
    if not result.success:
        logger.error("Send failed: %s (%s)", result.error_code, result.error_content)

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

Place a call

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

import uuid

from agentduet import Address


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

    if await call.dial(ring_time_seconds=30):
        logger.info("Outbound call %s answered", call.id)
        async for chunk in call.audio_stream():
            await call.send_audio(chunk)
    else:
        logger.error("Dial was not answered")

Authentication

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

API key (with your connector UUID):

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

mTLS (client certificate and key, 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 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. Required when constructing CallAudioConfig yourself. 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 (track 0). ISOLATED keeps each leg as a separate track, which you read with call.audio_stream(track_id=...) (caller 0, callee 1).
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 audio_stream() in real time.

from agentduet import CallAudioConfig

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

Architecture

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

1. Session manager connection. One persistent control connection to the server. It authenticates, keeps a heartbeat, reconnects automatically on drops, and delivers incoming call and message notifications. You hold a single SessionManager for the life of your process.
2. Session. An ephemeral, per-subscriber handle you open with sm.open_session(session_id, subscriber). It carries your calls (inbound_call / outbound_call, returning a Call) and outbound messages (send_message) for that subscriber.
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(), dial(), connect(), send_audio(), and so on) and closes it with the call; you never manage it directly, you just drive the Call.

Trigger Conditions

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

Configure trigger conditions when you want to change what the server routes to you, for example to receive only missed calls or to stop receiving inbound calls. Build the conditions with TriggerConditionsBuilder and send them once after connecting.

from agentduet import InboundCallMode, TriggerConditionsBuilder

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

await sm.setup_trigger_conditions(trigger_config)

InboundCallMode controls which inbound calls reach you:

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

Three things to keep in mind:

• Trigger conditions route notifications; they never gate what the SDK can do. Calls you place yourself with session.outbound_call() and dial() work regardless of this configuration. The builder's outbound_call / outbound_message toggles belong to an upcoming feature (notifying your agent when a subscriber places a call or sends a message, so it can step in before the call or message reaches its destination) and have no effect yet.
• The builder starts 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, given a call you obtained from session.inbound_call(noti) (inbound) or session.outbound_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
    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 subscriber (the party your agent represents) 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.

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

Hand the call to another node

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

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

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

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_terminated
    def on_terminated(evt):
        print("Call ended")

    # Server-side call errors
    @call.on_call_event(CallEvent.CALL_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_terminated is shorthand for on_call_event(CallEvent.CALL_TERMINATED); its payload is always None (the event carries no data). A CALL_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 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 @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.

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:

import uuid
from agentduet import IncomingCallNotification


@sm.on_incoming_call
async def on_call(noti: IncomingCallNotification):
    session = await sm.open_session(str(uuid.uuid4()), noti.subscriber)
    call = await session.inbound_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

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) after session.inbound_call(noti).

API Reference

SessionManager

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

Method	Description
@on_incoming_call	Decorator. Receives an IncomingCallNotification for every incoming call.
@on_incoming_message	Decorator. Receives an IncomingMessage for every incoming WhatsApp message.
await open_session(session_id, subscriber)	Open an ephemeral session for subscriber (get-or-create). session_id is any unique string you supply. Returns a Session.
await list_sessions(subscriber=None, offset=0, limit=50)	List active sessions for this connector (observability). Returns list[SessionInfo].
await setup_trigger_conditions(config)	Set which call and message flows the server routes to you. See Trigger Conditions.
start()	Begin delivering inbound notifications. Idempotent. run_forever() calls it for you; call it directly only if you drive your own loop.
await run_forever()	Start delivery, then run until interrupted or the context manager exits.

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

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

Session

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

Method	Description
await inbound_call(noti)	Attach to a notified incoming call; returns a ready-to-use Call.
await outbound_call(dest)	Create an outbound Call to dest (an Address), in CallState.NEW. Call dial() to ring.
await send_message(msg)	Send an outbound message (for example a SendWAMessage). Returns a SendMessageResult.

Properties: id, subscriber.

Address

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

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

IncomingCallNotification

Delivered to your @sm.on_incoming_call handler.

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

IncomingMessage

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

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

Call

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

Method	Description
await answer()	Answer an incoming call.
await dial(ring_time_seconds=60)	Ring an outbound call built by outbound_call(). Falsy with CALL_UNANSWERED / TIMEOUT if it does not answer.
await connect(ring_time_seconds=60)	Start a 3-way conference between caller, callee, and agent.
await whisper()	After connect(): agent is heard only by the subscriber.
await barge()	After connect(): agent is heard by both parties.
await spy()	After connect(): agent hears both parties, neither hears the agent.
await disconnect()	End the call for all parties.
await close()	Release the agent. After connect(), caller and callee stay connected; after answer() alone, the call ends for both.
await send_audio(audio_data)	Queue binary audio for sending.
await clear_send_audio_buffer()	Drop queued outgoing audio and stop playback (use on interruption).
await get_send_audio_buffer_size()	Current size of the outgoing buffer, in bytes.
audio_stream(track_id=0)	Async iterator of incoming audio chunks. Track 0 in MIXED; caller 0 / callee 1 in ISOLATED.
@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 (only in CallState.NEW).

Properties: id, participant (Address), subscriber, caller, callee, caller_track_id, callee_track_id, 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.

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.

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: 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.
  • CallStateError: operation attempted in an incompatible call state.
  • CallCommandError / CallCommandTimeoutError: a command was rejected or timed out (usually surfaced as a CommandResult).
• SessionError: base for session problems.
  • SessionAlreadyExistsError: a session with that id already exists on the server.
  • SessionNotFoundError: the session id does not exist on the server.
  • SubscriberMismatchError: the session subscriber does not match the call's.
  • ParticipantsFullError: the session already holds the maximum participants.
• CallNotFoundError: no pending call for the given id (expired or already taken).
• ChannelNotConfiguredError: the channel the operation needs (for example WhatsApp messaging) is not configured on the server.
• QuotaExceededError / OutboundOverflowError / ForbiddenError / InvalidRequestError: request-level rejections.
• BufferFullError: outgoing audio buffer is full.
• 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
QUOTA_EXCEEDED	Message sending limit exceeded for this connector.
CHANNEL_NOT_CONFIGURED	The session channel is not configured for messaging.
SESSION_BUSY	The session is bound to another live connection.
SESSION_NOT_FOUND	The session id does not exist on the server.
SESSION_CLOSED	The session has already closed.
SESSION_ALREADY_EXISTS	A session with that id already exists.
SUBSCRIBER_MISMATCH	The subscriber does not match the session.
PARTICIPANTS_FULL	The session already holds the maximum participants.
CALL_NOT_FOUND	No pending call for the given id.
INVALID_REQUEST	Request payload format or structure is invalid.
REMOTE_ERROR	The provider (for example, Meta/WhatsApp) rejected the message.
UNKNOWN	An unrecognized code (forward-compatible fallback).

Failure Modes and Delivery Semantics

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

• Redelivery and deduplication. Inbound delivery is at-least-once: a notification can be redelivered after a reconnect or a competing-consumer reclaim. Dedup inbound messages on IncomingMessage.id (a server-generated unique id) and calls on IncomingCallNotification.call_id.
• Ordering. Delivery is connector-wide competing-consumer: notifications may arrive concurrently and out of order. Correlate by (subscriber, participant) yourself — there is no per-session ordering guarantee.
• Session idle eviction. Server sessions have a 30-minute sliding TTL. If a session idle-evicts, the next inbound_call / outbound_call / send_message transparently re-opens it and retries once — you don't handle SESSION_NOT_FOUND yourself.
• Reconnects. The session-manager link auto-reconnects with jittered exponential backoff. Established calls are unaffected (the voice WebSocket is independent). If the voice transport drops mid-call, the call fires on_terminated and any in-flight command raises SessionError 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 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.
• 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.