Move from LiveKit or Pipecat

curl -X POST https://api.60db.ai/agents/phone-lines \
  -H "Authorization: Bearer $SIXTYDB_API_KEY" -H "Content-Type: application/json" \
  -d '{
    "kind": "byo", "name": "main", "domain": "sip.carrier.example", "caller_id": "+15550100",
    "inbound_number": "+15550301", "agent_id": "AGENT_ID",
    "allowed_addresses": ["203.0.113.7"]
  }'

{
  "name": "Main inbound",
  "numbers": ["+15550301", "+15550302"],
  "allowed_addresses": ["203.0.113.7", "198.51.100.0/24"],
  "allowed_numbers": ["+919876543210"],
  "krisp_enabled": true
}

{
  "name": "Main dispatch",
  "trunk_ids": ["ST_xxxx"],
  "rule": { "dispatchRuleIndividual": { "roomPrefix": "call-" } },
  "roomConfig": { "agents": [{ "agentName": "front-desk" }] }
}

{
  "kind": "byo",
  "name": "Main inbound",
  "domain": "sip.carrier.example",
  "caller_id": "+15550301",
  "inbound_number": "+15550301",
  "numbers": [{ "number": "+15550302", "agent_id": null }],
  "agent_id": "9c2e1d7a-0b3f-4e6a-8f1d-5a7b9c0e2d44",
  "allowed_addresses": ["203.0.113.7", "198.51.100.0/24"],
  "allowed_numbers": ["+919876543210"]
}

{
  "name": "Main outbound",
  "address": "sip.carrier.example",
  "transport": "SIP_TRANSPORT_AUTO",
  "numbers": ["+15550100"],
  "auth_username": "office-trunk",
  "auth_password": "…"
}

{
  "kind": "byo",
  "name": "Main outbound",
  "domain": "sip.carrier.example",
  "username": "office-trunk",
  "password": "…",
  "caller_id": "+15550100",
  "inbound_number": null
}

{"event":"start","start":{"mediaFormat":{"encoding":"audio/x-mulaw","sampleRate":8000,"channels":1}}}

"""Stream-call adapter: Pipecat-style media events <-> a 60db stream call.

pip install websockets requests
export SIXTYDB_API_KEY=... AGENT_ID=...
python stream_adapter.py
"""
import asyncio
import base64
import json
import os

import requests
import websockets

API = "https://api.60db.ai"
FRAME_SECONDS = 0.02  # 20 ms
MULAW_SILENCE = b"\xff" * 160  # 160 bytes of mu-law silence = 20 ms at 8 kHz

# Must be the first message on the socket, before any audio.
START_MESSAGE = {
    "event": "start",
    "start": {"mediaFormat": {"encoding": "audio/x-mulaw", "sampleRate": 8000, "channels": 1}},
}


def start_stream_call(agent_id: str) -> str:
    """Server side only: this call needs your API key. Hand only the returned URL onward."""
    r = requests.post(
        f"{API}/agents/{agent_id}/stream-calls",
        headers={"Authorization": f"Bearer {os.environ['SIXTYDB_API_KEY']}"},
        json={},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()["data"]["url"]  # single use, valid for 55 seconds


class StreamCallAdapter:
    """Opens the wss URL, sends the startup message, then forwards audio both ways.

    In a Pipecat pipeline, call `send_media(event)` with each Twilio-serializer media
    message (a dict) from your transport's output, and pass `on_audio` / `on_clear`
    callbacks that push agent audio and barge-in back into the pipeline.
    """

    def __init__(self, url, on_audio, on_clear, on_event=None):
        self.url = url
        self.on_audio = on_audio  # (mulaw_bytes) -> None
        self.on_clear = on_clear  # () -> None
        self.on_event = on_event or (lambda message: None)
        self._ws = None

    async def send_media(self, event: dict) -> None:
        """Forward one base64 mu-law `media` event from the serializer."""
        if event.get("event") != "media" or self._ws is None:
            return
        payload = event["media"]["payload"]
        await self._ws.send(json.dumps({"event": "media", "media": {"payload": payload}}))

    async def stop(self) -> None:
        """The caller hung up."""
        if self._ws is not None:
            await self._ws.send(json.dumps({"event": "stop"}))

    async def run(self, ready: asyncio.Event) -> dict | None:
        """Connect, handshake, then read replies until the call ends. Returns the summary."""
        async with websockets.connect(self.url) as ws:
            self._ws = ws
            await ws.send(json.dumps(START_MESSAGE))  # before forwarding any audio
            ready.set()
            summary = None
            async for raw in ws:
                if isinstance(raw, bytes):
                    continue  # this style carries audio as JSON media events
                message = json.loads(raw)
                kind = message.get("event")
                if kind == "media":
                    self.on_audio(base64.b64decode(message["media"]["payload"]))
                elif kind == "clear":
                    self.on_clear()
                elif kind == "ended":
                    summary = message.get("summary")
                    break
                else:
                    self.on_event(message)  # dtmf, transfer, hangup
            self._ws = None
            return summary


async def demo(agent_id: str, seconds: int = 8) -> None:
    played = 0

    def on_audio(chunk: bytes) -> None:
        nonlocal played
        played += len(chunk)

    adapter = StreamCallAdapter(
        start_stream_call(agent_id),
        on_audio=on_audio,
        on_clear=lambda: print("clear: drop buffered agent audio"),
        on_event=lambda m: print("event:", m.get("event")),
    )
    ready = asyncio.Event()
    runner = asyncio.create_task(adapter.run(ready))
    await ready.wait()

    silence = base64.b64encode(MULAW_SILENCE).decode()
    for _ in range(int(seconds / FRAME_SECONDS)):  # real-time pacing, a frame every 20 ms
        await adapter.send_media({"event": "media", "media": {"payload": silence}})
        await asyncio.sleep(FRAME_SECONDS)
    await adapter.stop()

    summary = await asyncio.wait_for(runner, timeout=30)
    print(f"agent audio received: {played} bytes; summary keys: {sorted(summary or {})}")


if __name__ == "__main__":
    asyncio.run(demo(os.environ["AGENT_ID"]))