WhatsApp

pip install fastapi uvicorn httpx anthropic geniffy

uv add fastapi uvicorn httpx anthropic geniffy

import hashlib
import hmac
import json
import os

import httpx
from anthropic import AsyncAnthropic
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from fastapi.responses import PlainTextResponse
from geniffy import AsyncGeniffy

geniffy = AsyncGeniffy()                          # reads GENIFFY_API_KEY
claude = AsyncAnthropic()                         # reads ANTHROPIC_API_KEY
graph = httpx.AsyncClient(base_url="https://graph.facebook.com/v25.0",
                          headers={"Authorization": f"Bearer {os.environ['WHATSAPP_TOKEN']}"})
app = FastAPI()
seen: dict[str, None] = {}                        # recent message ids: Meta can send one message more than once
recent: dict[str, list] = {}                      # each sender's last few turns, for "yes" and "that one"


@app.get("/whatsapp")
async def verify(request: Request):
    """Meta checks the address once, when you add it."""
    q = request.query_params
    if q.get("hub.mode") == "subscribe" and hmac.compare_digest(q.get("hub.verify_token", ""),
                                                               os.environ["WHATSAPP_VERIFY_TOKEN"]):
        return PlainTextResponse(q.get("hub.challenge", ""))
    raise HTTPException(403)


@app.post("/whatsapp")
async def receive(request: Request, background: BackgroundTasks):
    body = await request.body()
    signed = "sha256=" + hmac.new(os.environ["WHATSAPP_APP_SECRET"].encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(request.headers.get("X-Hub-Signature-256", ""), signed):
        raise HTTPException(401)
    for entry in json.loads(body).get("entry", []):
        for change in entry.get("changes", []):
            value = change.get("value", {})
            for message in value.get("messages", []):  # delivery and read receipts come as "statuses"
                if message.get("type") != "text" or message["id"] in seen:
                    continue
                seen[message["id"]] = None
                if len(seen) > 10_000:
                    seen.pop(next(iter(seen)))
                background.add_task(answer, value["metadata"]["phone_number_id"], message["from"],
                                    message["text"]["body"])
    return {}                                     # at once: Meta sends a message again when it has to wait


async def answer(phone_number_id: str, sender: str, text: str) -> None:
    mem = geniffy.space(f"user_{sender}")         # the sender's WhatsApp number, as digits
    turn = {"role": "user", "content": text}
    context = await mem.context(text)             # what is known that bears on the message
    past = recent.setdefault(sender, [])
    response = await claude.messages.create(
        model="claude-opus-5-5",
        max_tokens=1024,
        system=f"You are a helpful assistant on WhatsApp. Keep replies short.\n\n<memory>\n{context}\n</memory>",
        messages=[*past[-10:], turn],
    )
    reply = "".join(block.text for block in response.content if block.type == "text")
    sent = await graph.post(f"/{phone_number_id}/messages", json={
        "messaging_product": "whatsapp", "to": sender, "type": "text", "text": {"body": reply}})
    sent.raise_for_status()
    past += [turn, {"role": "assistant", "content": reply}]
    await mem.memories.add(messages=[turn, {"role": "assistant", "content": reply}])