Twilio

pip install fastapi uvicorn python-multipart anthropic twilio geniffy

uv add fastapi uvicorn python-multipart anthropic twilio geniffy

import os

from anthropic import AsyncAnthropic
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request, Response
from geniffy import AsyncGeniffy
from twilio.request_validator import RequestValidator
from twilio.twiml.messaging_response import MessagingResponse

geniffy = AsyncGeniffy()                          # reads GENIFFY_API_KEY
claude = AsyncAnthropic()                         # reads ANTHROPIC_API_KEY
twilio = RequestValidator(os.environ["TWILIO_AUTH_TOKEN"])
app = FastAPI()
recent: dict[str, list] = {}                      # each person's last few turns, for "yes" and "that one"


def space_for(number: str) -> str:
    """The sender, by the digits of their number: the same person whether they text or use WhatsApp."""
    digits = "".join(ch for ch in number if ch.isdigit())
    if not digits:
        raise HTTPException(400, "This message has no sender number.")
    return f"user_{digits}"


@app.post("/twilio")
async def receive(request: Request, background: BackgroundTasks):
    form = {k: str(v) for k, v in (await request.form()).items()}
    if not twilio.validate(os.environ["TWILIO_WEBHOOK_URL"], form, request.headers.get("X-Twilio-Signature", "")):
        raise HTTPException(403)
    text, reply = (form.get("Body") or "").strip(), MessagingResponse()
    if text:
        mem = geniffy.space(space_for(form.get("From", "")))
        turn = {"role": "user", "content": text}
        past = recent.setdefault(form["From"], [])
        response = await claude.messages.create(
            model="claude-opus-5-5",
            max_tokens=600,
            system=f"You are a helpful assistant over text messages. Keep replies short.\n\n"
                   f"<memory>\n{await mem.context(text)}\n</memory>",
            messages=[*past[-10:], turn],
        )
        said = "".join(block.text for block in response.content if block.type == "text")
        reply.message(said)
        past += [turn, {"role": "assistant", "content": said}]
        background.add_task(mem.memories.add, messages=[turn, {"role": "assistant", "content": said}])
    return Response(str(reply), media_type="application/xml")