Telephony with Twilio

import asyncio
import json

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

from docsassistant.core import ask   # async iterator of text tokens, markdown-free for voice

app = FastAPI()


@app.websocket("/relay")
async def relay(ws: WebSocket):
    await ws.accept()
    call_sid, caller = "unknown", "anonymous"
    answer: asyncio.Task | None = None

    async def speak(text: str):
        try:
            async for token in ask(call_sid, caller, text):
                await ws.send_text(json.dumps({"type": "text", "token": token, "last": False}))
            await ws.send_text(json.dumps({"type": "text", "token": "", "last": True}))
        except asyncio.CancelledError:
            raise   # caller interrupted: stop generating, send nothing more

    try:
        while True:
            message = json.loads(await ws.receive_text())
            kind = message["type"]
            if kind == "setup":
                call_sid, caller = message["callSid"], message["from"]
            elif kind == "prompt" and message.get("last", True):
                if answer and not answer.done():
                    answer.cancel()
                answer = asyncio.create_task(speak(message["voicePrompt"]))
            elif kind == "interrupt":
                if answer and not answer.done():
                    answer.cancel()
                # message["utteranceUntilInterrupt"] is what the caller actually heard; store it in memory
            elif kind == "dtmf":
                if message["digit"] == "0":
                    if answer and not answer.done():    # stop the answer before ending, or tokens follow "end"
                        answer.cancel()
                        await asyncio.gather(answer, return_exceptions=True)
                    await ws.send_text(json.dumps({"type": "end", "handoffData": json.dumps({"reason": "agent_requested"})}))
            elif kind == "error":
                print("relay error:", message["description"])
    except WebSocketDisconnect:
        if answer:
            answer.cancel()

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.memory.ChatMemory;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import reactor.core.Disposable;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;

public class RelayHandler extends TextWebSocketHandler {

    private final ChatClient docsAssistant;
    private final ObjectMapper json = new ObjectMapper();
    private final Map<String, Disposable> running = new ConcurrentHashMap<>();

    public RelayHandler(ChatClient docsAssistant) {
        this.docsAssistant = docsAssistant;
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws IOException {
        JsonNode in = this.json.readTree(message.getPayload());
        switch (in.path("type").asText()) {
            case "setup" -> {
                session.getAttributes().put("callSid", in.path("callSid").asText());
                session.getAttributes().put("from", in.path("from").asText());
            }
            case "prompt" -> {
                if (in.path("last").asBoolean(true)) {
                    cancel(session);
                    String callSid = (String) session.getAttributes().get("callSid");
                    Disposable d = this.docsAssistant.prompt()
                            .user(in.path("voicePrompt").asText())
                            .advisors(a -> a.param(ChatMemory.CONVERSATION_ID, callSid))
                            .stream().content()
                            .doOnNext(t -> send(session, Map.of("type", "text", "token", t, "last", false)))
                            .doOnComplete(() -> send(session, Map.of("type", "text", "token", "", "last", true)))
                            .subscribe();
                    this.running.put(session.getId(), d);
                }
            }
            case "interrupt" -> cancel(session);
            case "dtmf" -> {
                if ("0".equals(in.path("digit").asText())) {
                    cancel(session);   // stop the running answer first, or its tokens would follow "end"
                    send(session, Map.of("type", "end", "handoffData", "{\"reason\":\"agent_requested\"}"));
                }
            }
            default -> { }
        }
    }

    private void cancel(WebSocketSession session) {
        Disposable d = this.running.remove(session.getId());
        if (d != null) {
            d.dispose();
        }
    }

    private void send(WebSocketSession session, Map<String, Object> body) {
        try {
            synchronized (session) {
                session.sendMessage(new TextMessage(this.json.writeValueAsString(body)));
            }
        } catch (IOException e) {
            cancel(session);
        }
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        cancel(session);
    }
}

from fastapi import FastAPI, WebSocket
from pipecat.runner.utils import parse_telephony_websocket
from pipecat.serializers.twilio import TwilioFrameSerializer
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams, FastAPIWebsocketTransport

app = FastAPI()


@app.websocket("/media")
async def media(ws: WebSocket):
    await ws.accept()
    transport_type, call_data = await parse_telephony_websocket(ws)   # reads connected and start
    serializer = TwilioFrameSerializer(
        stream_sid=call_data["stream_id"],
        call_sid=call_data["call_id"],
        account_sid="<Twilio account SID>",
        auth_token="<Twilio auth token>",   # lets the serializer end the call through the REST API
    )
    transport = FastAPIWebsocketTransport(
        websocket=ws,
        params=FastAPIWebsocketParams(audio_in_enabled=True, audio_out_enabled=True, serializer=serializer),
    )
    # build the pipeline: transport.input() -> stt -> context aggregator -> llm (DocsAssistant) -> tts
    #                      -> transport.output(), then run it with a PipelineTask