Build a WhatsApp AI Agent with FastAPI and Zavu

Build a high-performance WhatsApp AI agent using FastAPI's async capabilities and Zavu's managed AI Gateway. The best part? You don't need to manage any external AI API keys - Zavu gives you access to all top-tier AI models (GPT-4, Claude, Gemini, Mistral, and more) directly from your dashboard.

What We're Building

A FastAPI application that:

• Receives incoming WhatsApp messages via async webhooks
• Verifies webhook signatures for security
• Uses Zavu's managed AI agents for intelligent responses
• Automatically maintains conversation context

Prerequisites

• Python 3.9+
• A Zavu account with API credentials
• Basic FastAPI knowledge

No external AI API keys required! Zavu's AI Gateway provides access to all major AI models through a single integration.

Installation

Create a new project and install dependencies:

bash
mkdir whatsapp-agent && cd whatsapp-agent
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

pip install fastapi uvicorn zavudev python-dotenv

Project Structure

text
whatsapp-agent/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── services.py
│   └── webhook.py
├── .env
└── requirements.txt

Environment Configuration

Create a .env file:

bash
ZAVUDEV_API_KEY=your_zavu_api_key
ZAVU_WEBHOOK_SECRET=your_webhook_secret

That's it! No OpenAI, Anthropic, or other AI provider keys needed.

How Zavu AI Gateway Works

Zavu provides a unified AI Gateway that gives you access to all top-tier AI models without managing individual API keys:

• GPT-4o, GPT-4o-mini - OpenAI's latest models
• Claude 3.5 Sonnet, Claude 3 Opus - Anthropic's models
• Gemini Pro - Google's AI models
• Mistral Large - Mistral AI's models

You can configure which model to use directly in your Zavu dashboard, and optionally add your own API keys if you prefer to use your own credentials for billing purposes.

Create a Managed AI Agent

The easiest way to add AI capabilities is through Zavu's managed agents. You have two options:

Option 1: Via Zavu Dashboard

Create an agent for your sender directly in the Zavu Dashboard:

Go to Senders in your Zavu dashboard

Select the sender you want to add AI capabilities to

Click on AI Agent tab

Configure your agent:

- Name: Give your agent a descriptive name (e.g., "Customer Support") - Provider: Select "Zavu" to use Zavu's AI Gateway (no external keys needed) - Model: Choose from available models (gpt-4o-mini, claude-3-5-sonnet, etc.) - System Prompt: Define your agent's personality and instructions - Context Window: Set how many previous messages to include for context

Enable the agent and save

With the zavu provider, all AI processing is handled by Zavu's managed gateway. Your agent will automatically respond to incoming WhatsApp messages without any additional code!

Option 2: Via Python SDK

You can also create and configure AI agents programmatically using the zavudev SDK. Add these helper functions to app/services.py:

python
import os
from zavudev import Zavudev

zavu_client = Zavudev(api_key=os.environ.get("ZAVUDEV_API_KEY"))


async def create_ai_agent(sender_id: str, name: str, system_prompt: str, model: str = "gpt-4o-mini") -> dict:
    """Create an AI agent for a sender using Zavu's AI Gateway."""
    response = zavu_client.senders.agent.create(
        sender_id=sender_id,
        name=name,
        provider="zavu",  # Use Zavu's AI Gateway (no external API keys needed)
        model=model,
        system_prompt=system_prompt,
        context_window_messages=10,  # Include last 10 messages for context
        include_contact_metadata=True,  # Include contact info in context
        enabled=True
    )
    return response.agent


def get_agent_stats(sender_id: str) -> dict:
    """Get statistics for an AI agent."""
    response = zavu_client.senders.agent.stats(sender_id)
    return response


def list_agent_executions(sender_id: str, limit: int = 50) -> list:
    """List recent agent executions."""
    response = zavu_client.senders.agent.executions.list(
        sender_id=sender_id,
        limit=limit
    )
    return response.items

Now you can create an agent from FastAPI code:

python
# Example: Create an agent from a startup event or API endpoint
from .services import create_ai_agent

agent = await create_ai_agent(
    sender_id="sender_abc123",
    name="Customer Support Bot",
    model="gpt-4o-mini",
    system_prompt="""You are a helpful customer support assistant for Zavu.

Rules:
- Be friendly and concise
- Help with WhatsApp, SMS, and email messaging questions
- If unsure, offer to connect with a human agent
- Respond in the same language as the customer"""
)

print(f"Agent created: {agent['id']}")

Note: When using provider="zavu", Zavu's AI Gateway handles all AI model access. No external API keys needed!

Available AI Models

When creating an agent, you can choose from these models:

Provider	Models

OpenAI	gpt-4o, gpt-4o-mini, gpt-4-turbo
Anthropic	claude-3-5-sonnet, claude-3-opus, claude-3-haiku
Google	gemini-pro, gemini-1.5-pro
Mistral	mistral-large, mistral-medium

Simply change the model in your agent configuration through the dashboard or SDK.

Define Pydantic Models

Create app/models.py:

python
from pydantic import BaseModel
from typing import Optional
from enum import Enum


class MessageType(str, Enum):
    TEXT = "text"
    IMAGE = "image"
    AUDIO = "audio"
    VIDEO = "video"
    DOCUMENT = "document"


class WebhookMessage(BaseModel):
    id: str
    messageType: MessageType
    text: Optional[str] = None
    from_: Optional[str] = None

    class Config:
        populate_by_name = True
        extra = "allow"

    def __init__(self, data):
        if "from" in data:
            data["from_"] = data.pop("from")
        super().__init__(data)


class WebhookPayload(BaseModel):
    type: str
    data: WebhookMessage

    class Config:
        extra = "allow"

Create the Services Layer

Create app/services.py:

python
import os
import hmac
import hashlib
from zavudev import Zavudev

zavu_client = Zavudev(api_key=os.environ.get("ZAVUDEV_API_KEY"))


def verify_signature(payload: bytes, signature: str) -> bool:
    """Verify webhook signature from Zavu."""
    secret = os.environ.get("ZAVU_WEBHOOK_SECRET", "")

    if not signature or not secret:
        return False

    expected = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(f"sha256={expected}", signature)


async def send_whatsapp_message(to: str, text: str) -> dict:
    """Send a WhatsApp message using Zavu."""
    response = zavu_client.messages.send(
        to=to,
        text=text,
        channel="whatsapp"
    )
    return response.message

Create the Webhook Router

Create app/webhook.py:

python
import logging
from fastapi import APIRouter, Request, HTTPException, Header
from typing import Optional

from .models import WebhookPayload
from .services import verify_signature, send_whatsapp_message

router = APIRouter()
logger = logging.getLogger(__name__)


@router.post("/webhook")
async def handle_webhook(
    request: Request,
    x_zavu_signature: Optional[str] = Header(None, alias="X-Zavu-Signature")
):
    """Handle incoming WhatsApp messages from Zavu."""
    body = await request.body()

    if not verify_signature(body, x_zavu_signature or ""):
        raise HTTPException(status_code=401, detail="Invalid signature")

    try:
        payload = WebhookPayload.model_validate_json(body)
    except Exception as e:
        logger.error(f"Invalid payload: {e}")
        raise HTTPException(status_code=400, detail="Invalid payload")

    if payload.type == "message.inbound":
        await process_inbound_message(payload)

    return {"status": "ok"}


async def process_inbound_message(payload: WebhookPayload):
    """Process incoming message - AI agent handles response automatically."""
    message = payload.data

    if message.messageType != "text":
        logger.info(f"Ignoring non-text message: {message.messageType}")
        return

    sender = message.from_
    text = message.text

    if not sender or not text:
        return

    logger.info(f"Received from {sender}: {text}")

    # With a managed AI agent configured, Zavu automatically handles the AI response
    # The webhook is primarily for logging or custom business logic

    # Example: Escalate to human on specific keyword
    if text.lower() == "human":
        await send_whatsapp_message(
            to=sender,
            text="Connecting you with a human agent..."
        )

Main Application

Create app/main.py:

python
import os
from dotenv import load_dotenv
from fastapi import FastAPI
from contextlib import asynccontextmanager

load_dotenv()

from .webhook import router as webhook_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Starting WhatsApp AI Agent...")
    yield
    print("Shutting down...")


app = FastAPI(
    title="WhatsApp AI Agent",
    description="AI-powered WhatsApp agent using Zavu",
    version="1.0.0",
    lifespan=lifespan
)

app.include_router(webhook_router, prefix="/api")


@app.get("/health")
async def health_check():
    return {"status": "healthy"}

With a managed AI agent configured, Zavu automatically handles the AI response. The webhook is primarily for logging or custom business logic.

Run the Application

bash
uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

Configure Webhook

Deploy your FastAPI app with HTTPS

In your Zavu dashboard:

- Go to Senders > Select sender > Webhook - Set URL: https://yourdomain.com/api/webhook - Enable: message.inbound - Copy the webhook secret to your .env

Create an AI agent for your sender in the Zavu dashboard or via SDK

Local Development with ngrok

bash
ngrok http 8000

Use the HTTPS URL from ngrok in your Zavu webhook settings.

API Documentation

FastAPI automatically generates OpenAPI docs at /docs:

bash
curl http://localhost:8000/docs

Managed Agent Features

Zavu's AI agents come with powerful built-in features:

Knowledge Bases

Create RAG-powered responses by uploading documents to give your agent domain-specific knowledge.

#### Via Dashboard

Go to Senders > Select your sender > AI Agent > Knowledge Bases

Click Create Knowledge Base

Upload documents (PDF, TXT, MD) or paste content directly

The agent will automatically use this knowledge to answer questions

#### Via Python SDK

python
async def create_knowledge_base(sender_id: str, name: str, description: str = None) -> dict:
    """Create a new knowledge base for an agent."""
    response = zavu_client.senders.agent.knowledge_bases.create(
        sender_id=sender_id,
        name=name,
        description=description
    )
    return response.knowledge_base


async def add_document_to_kb(sender_id: str, kb_id: str, title: str, content: str) -> dict:
    """Add a document to a knowledge base."""
    response = zavu_client.senders.agent.knowledge_bases.documents.create(
        sender_id=sender_id,
        kb_id=kb_id,
        title=title,
        content=content
    )
    return response.document


# Example usage
kb = await create_knowledge_base(
    sender_id="sender_abc123",
    name="Product FAQ",
    description="Frequently asked questions about our products"
)

await add_document_to_kb(
    sender_id="sender_abc123",
    kb_id=kb['id'],
    title="Return Policy",
    content="Our return policy allows returns within 30 days of purchase."
)

Custom Tools

Connect external APIs to extend your agent's capabilities by creating tools that the AI can call.

#### Via Dashboard

Go to Senders > Select your sender > AI Agent > Tools

Click Create Tool

Configure the tool with name, description, webhook URL, and parameters

#### Via Python SDK

python
async def create_tool(
    sender_id: str,
    name: str,
    description: str,
    webhook_url: str,
    parameters: dict,
    webhook_secret: str = None
) -> dict:
    """Create a tool for an agent."""
    response = zavu_client.senders.agent.tools.create(
        sender_id=sender_id,
        name=name,
        description=description,
        webhook_url=webhook_url,
        webhook_secret=webhook_secret,
        parameters=parameters
    )
    return response.tool


# Example: Create a tool to check order status
tool = await create_tool(
    sender_id="sender_abc123",
    name="get_order_status",
    description="Get the current status of a customer order",
    webhook_url="https://yourdomain.com/api/tools/order-status",
    webhook_secret="your_tool_webhook_secret",
    parameters={
        "type": "object",
        "properties": {
            "order_id": {
                "type": "string",
                "description": "The order ID to look up"
            }
        },
        "required": ["order_id"]
    }
)

Then create an endpoint to handle tool invocations:

python
@router.post("/tools/order-status")
async def handle_order_tool(request: Request):
    """Handle tool calls from the AI agent."""
    import hmac
    import hashlib

    # Verify tool webhook signature
    signature = request.headers.get('X-Zavu-Tool-Signature', '')
    secret = os.environ.get('ZAVU_TOOL_WEBHOOK_SECRET', '')

    if signature and secret:
        body = await request.body()
        expected = hmac.new(
            secret.encode('utf-8'),
            body,
            hashlib.sha256
        ).hexdigest()
        if not hmac.compare_digest(f"sha256={expected}", signature):
            raise HTTPException(status_code=401)

    payload = await request.json()
    order_id = payload.get('order_id')

    # Call your internal API/service
    order_status = await get_order_from_database(order_id)

    return {
        "success": True,
        "data": {
            "order_id": order_id,
            "status": order_status.status,
            "estimated_delivery": order_status.delivery_date
        }
    }

Now when a customer asks "Where's my order ORD-12345?", the AI will automatically call your tool with the order ID and provide the response.

Conversation Flows

Build guided conversation paths for specific workflows through the dashboard flow builder.

Analytics

Track token usage, costs, and performance metrics directly in your Zavu dashboard under the Analytics section.

Using Your Own AI Credentials (Optional)

If you prefer to use your own AI provider credentials for billing purposes, you can configure them in your Zavu dashboard:

Go to Settings > AI Providers

Add your API keys for OpenAI, Anthropic, Google, or Mistral

When creating an agent, use provider: "openai" (or your preferred provider) instead of provider: "zavu"

This gives you full control over billing while still benefiting from Zavu's unified API and conversation management.

Custom Webhook Logic (Advanced)

If you need custom processing alongside the AI agent, you can add logic in your webhook:

python
@router.post("/webhook")
async def handle_webhook(
    request: Request,
    x_zavu_signature: Optional[str] = Header(None, alias="X-Zavu-Signature")
):
    """Handle incoming WhatsApp messages from Zavu."""
    body = await request.body()

    if not verify_signature(body, x_zavu_signature or ""):
        raise HTTPException(status_code=401, detail="Invalid signature")

    try:
        payload = WebhookPayload.model_validate_json(body)
    except Exception as e:
        logger.error(f"Invalid payload: {e}")
        raise HTTPException(status_code=400, detail="Invalid payload")

    if payload.type == "message.inbound":
        message = payload.data
        sender = message.from_
        text = message.text

        logger.info(f"Message from {sender}: {text}")

        if text and text.lower() == 'human':
            await send_whatsapp_message(
                to=sender,
                text="Connecting you with a human agent..."
            )

    return {"status": "ok"}

Next Steps

• Explore the AI Agents dashboard to monitor conversations
• Add knowledge bases for domain-specific responses
• Create custom tools to integrate with your backend systems
• Build conversation flows for specific use cases
• Add support for media messages (images, documents)

Check out the Zavu documentation for more examples and advanced features.