Sessions and Handoff

1
session = await sm.open_session(session_id, subscriber)

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

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

1
import asyncio
2
import logging
3
import os
4
import pathlib
5
6
from agentduet import (
7
    CallState,
8
    CallStateError,
9
    IncomingCallNotification,
10
    SessionManager,
11
    SessionManagerConfig,
12
    new_session_id,
13
)
14
15
logging.basicConfig(level=logging.INFO)
16
logger = logging.getLogger(__name__)
17
18
HANDOFF_FILE = pathlib.Path("/tmp/pending_call.json")
19
20
21
async def node_a() -> None:
22
    config = SessionManagerConfig.create(
23
        api_key=os.getenv("AGENTDUET_API_KEY"),
24
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
25
    )
26
27
    async with SessionManager(config) as sm:
28
        logger.info("[Node A] Waiting for incoming call...")
29
30
        @sm.on_incoming_call
31
        async def on_call(noti: IncomingCallNotification):
32
            session = await sm.open_session(new_session_id(), noti.subscriber)
33
            call = await session.process_call(noti)
34
            logger.info("[Node A] Got call %s (state=%s)", call.id, call.state)
35
36
            if call.state != CallState.NEW:
37
                logger.error("[Node A] Expected NEW; got %s", call.state)
38
                return
39
40
            try:
41
                handoff_json = call.to_json()
42
            except CallStateError:
43
                logger.error("[Node A] to_json() failed - media already open?")
44
                return
45
46
            HANDOFF_FILE.write_text(handoff_json)
47
            logger.info("[Node A] Handoff written to %s", HANDOFF_FILE)
48
49
        await sm.run_forever()
50
51
52
if __name__ == "__main__":
53
    asyncio.run(node_a())

1
import asyncio
2
import logging
3
import pathlib
4
5
from agentduet import Call, CallClosedError, CallState
6
7
logging.basicConfig(level=logging.INFO)
8
logger = logging.getLogger(__name__)
9
10
HANDOFF_FILE = pathlib.Path("/tmp/pending_call.json")
11
12
13
async def node_b() -> None:
14
    if not HANDOFF_FILE.exists():
15
        logger.error(
16
            "[Node B] No handoff file at %s - run Node A first.", HANDOFF_FILE
17
        )
18
        return
19
20
    handoff_json = HANDOFF_FILE.read_text()
21
    call = Call.from_json(handoff_json)
22
    logger.info("[Node B] Restored call %s (state=%s)", call.id, call.state)
23
24
    if call.state != CallState.NEW:
25
        logger.error("[Node B] Expected NEW; got %s", call.state)
26
        return
27
28
    @call.on_hangup
29
    def on_hangup(evt):
30
        logger.info("[Node B] Call %s hung up", call.id)
31
32
    # Voice WebSocket opens lazily here on Node B.
33
    result = await call.answer()
34
    if not result:
35
        logger.error(
36
            "[Node B] answer failed: %s (%s)",
37
            result.error_message,
38
            result.error_code,
39
        )
40
        return
41
42
    try:
43
        async for chunk in call.caller.audio_stream():
44
            await call.send_audio(chunk)
45
    except CallClosedError:
46
        logger.info("[Node B] Call %s closed", call.id)
47
48
49
if __name__ == "__main__":
50
    asyncio.run(node_b())

1
import os
2
3
if __name__ == "__main__":
4
    node = os.getenv("NODE", "A").upper()
5
    if node == "A":
6
        asyncio.run(node_a())
7
    elif node == "B":
8
        asyncio.run(node_b())
9
    else:
10
        raise SystemExit("Set NODE=A or NODE=B")