Metadata-Version: 2.4
Name: agentduet
Version: 1.0.0b10
Summary: Build AI agents for phone and WhatsApp calls with real-time audio and messaging
Keywords: websocket,sdk,voip,telephony,whatsapp,ai,voice-agent,real-time-audio
Author: AgentDuet
License-Expression: LicenseRef-Proprietary
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Dist: websockets>=15.0,<16
Requires-Dist: httpx>=0.27,<1
Requires-Dist: abxbus>=2.4,<3
Requires-Dist: agentduet-adapters[all]>=0.1.0b1 ; extra == 'adapters'
Requires-Python: >=3.12
Project-URL: Homepage, https://agentduet.com
Provides-Extra: adapters
Description-Content-Type: text/markdown

# 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.
- **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, 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

```bash
pip install --pre agentduet
```

The package is currently published as `1.0.0bN` pre-releases, so `--pre` is required until
`1.0.0` final ships. 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:

```bash
pip install --pre "agentduet[adapters]"
```

That brings in [`agentduet-adapters`](https://github.com/AgentDuet/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]"`. See [VoiceAgent](#voiceagent-the-same-thing-in-three-lines).

## Quick Start

Get your API key and connector UUID at [agentduet.com](https://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 only: a `subscriber` (the principal the event belongs to) and the external `participant`. To act on one, you open a short-lived **`Session`** for that subscriber with `sm.open_session(session_id, subscriber)`:

- 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; the server infers the recipient from the payload.

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.

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

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

### 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()`.

```python
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.participant, msg.payload)

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

```python
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 two modes when you build the config. Both go through `SessionManagerConfig.create()`.

**API key** (with your connector UUID):

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

**mTLS** (client certificate and key):

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

```python
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 (`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`**, 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` 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`.

### 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` (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 `Call`s 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 or to stop receiving inbound calls. Build the conditions with `TriggerConditionsBuilder` (its `build()` returns the `TriggerConditions` object that `setup_trigger_conditions` takes) and send them once after connecting.

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

## 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.

```python
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. After `connect()`, switch the agent's audio between the three ambient modes from the Participant Model: `whisper()` (subscriber only), `barge()` (everyone), `spy()` (listen only).

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

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

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

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

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

```python
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`](https://github.com/AgentDuet/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:

```bash
pip install --pre "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:

```python
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=...)`, 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](https://ai.google.dev/gemini-api/docs/live) and plays the model's audio back, clearing the buffer on interruption.

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

```python
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)
    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](https://pypi.org/project/google-adk/) `Runner` and `LiveRequestQueue`.

```python
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_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(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, 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 process_call(noti)` | Attach to a notified call; returns a ready-to-use `Call`. |
| `await make_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 (a `BaseSendMessage` subclass such as `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 `process_call`). |
| `subscriber` | The principal the call runs on behalf of; pass it to `open_session()`. |
| `participant` | The external party, as an `Address`. |
| `network` | Shorthand for `participant.network`. |
| `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). |
| `network` | Shorthand for `participant.network`. |
| `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 `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. |
| `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. |
| `@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`, `caller` (`CallParty`), `callee` (`CallParty`), `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. |

### 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 `TranscriptDelta`s, `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` and `AGENTDUET_CONNECTOR_UUID`. 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 `ModelEvent`s (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, token, or mTLS).
- `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.
- `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.

```python
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 `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 jittered exponential
  backoff. 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.

```python
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](mailto:team@agentduet.com). To get an API key and connector UUID, visit [agentduet.com](https://agentduet.com).
