A WhatsApp bot in 15 minutes

# bot.py: run with  uvicorn bot:app --port 8000
import os
import httpx
from fastapi import FastAPI, Request

TEXA = httpx.Client(base_url="https://api.texadesk.com/v1", timeout=90,
                    headers={"Authorization": f"Bearer {os.environ['TEXA_API_KEY']}"})
META = httpx.Client(base_url="https://graph.facebook.com/v21.0", timeout=30,
                    headers={"Authorization": f"Bearer {os.environ['META_TOKEN']}"})
PHONE_ID = os.environ["META_PHONE_NUMBER_ID"]
conversations: dict[str, str] = {}      # customer's number -> Texa conversation (use your DB)
app = FastAPI()


def send_whatsapp(to: str, text: str) -> None:
    META.post(f"/{PHONE_ID}/messages", json={
        "messaging_product": "whatsapp", "to": to, "type": "text", "text": {"body": text}})


def conversation_for(number: str, name: str | None) -> str:
    if number not in conversations:
        r = TEXA.post("/conversations", headers={"Idempotency-Key": f"wa-open-{number}"},
                      json={"channel": "whatsapp", "customer_ref": f"wa:{number}",
                            "first_name": name})
        conversations[number] = r.json()["conversation_id"]
    return conversations[number]


@app.post("/whatsapp")                  # Meta's webhook (verify its signature too)
async def incoming(request: Request):
    body = await request.json()
    for entry in body.get("entry", []):
        for change in entry.get("changes", []):
            value = change.get("value", {})
            names = {c["wa_id"]: c["profile"]["name"] for c in value.get("contacts", [])}
            for m in value.get("messages", []):
                if m.get("type") != "text":
                    continue
                number = m["from"]
                cid = conversation_for(number, names.get(number))
                r = TEXA.post(f"/conversations/{cid}/messages",
                              json={"message_id": m["id"], "text": m["text"]["body"]})
                if r.status_code == 409 and r.json()["error"]["code"] == "conversation_closed":
                    conversations.pop(number)          # it ended: start a new one
                    cid = conversation_for(number, names.get(number))
                    r = TEXA.post(f"/conversations/{cid}/messages",
                                  json={"message_id": m["id"], "text": m["text"]["body"]})
                for event in r.json().get("events", []):
                    if event["type"] == "reply":
                        send_whatsapp(number, event["text"])
    return {"ok": True}