Memory for agents on Twilio.

pip install 'niadra[twilio]'   # no framework dependency

"""A Twilio voice webhook that verifies the carrier's attestation before the first context."""

import os

from flask import Flask, request

from niadra import Niadra
from niadra.integrations.twilio import parse_call

niadra = Niadra(channel="voice")
app = Flask(__name__)


@app.post("/twilio/voice")
def incoming_call() -> tuple[str, int]:
    call = parse_call(request.get_data(), request.headers, request.url, os.environ["TWILIO_AUTH_TOKEN"])
    if call is None:
        return "", 403
    conversation = call.conversation(niadra)
    call.verify(conversation)  # StirVerstat: A proves V2, B and C prove V1
    context = conversation.ready()  # the first read, started above; the phone rings meanwhile
    return connect_your_voice_agent(call.call_sid, context.system_block), 200


@app.post("/twilio/status")
def status() -> tuple[str, int]:
    call = parse_call(request.get_data(), request.headers, request.url, os.environ["TWILIO_AUTH_TOKEN"])
    if call is not None:
        call.ended(call.conversation(niadra))
    return "", 204

npm install @niadra/sdk   # @niadra/sdk/twilio

// A Twilio Programmable Voice webhook on Hono: the carrier's attestation proves the caller, the
// context is read before the first answer, and each recognized sentence is recorded.
import { Hono } from "hono";
import { Niadra } from "@niadra/sdk";
import { readTwilio, recordTwilioInbound, verifyTwilio } from "@niadra/sdk/twilio";

const niadra = new Niadra();
const authToken = process.env.TWILIO_AUTH_TOKEN ?? "";
const publicUrl = process.env.PUBLIC_URL ?? "";

export const app = new Hono();

app.post("/twilio/voice", async (c) => {
  const { status, request } = await readTwilio(`${publicUrl}/twilio/voice`, await c.req.text(), c.req.raw.headers, { authToken });
  if (!request) return c.body(null, status as 403);
  // The attestation is recorded once, on the call's first webhook; later ones open at the proven level.
  const first = request.params.CallStatus === "ringing";
  const convo = niadra.conversation({
    subject: request.subject,
    channel: "voice",
    conversation_id: request.conversationId ?? "",
    verification: first ? "V0" : (request.proof?.level ?? "V0"),
  });
  if (first) await verifyTwilio(convo, request);
  recordTwilioInbound(convo, request);
  const ctx = await convo.ready(); // the first read, started on the ringing webhook
  convo.markInjected(ctx);
  const reply = await answer(ctx.text, request.text);
  convo.agent(reply);
  const twiml = `<Response><Gather input="speech" action="/twilio/voice"><Say>${escape(reply)}</Say></Gather></Response>`;
  return c.body(twiml, 200, { "content-type": "text/xml" });
});

export default app;