> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.agentduet.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.agentduet.com/_mcp/server.

# Sessions and Handoff

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

```python
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:

```python
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

```python
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](/concepts/whats-app-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

```python
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.

```python
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`:

```python
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.

## Related

- [Architecture](/concepts/architecture)
- [WhatsApp Messaging](/concepts/whats-app-messaging)
- [Call States](/concepts/call-states-and-lifecycle)