WhatsApp API with Python: Flask and FastAPI Tutorial
Python is the obvious glue for WhatsApp: a cron job that sends reminders, a Django or FastAPI backend that answers customers, a notebook that pushes alerts. The hard part is not the HTTP call. It is the webhook, where a missing raw body, a JSON-encoded challenge or a slow handler quietly breaks everything.
This WhatsApp API Python tutorial builds a small, correct integration with requests for sending and both Flask and FastAPI for receiving. You will verify Meta's signature, track the 24-hour window yourself so you know when a template is needed, and reply without making Meta wait.
The last section shows the same flow on Tenashi's API, including a retry loop that cannot double-send.
Key takeaways
- Reuse one
requests.Sessionwith the bearer token and always pass atimeout. - Hash
request.get_data()in Flask orawait request.body()in FastAPI, never the parsed JSON. - FastAPI needs
Query(alias="hub.mode")for Meta's dotted parameter names and aPlainTextResponsefor the challenge. - Record when each user last wrote to you. That timestamp decides between free-form text and a template.
Set up: credentials, packages and a session
You need a system user access token, the phone number ID of the sending number (not the WhatsApp Business Account ID), your app secret and a verify token you invent. If you only have the dashboard's temporary token, swap it now; our guide to a permanent WhatsApp API access token shows how.
python3 -m venv .venv && source .venv/bin/activate
pip install requests flask # or: pip install requests fastapi uvicorn
export WHATSAPP_TOKEN="$WHATSAPP_TOKEN"
export PHONE_NUMBER_ID="PHONE_NUMBER_ID"
export APP_SECRET="your-app-secret"
export VERIFY_TOKEN="any-string-you-choose"
The code targets Python 3.9 and later. It uses Graph API v23.0; use the current version from Meta's changelog when you build.
Send text and template messages with requests
Wrap the API in one module. A shared session reuses the TLS connection, which matters when a job sends a few hundred messages, and a custom exception keeps Meta's numeric error code available to the caller.
# wa.py: send messages through the Cloud API with requests
import os
from datetime import datetime, timedelta, timezone
import requests
GRAPH = "https://graph.facebook.com/v23.0"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['WHATSAPP_TOKEN']}"
class WhatsAppError(Exception):
def __init__(self, code, message):
super().__init__(f"{code}: {message}")
self.code = code
def _post(payload: dict) -> dict:
r = session.post(
f"{GRAPH}/{os.environ['PHONE_NUMBER_ID']}/messages",
json={"messaging_product": "whatsapp", **payload},
timeout=10,
)
data = r.json()
if not r.ok:
err = data.get("error", {})
raise WhatsAppError(err.get("code"), err.get("message", r.text))
return data
def recipient(msg: dict) -> dict:
"""Reply address for an inbound message: phone if present, else the BSUID."""
return {"to": msg["from"]} if msg.get("from") else {"recipient": msg["from_user_id"]}
def send_text(who: dict, body: str) -> str:
data = _post({"recipient_type": "individual", **who, "type": "text", "text": {"body": body}})
return data["messages"][0]["id"] # wamid
def send_template(who: dict, name: str, language: str, params: list) -> str:
data = _post({
"recipient_type": "individual",
**who,
"type": "template",
"template": {
"name": name,
"language": {"code": language},
"components": [
{"type": "body", "parameters": [{"type": "text", "text": p} for p in params]}
],
},
})
return data["messages"][0]["id"]
def mark_read(wamid: str) -> None:
_post({"status": "read", "message_id": wamid})
def window_open(last_inbound_at: datetime) -> bool:
"""True while the 24-hour customer service window is open."""
return datetime.now(timezone.utc) - last_inbound_at < timedelta(hours=24)
The who dictionary is either {"to": "+919811122233"} or, from July 2026, {"recipient": "US.1349..."}. Meta now identifies users with business-scoped user IDs, and a user with a username may reach you without a phone number in the webhook. recipient() picks whichever is present.
window_open() exists because the 24-hour rule is the decision every sender has to make. Free-form text only works within 24 hours of the user's last message; outside that, only an approved template goes through, and the failure (error 131047) often arrives later in a status webhook rather than in the send response. Deciding up front is cleaner than waiting for a failure:
if window_open(customer.last_inbound_at):
send_text({"to": customer.phone}, f"Your order {order_id} is ready for pickup.")
else:
send_template({"to": customer.phone}, "order_ready", "en_US", [customer.first_name, order_id])
The template has to exist and be approved in that exact language code, or Meta returns error 132001.
Receive and verify webhooks in Flask
Meta sends a GET once to verify your URL, then signed POSTs for every message and status. The signature in X-Hub-Signature-256 is an HMAC-SHA256 of the raw body, keyed with your app secret.
# app_flask.py
import hashlib
import hmac
import os
import threading
from datetime import datetime, timezone
from flask import Flask, abort, request
from wa import mark_read, recipient, send_text
app = Flask(__name__)
APP_SECRET = os.environ["APP_SECRET"].encode()
seen = set() # message ids already handled; use Redis or a unique column in production
last_inbound = {} # user id -> datetime of their last message; persist this too
@app.get("/webhook")
def verify():
if (
request.args.get("hub.mode") == "subscribe"
and request.args.get("hub.verify_token") == os.environ["VERIFY_TOKEN"]
):
return request.args.get("hub.challenge", ""), 200, {"Content-Type": "text/plain"}
abort(403)
def signature_ok(raw: bytes, header: str) -> bool:
expected = "sha256=" + hmac.new(APP_SECRET, raw, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected.encode(), header.encode())
@app.post("/webhook")
def receive():
raw = request.get_data() # the exact bytes; read them before touching request.json
if not signature_ok(raw, request.headers.get("X-Hub-Signature-256", "")):
abort(401)
payload = request.get_json(silent=True) or {}
threading.Thread(target=process, args=(payload,), daemon=True).start()
return "", 200
def process(payload: dict) -> None:
for entry in payload.get("entry", []):
for change in entry.get("changes", []):
value = change.get("value", {})
for msg in value.get("messages", []):
if msg["id"] in seen:
continue # Meta retries, so duplicates happen
seen.add(msg["id"])
user = msg.get("from_user_id") or msg.get("from")
last_inbound[user] = datetime.fromtimestamp(int(msg["timestamp"]), timezone.utc)
mark_read(msg["id"])
if msg.get("type") == "text":
send_text(recipient(msg), "Thanks, we have your message. A person will reply here soon.")
for status in value.get("statuses", []):
errors = status.get("errors") or [{}]
print(status["id"], status["status"], errors[0].get("code", ""))
A few Python-specific notes. hmac.compare_digest runs in constant time and accepts bytes of different lengths without raising, so a junk header simply fails. Comparing bytes rather than str also avoids a TypeError if a header contains non-ASCII characters.
The thread is there because Meta treats anything other than a 200 as a failure and retries, with decreasing frequency, for up to 7 days. Sending a reply inside the request couples your response time to Meta's API. A daemon thread is fine for a demo; in production, push the payload onto RQ, Celery or a database table and return.
For the full payload shapes, including media, reactions and location messages, see receiving WhatsApp messages through the API, and for every status and error field, WhatsApp message status webhooks.
The FastAPI version, and its two traps
FastAPI is a natural fit for webhooks, but two defaults work against you here.
# app_fastapi.py
import hashlib
import hmac
import json
import os
from fastapi import BackgroundTasks, FastAPI, HTTPException, Query, Request
from fastapi.responses import PlainTextResponse
from app_flask import process # reuse the same processing function
app = FastAPI()
@app.get("/webhook", response_class=PlainTextResponse)
def verify(
mode: str = Query(alias="hub.mode"),
token: str = Query(alias="hub.verify_token"),
challenge: str = Query(alias="hub.challenge"),
):
if mode == "subscribe" and token == os.environ["VERIFY_TOKEN"]:
return challenge # plain text: a JSON string in quotes fails verification
raise HTTPException(status_code=403)
@app.post("/webhook")
async def receive(request: Request, background: BackgroundTasks):
raw = await request.body()
expected = "sha256=" + hmac.new(
os.environ["APP_SECRET"].encode(), raw, hashlib.sha256
).hexdigest()
received = request.headers.get("x-hub-signature-256", "")
if not hmac.compare_digest(expected.encode(), received.encode()):
raise HTTPException(status_code=401)
background.add_task(process, json.loads(raw))
return {"received": True}
Trap one: the dotted names. Python identifiers cannot contain dots, so hub.mode has to be mapped with Query(alias=...). Without the alias, FastAPI looks for a parameter called mode, finds nothing and answers 422.
Trap two: the challenge. Return a plain str from a normal FastAPI route and it is serialised as JSON, with quotes around it. Meta compares the body to the challenge it sent, so "1158201444" with quotes fails. response_class=PlainTextResponse sends the bare value.
Do not declare a Pydantic model for the POST body either. Reading the model means FastAPI parses the JSON first, and you need the untouched bytes for the HMAC. BackgroundTasks runs process after the 200 has gone out, and because process is a plain function using requests, FastAPI runs it in its thread pool rather than blocking the event loop.
Run it with uvicorn app_fastapi:app --port 8000 behind an HTTPS tunnel. Meta requires a valid certificate.
Tenashi's API from Python, with retries that never double-send
When messages should come from the number your team already uses on WhatsApp, Tenashi's API is shorter: no templates, no window, one endpoint. The interesting part in Python is retrying safely, because a timeout does not tell you whether the message was queued.
# tenashi_client.py: send through Tenashi with safe retries
import os
import time
import uuid
import requests
API = "https://usetenashi.com/api/v1"
RETRYABLE = {"rate_limited", "idempotency_in_progress", "internal", "unavailable"}
class TenashiError(Exception):
def __init__(self, status, code, message):
super().__init__(f"{status} {code}: {message}")
self.status, self.code = status, code
def tenashi_send(text, *, to=None, conversation_id=None, idempotency_key=None, attempts=4):
body = {"text": text}
body.update({"conversationId": conversation_id} if conversation_id else {"to": to})
headers = {
"Authorization": f"Bearer {os.environ['TENASHI_KEY']}",
# The same key on every retry: Tenashi returns the first response, sends once.
"Idempotency-Key": idempotency_key or str(uuid.uuid4()),
}
for attempt in range(attempts):
r = requests.post(f"{API}/messages", json=body, headers=headers, timeout=10)
if r.status_code == 202:
return r.json()["message"] # status "queued"
try:
err = r.json().get("error", {})
except ValueError:
err = {"code": "unavailable", "message": r.text[:200]}
if err.get("code") in RETRYABLE and attempt < attempts - 1:
time.sleep(int(r.headers.get("Retry-After", 2 ** attempt)))
continue
# new_chat_limit, send_blocked, invalid_phone...: retrying will not help
raise TenashiError(r.status_code, err.get("code"), err.get("message"))
Build the key from something stable, such as f"invoice-{invoice_id}-reminder", and a crashed job that reruns tomorrow morning still sends once, as long as it runs within 24 hours. The guide to duplicate WhatsApp API messages covers key design in more depth.
Notice which errors are not retried. new_chat_limit means the number has opened its 50 new chats for the day, and send_blocked means the same text went to 5 chats within 10 minutes. Both are ban protection, and the fix is to wait or personalise the text, not to hammer the endpoint.
Verifying Tenashi webhooks follows the same raw-body rule, with the timestamp folded into the signed string:
import hashlib, hmac, time
def verify_tenashi(headers, raw_body: bytes, secret: str) -> bool:
timestamp = headers["X-Tenashi-Timestamp"]
signature = headers["X-Tenashi-Signature"].removeprefix("v1=")
expected = hmac.new(
secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256
).hexdigest()
fresh = abs(time.time() - int(timestamp)) < 300
return fresh and hmac.compare_digest(signature, expected)
Call it with request.headers and request.get_data() in Flask. Events arrive at least once, so skip any event id you have already processed, and use data.message.chat.type to tell a DM from a group.
Where Tenashi fits
If your Python service has to send OTPs, run a verified green-tick sender or push high-volume alerts, Meta's Cloud API code above is the right tool. If the job is simpler, such as reminders, follow-ups and lead alerts from the number your customers already save, the Cloud API's templates and window are overhead you do not need.
Tenashi links your existing WhatsApp or WhatsApp Business number with a QR scan. Your scripts send through its API, the messages appear in a shared team inbox with a CRM under each chat, and replies come back as signed webhooks, from DMs and groups. The honest limits: it is not Meta's platform, it has no templates or interactive buttons, and sends are paced at one every 4 to 12 seconds per number, so a batch of 100 takes about 10 minutes. Pacing lowers ban risk but WhatsApp can still ban a number that spams. From $18/mo.
Read the API docs · Get started
FAQ
Should I use Flask or FastAPI for a WhatsApp webhook?
Either works. Flask is simpler and request.get_data() gives you the raw body directly. FastAPI suits async stacks but needs a Query alias for Meta's dotted parameters and a PlainTextResponse for the challenge. Pick the framework the rest of your service already uses; the verification logic is identical.
Is there an official WhatsApp API library for Python?
Meta documents the Cloud API as plain HTTP endpoints, and this tutorial uses only requests and a web framework. Community libraries exist, but a thin module of your own is easier to keep in step with Graph API version changes and the 2026 move to business-scoped user IDs, which older libraries may not handle yet.
Why does Meta say my verify token is wrong when it matches?
Usually the challenge, not the token, is the problem. If your endpoint returns the challenge as JSON, wrapped in quotes, or adds a newline, verification fails. Return the exact hub.challenge value as plain text with status 200. Also check that the route accepts GET and that no auth middleware intercepts the request first.
How do I avoid sending duplicate WhatsApp replies from Python?
Deduplicate on the incoming message ID before doing anything, because Meta retries deliveries and can send the same event twice. Store handled IDs in Redis or a table with a unique constraint. For outbound sends through Tenashi, add an Idempotency-Key header so a retried request returns the first response without sending again.
Can a Python script send to many WhatsApp numbers at once?
On the Cloud API, yes, within your messaging limit: 250 unique users outside the service window per 24 hours for new portfolios, rising as you scale. Throughput defaults to 80 messages per second per number. On Tenashi, sends are queued and paced per number, with 50 new chats per day through the API.
Python makes the WhatsApp side easy once the webhook basics are right. For pricing, templates and limits in one place, read the complete guide to the WhatsApp API, and to send from your team's own number, start with the Tenashi API docs.