Sessions and Handoff | AgentDuet SDK

A Session is a short-lived, per-subscriber handle. Reuse a session_id to continue a conversation; allocate a new id to start a new one. Cross-node handoff moves a call to another process while the call is still CallState.NEW, before the voice connection opens.

Opening a session

session = await sm.open_session(session_id, subscriber)

Argument	Meaning
`session_id`	Any unique string you supply. Reuse to continue; new id to start fresh.
`subscriber`	From the notification (`noti.subscriber`), or your outbound calling identity.

Behavior is get-or-create. Server sessions have a ~30-minute sliding idle TTL; the next process_call / make_call / send_message transparently re-opens if needed.

Rules:

Never share one session id across different subscribers.
Sessions have a small participant cap - do not pile unrelated customers into one id.
Dedup inbound calls with IncomingCallNotification.call_id and inbound WhatsApp with IncomingMessage.id (at-least-once delivery).

Observability:

sessions = await sm.list_sessions(subscriber=None, offset=0, limit=50)
# list[SessionInfo] - session_id, subscriber, participants, created_at, last_activity_at

Attach or create a call

call = await session.process_call(noti)           # inbound → ready Call
call = await session.make_call(Address.telco(n))  # outbound, still NEW
result = await session.send_message(SendWAMessage(...))  # WhatsApp

process_call attaches media credentials (URL + JWT). Until you answer / dial / connect / send_audio, state stays NEW. WhatsApp send does not open a voice connection. See WhatsApp Messaging.

Tutorial: cross-node handoff

Move media processing to another server before the voice connection opens. Node A receives the call, serializes with to_json() while still CallState.NEW, and publishes the payload. Node B reconstructs with Call.from_json() and answers - no SessionManager required on Node B.

This demo uses a file as the queue (/tmp/pending_call.json). Swap that for Redis, SQS, or your bus.

Node A - serialize while NEW

import asyncio
import logging
import os
import pathlib
from agentduet import (
    CallState,
    CallStateError,
    IncomingCallNotification,
    SessionManager,
    SessionManagerConfig,
    new_session_id,
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
HANDOFF_FILE = pathlib.Path("/tmp/pending_call.json")
async def node_a() -> None:
    config = SessionManagerConfig.create(
        api_key=os.getenv("AGENTDUET_API_KEY"),
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
    )
    async with SessionManager(config) as sm:
        logger.info("[Node A] Waiting for incoming call...")
        @sm.on_incoming_call
        async def on_call(noti: IncomingCallNotification):
            session = await sm.open_session(new_session_id(), noti.subscriber)
            call = await session.process_call(noti)
            logger.info("[Node A] Got call %s (state=%s)", call.id, call.state)
            if call.state != CallState.NEW:
                logger.error("[Node A] Expected NEW; got %s", call.state)
                return
            try:
                handoff_json = call.to_json()
            except CallStateError:
                logger.error("[Node A] to_json() failed - media already open?")
                return
            HANDOFF_FILE.write_text(handoff_json)
            logger.info("[Node A] Handoff written to %s", HANDOFF_FILE)
        await sm.run_forever()
if __name__ == "__main__":
    asyncio.run(node_a())

Node B - restore and answer

Run Node A first (leave it running). Then start Node B after the handoff file exists.

import asyncio
import logging
import pathlib
from agentduet import Call, CallClosedError, CallState
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
HANDOFF_FILE = pathlib.Path("/tmp/pending_call.json")
async def node_b() -> None:
    if not HANDOFF_FILE.exists():
        logger.error(
            "[Node B] No handoff file at %s - run Node A first.", HANDOFF_FILE
        )
        return
    handoff_json = HANDOFF_FILE.read_text()
    call = Call.from_json(handoff_json)
    logger.info("[Node B] Restored call %s (state=%s)", call.id, call.state)
    if call.state != CallState.NEW:
        logger.error("[Node B] Expected NEW; got %s", call.state)
        return
    @call.on_hangup
    def on_hangup(evt):
        logger.info("[Node B] Call %s hung up", call.id)
    # Voice WebSocket opens lazily here on Node B.
    result = await call.answer()
    if not result:
        logger.error(
            "[Node B] answer failed: %s (%s)",
            result.error_message,
            result.error_code,
        )
        return
    try:
        async for chunk in call.caller.audio_stream():
            await call.send_audio(chunk)
    except CallClosedError:
        logger.info("[Node B] Call %s closed", call.id)
if __name__ == "__main__":
    asyncio.run(node_b())

Single-process switcher (optional)

For local testing you can put both in one file and select with NODE=A / NODE=B:

import os
if __name__ == "__main__":
    node = os.getenv("NODE", "A").upper()
    if node == "A":
        asyncio.run(node_a())
    elif node == "B":
        asyncio.run(node_b())
    else:
        raise SystemExit("Set NODE=A or NODE=B")

Constraints:

to_json() after media opens raises CallStateError.
Node B still needs network reachability to AgentDuet media endpoints carried in the payload.
Do not answer on both nodes.

Next Step

WhatsApp Messaging

Handle inbound messages, parse raw webhook payloads, and send WhatsApp responses in the same subscriber session.