Making a WhatsApp AI Agent with GPT-4o | by Lukasz Kowejsza | Dec, 2024

Since our server is working domestically, the WhatsApp Webhook can’t name the endpoint for verification. What we want is a public URL that can be utilized by the webhook. There are two choices: deploy the appliance to a cloud server or create a proxy server tunnel. Since we’re nonetheless within the improvement course of, we are going to use the second possibility.

1. Go to ngrok Signup and create a free account.
2. Set up ngrok domestically. Relying in your system, you should use Brew, Chocolatey, or just obtain and set up it. See: Setup & Set up.
3. After set up, add your authentication code utilizing the next command in your terminal. Exchange $YOUR-AUTHENTICATION_TOKEN together with your ngrok authentication token, which will be discovered underneath “Your Authtoken” within the ngrok dashboard.
4. Start forwarding visitors out of your localhost on port 8000 by working the next command in your terminal:

> ngrok config add-authtoken $YOUR-AUTHENTICATION_TOKEN
> ngrok http http://localhost:8000
Forwarding                    https://.ngrok.io -> http://localhost:8000

Your native server is now accessible by way of public URLs supplied by ngrok. It’s best to see one thing like this:

Forwarding                    https://.ngrok.io -> http://localhost:8000

Use the HTTPS URL supplied by ngrok for the webhook configuration.

Now allow us to return to Meta’s Cloud API to implement the specified webhook.

1. Navigate to Meta for Builders and choose the app created earlier than.
2. Within the left-hand menu go to WhatsApp > Configuration.
3. Within the Webhook part paste your ngrok HTTPS forwarding URL into the Callback URL area and enter the VERIFICATION_TOKEN outlined in primary.py into the Verification Token area.
4. Click on the verify and save button and anticipate the webhook to confirm your backend.
5. Within the part Webhook Fields allow the messages toggle underneath Subscribed Fields.

That’s it! It’s best to now have the ability to obtain WhatsApp messages in your Python backend server.

Webhooks are HTTP callbacks that allow packages to obtain real-time updates when sure occasions happen equivalent to a brand new message or a standing change. Webhooks make system integrations and automation attainable by delivering an HTTP request containing occasion information to a pre-configured URL (in our case the ngrok proxy server url).

To know the logic and pricing behind webhooks within the Meta cosmos it’s useful to know some fundamental rules about conversations.

A ‘dialog’ on WhatsApp API begins when:
1. The Person sends a message: This opens a 24-hour window, throughout which you’ll reply with messages together with textual content, pictures, or different media with out further prices.

2. The Enterprise Initiates Contact: If no consumer message has been acquired lately (no open 24-hour window), your AI assistant should use a pre-approved template message to start out the dialog. You’ll be able to add customized templates however they should be permitted by Meta.

So long as the consumer retains replying, the 24-hour window resets with every new message. This makes it attainable to have steady interplay with out further prices. A Dialog prices about 0.00–0.08 USD. The concrete pricing relies on you dialog kind Advertising and marketing, Utility, Service and your location. FYI: Service Conversations appear to be these days at no cost. You’ll find the concrete pricing right here: Whatsapp Pricing

Now we’re in a position to obtain messages in our backend. Since we have now subscribed to message objects, every time a message is distributed to your check quantity, the webhook will create a POST request to the callback URL that you simply outlined within the earlier step. What we have to do subsequent is to construct an endpoint for POST requests in our FastAPI utility.

Allow us to first outline the necessities:

• Return a 200 HTTP Standing Code: That is important to tell CloudAPI that the message has been acquired efficiently. Failing to take action will trigger CloudAPI to retry sending the message for as much as 7 days.
• Extract Cellphone Quantity and Message: The payload of the incoming request comprises information that features the cellphone quantity and the message. Which we have to course of within the backend.
• Filter Incoming Objects: Since CloudAPI would possibly ship a number of occasions for a similar message (equivalent to despatched, acquired, and skim), the backend must ensures that just one occasion of the message is processed.
• Deal with A number of Message Sorts: The backend can deal with various kinds of messages, equivalent to textual content, voice messages, and pictures. In an effort to not unfold the scope of the artice we are going to solely lay the muse for pictures however not implement it to the top.
• Course of with LLM-Agent Workflow: The extracted data is processed utilizing the LLM-Agent workflow, which we have now developed with earlier elements of this collection. It’s also possible to use one other agentic implementation, e.g. Langchain or Langgraph

We are going to obtain a payload from a webhook. You’ll find instance payloads in Meta’s documentation: Instance Payload

I favor to jot down my code with Pydantic so as to add kind security to my Python code. Furthermore, kind annotations and Pydantic are an optimum match for FastAPI purposes. So, let’s first outline the fashions utilized in our endpoint:

# app/schema.py
from typing import Checklist, Non-compulsory  
from pydantic import BaseModel, Area   
class Profile(BaseModel):  
title: str  
class Contact(BaseModel):  
profile: Profile  
wa_id: str  
class Textual content(BaseModel):  
physique: str
class Picture(BaseModel):  
mime_type: str  
sha256: str  
id: str  
class Audio(BaseModel):  
mime_type: str  
sha256: str  
id: str  
voice: bool  
class Message(BaseModel):  
from_: str = Area(..., alias="from")  
id: str  
timestamp: str  
textual content: Textual content | None = None  
picture: Picture | None = None  
audio: Audio | None = None  
kind: str
class Metadata(BaseModel):  
display_phone_number: str  
phone_number_id: str
class Worth(BaseModel):  
messaging_product: str  
metadata: Metadata  
contacts: Checklist[Contact] | None = None  
messages: Checklist[Message] | None = None  
class Change(BaseModel):  
worth: Worth  
area: str  
statuses: Checklist[dict] | None = None  
class Entry(BaseModel):  
id: str  
adjustments: Checklist[Change]  
class Payload(BaseModel):  
object: str  
entry: Checklist[Entry]
class Person(BaseModel):  
id: int  
first_name: str  
last_name: str  
cellphone: str
function: str
class UserMessage(BaseModel):  
consumer: Person  
message: str | None = None  
picture: Picture | None = None  
audio: Audio | None = None

Subsequent, we’re going to create some helper capabilities for utilizing dependency injection in FastAPI:

# app/primary.py
from app.area import message_service
def parse_message(payload: Payload) -> Message | None:  
if not payload.entry[0].adjustments[0].worth.messages:  
return None  
return payload.entry[0].adjustments[0].worth.messages[0]  
def get_current_user(message: Annotated[Message, Depends(parse_message)]) -> Person | None:  
if not message:  
return None  
return message_service.authenticate_user_by_phone_number(message.from_)  
def parse_audio_file(message: Annotated[Message, Depends(parse_message)]) -> Audio | None:  
if message and message.kind == "audio":  
return message.audio  
return None  
def parse_image_file(message: Annotated[Message, Depends(parse_message)]) -> Picture | None:  
if message and message.kind == "picture":  
return message.picture  
return None  
def message_extractor(  
message: Annotated[Message, Depends(parse_message)],  
audio: Annotated[Audio, Depends(parse_audio_file)],  
):  
if audio:  
return message_service.transcribe_audio(audio)  
if message and message.textual content:  
return message.textual content.physique  
return None

• Parsing the Payload: The parse_message operate extracts the primary message from the incoming payload if it exists. This operate returns None if no messages are discovered, in order that solely legitimate messages are processed.
• Person Authentication: The get_current_user operate makes use of the parse_message dependency injection to extract the message after which authenticates the consumer primarily based on the cellphone quantity related to the message. Right here we make sure that solely authenticated customers are allowed to ship messages.
• Audio and Picture Parsing: These capabilities extract audio or picture information from the message if the message kind is “audio” or “picture,” respectively. This enables the appliance to deal with various kinds of media.
• Message Extraction: The message_extractor operate makes an attempt to extract textual content from the message or transcribe audio into textual content. This ensures that whatever the message kind, the content material will be processed.

Right here we have now one import from our area layer. The entire script message_service is the place we place all domain-specific code for this implementation, equivalent to authenticate_user_by_phone_number and transcribe_audio.

# app/primary.py
import threading  
from typing_extensions import Annotated  
from fastapi import APIRouter, Question, HTTPException, Relies upon  
from app.area import message_service  
from app.schema import Payload, Message, Audio, Picture, Person  
# ... remainder of the code ...
@app.publish("/", status_code=200)  
def receive_whatsapp(  
consumer: Annotated[User, Depends(get_current_user)],  
user_message: Annotated[str, Depends(message_extractor)],  
picture: Annotated[Image, Depends(parse_image_file)],  
):  
if not consumer and never user_message and never picture:  
return {"standing": "okay"}  
if not consumer:  
elevate HTTPException(status_code=401, element="Unauthorized")  
if picture:  
return print("Picture acquired")  
if user_message:  
thread = threading.Thread(
goal=message_service.respond_and_send_message, 
args=(user_message, consumer)
)  
thread.daemon = True  
thread.begin()  
return {"standing": "okay"}

• POST Endpoint Implementation: This endpoint handles the incoming POST request. It checks if the consumer, message, or picture is legitimate. If none are legitimate, it merely returns a standing message to CloudAPI. If the consumer just isn’t authenticated, it raises an HTTPException with a 401 standing code.
• Processing Photographs and Messages: If a picture is acquired, we make a easy stdout print as a placeholder for future picture dealing with. If a textual content message is acquired, it’s processed asynchronously utilizing a separate thread to keep away from blocking the primary utility thread. The message_service.respond_and_send_message operate is invoked to deal with the message in response to the LLM-Agent workflow.

Rationalization for Utilizing Thread Pooling for the Webhook: WhatsApp will resend the webhook till it will get a 200 response, so thread pooling is used to make sure that message dealing with doesn’t block the webhook response.

In our presentation layer the place we beforehand outlined our endpoint, we use some message_service capabilities that should be outlined subsequent. Particularly, we want an implementation for processing and transcribing audio payloads, authenticating customers, and eventually invoking our agent and sending a response again. We are going to place all this performance inside area/message_service.py. In manufacturing settings, as your utility grows, I’d advocate splitting them additional down into, e.g., transcription_service.py, message_service.py, and authentication_service.py.

In a number of capabilities on this part, we are going to make requests to the Meta API "https://graph.fb.com/...". In all of those requests, we have to embody authorization headers with WHATSAPP_API_KEY, which we created in step 1.3, because the bearer token. I often retailer API keys and tokens in an .env file and entry them with the Python dotenv library. We additionally use the OpenAI consumer together with your OPENAI_API_KEY, which is also saved within the .env file.

However for simplicity, let’s simply place and initialize them on the prime of message_service.py scripts as follows:

import os  
import json  
import requests  
from typing import BinaryIO
WHATSAPP_API_KEY = "YOUR_ACCESS_TOKEN"
llm = OpenAI(api_key="YOUR_OPENAI_API_KEY")

Exchange “YOUR_ACCESS_TOKEN” together with your precise entry token that you simply created in step 1.3.

Dealing with voice information from a WhatsApp webhook just isn’t as easy as it could appear. Initially, you will need to know that the incoming webhook solely tells us the information kind and an object ID. So it doesn’t include the binary audio file. We first must obtain the audio file utilizing Meta’s Graph API. To obtain our acquired audio, we have to make two sequential requests. The primary one is a GET request with the object_id to acquire the obtain URL. This obtain URL is the goal of our second GET request.

def download_file_from_facebook(file_id: str, file_type: str, mime_type: str) -> str | None:  
# First GET request to retrieve the obtain URL  
url = f"https://graph.fb.com/v19.0/{file_id}"  
headers = {"Authorization": f"Bearer {WHATSAPP_API_KEY}"}  
response = requests.get(url, headers=headers)
if response.status_code == 200:  
download_url = response.json().get('url')  
# Second GET request to obtain the file  
response = requests.get(download_url, headers=headers)  
if response.status_code == 200:
# Extract file extension from mime_type    
file_extension = mime_type.cut up('/')[-1].cut up(';')[0]
# Create file_path with extension
file_path = f"{file_id}.{file_extension}"  
with open(file_path, 'wb') as file:  
file.write(response.content material)  
if file_type == "picture" or file_type == "audio":  
return file_path  
elevate ValueError(f"Didn't obtain file. Standing code: {response.status_code}")  
elevate ValueError(f"Didn't retrieve obtain URL. Standing code: {response.status_code}")

Right here, we principally get the obtain URL and obtain the file to the file system utilizing the item ID and the file extension as its file_path. If one thing fails, we elevate a ValueError that signifies the place the error occurred.

Subsequent, we merely outline a operate that takes the audio binary and transcribes it utilizing Whisper:

def transcribe_audio_file(audio_file: BinaryIO) -> str:  
if not audio_file:  
return "No audio file supplied"  
strive:  
transcription = llm.audio.transcriptions.create(  
file=audio_file,  
mannequin="whisper-1",  
response_format="textual content"  
)  
return transcription  
besides Exception as e:  
elevate ValueError("Error transcribing audio") from e

And at last, let’s convey the obtain and transcription capabilities collectively:

def transcribe_audio(audio: Audio) -> str:  
file_path = download_file_from_facebook(audio.id, "audio", audio.mime_type)  
with open(file_path, 'rb') as audio_binary:  
transcription = transcribe_audio_file(audio_binary)  
strive:  
os.take away(file_path)  
besides Exception as e:  
print(f"Didn't delete file: {e}")  
return transcription

Whereas utilizing the check quantity supplied by Meta, we have now to predefine which numbers our chatbot can ship messages to. I’m not fairly certain and haven’t examined if any quantity can ship a message to our chatbot. However anyway, as quickly as we swap to a customized quantity, we don’t need anybody to have the ability to execute our agent chatbot. So we want a technique to authenticate the consumer. We’ve got a number of choices to do that. Initially, we have now to consider the place to retailer consumer data. We might use, for instance, a database like PostgreSQL or a non-relational database like Firestore. We are able to predefine our customers within the file system in a JSON file or in an .env file. For this tutorial, I’ll go along with the only method and hardcode the consumer inside an inventory in our authentication operate.

An inventory entry has the construction of the Person mannequin as outlined in step 5.1. So a consumer consists of an ID, first title, final title, and cellphone quantity. We’ve got not carried out a job system in our agent workflow but. However in most use circumstances with completely different customers, equivalent to within the instance case of a small enterprise assistant, completely different customers can have completely different rights and entry scopes. For now, we simply move "default" as a placeholder function.

def authenticate_user_by_phone_number(phone_number: str) -> Person | None:  
allowed_users = [  
{"id": 1, "phone": "+1234567890", "first_name": "John", "last_name": "Doe", "role": "default"},  
{"id": 2, "phone": "+0987654321", "first_name": "Jane", "last_name": "Smith", "role": "default"}  
]    
for consumer in allowed_users:  
if consumer["phone"] == phone_number:  
return Person(**consumer)  
return None

So simply confirm if the cellphone quantity is in our checklist of allowed_users and return the consumer whether it is. In any other case, we return None. In the event you have a look at our endpoint in step 5.3, you will notice we elevate an error if the consumer is None to forestall additional processing of unauthorized consumer messages.

Now, our final helper operate earlier than we will really invoke our agent is send_whatsapp_message. I’ve included two modes into this operate due to some Meta-specific WhatsApp API logic.

Principally, you aren’t allowed to ship a customized message to a consumer as a dialog starter. This implies you may reply with a person textual content message if the consumer begins the dialog and writes a message to the chatbot first. In any other case, if you need the chatbot to provoke a dialog, you might be restricted to permitted templates, just like the “Hiya World” template.

Additionally necessary to say, once we discuss Meta logic, a dialog after being began opens a dialog window of 24 hours in which you’ll ship messages to that consumer. This dialog window can also be what will get charged, not the person message. It will get a bit extra complicated primarily based on the kind of dialog, equivalent to advertising, assist, and so forth.

It’s also possible to outline a template by yourself and let or not it’s permitted by Meta. I’ve not carried out that at this level, so to check if we will ship a message from our backend to a consumer, I take advantage of the “Hiya World” template. In the event you add some customized permitted templates, you may as well use this operate to ship them to the consumer.

So again to the code. To ship a message, we make a POST request and outline a payload that both contains the textual content physique or the template:

def send_whatsapp_message(to, message, template=True):  
url = f"https://graph.fb.com/v18.0/289534840903017/messages"  
headers = {  
"Authorization": f"Bearer " + WHATSAPP_API_KEY,  
"Content material-Kind": "utility/json"  
}  
if not template:  
information = {  
"messaging_product": "whatsapp",  
"preview_url": False,  
"recipient_type": "particular person",  
"to": to,  
"kind": "textual content",  
"textual content": {  
"physique": message  
}  
}  
else:  
information = {  
"messaging_product": "whatsapp",  
"to": to,  
"kind": "template",  
"template": {  
"title": "hello_world",  
"language": {  
"code": "en_US"  
}  
}  
}  
response = requests.publish(url, headers=headers, information=json.dumps(information))  
return response.json()

Lastly, we will combine our agent from our earlier examples. At this stage, you may as well combine your customized agent, a Langchain AgentExecutor, Langgraph AgentWorkflow, and so forth.

So our primary operate that shall be known as on every incoming message is respond_and_send_message, which takes the user_message string and passes it to our agent workflow because the enter object.

# app/area/message_service.py
import json  
import requests
from app.area.brokers.routing_agent import RoutingAgent  
from app.schema import Person  
def respond_and_send_message(user_message: str, consumer: Person):  
agent = RoutingAgent()  
response = agent.run(user_message, consumer.id)  
send_whatsapp_message(consumer.cellphone, response, template=False)

After invoking our agent, we get a response message that we wish to ship again to the consumer utilizing the send_whatsapp_message operate.