---
title: Build a WhatsApp AI Agent with FastAPI and Zavu
description: Learn how to build an async WhatsApp AI agent using FastAPI and Zavu's managed AI Gateway.
date: 2025-12-17
author: Victor Villalobos
locale: en
source: https://www.zavu.dev/en/blog/whatsapp-ai-agents-fastapi
tags: AI Agents, WhatsApp, FastAPI, Python
---

# 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

```
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**:

1. Go to **Senders** in your Zavu dashboard
2. Select the sender you want to add AI capabilities to
3. Click on **AI Agent** tab
4. 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
5. 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

1. Deploy your FastAPI app with HTTPS
2. 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`

3. 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

1. Go to **Senders** > Select your sender > **AI Agent** > **Knowledge Bases**
2. Click **Create Knowledge Base**
3. Upload documents (PDF, TXT, MD) or paste content directly
4. 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

1. Go to **Senders** > Select your sender > **AI Agent** > **Tools**
2. Click **Create Tool**
3. 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:

1. Go to **Settings** > **AI Providers**
2. Add your API keys for OpenAI, Anthropic, Google, or Mistral
3. 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](https://docs.zavu.dev) for more examples and advanced features.