# This is all it takes to send a WhatsApp message in Python
import requests

response = requests.post(
    'https://whatsapp-messaging-bot.p.rapidapi.com/v1/sendText',
    headers={
        'X-RapidAPI-Key': 'your-key',
        'X-RapidAPI-Host': 'whatsapp-messaging-bot.p.rapidapi.com',
        'Content-Type': 'application/json'
    },
    json={'chatId': '1234567890', 'text': 'Hello from Python!', 'session': 'default'}
)

User sends WhatsApp message
        ↓
WhatsApp API receives message
        ↓
Webhook triggers your FastAPI endpoint
        ↓
Your bot processes message (keyword matching, AI, etc.)
        ↓
Bot sends reply via API
        ↓
User receives response in WhatsApp

mkdir whatsapp-chatbot-python
cd whatsapp-chatbot-python

# Create virtual environment
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install fastapi uvicorn requests python-dotenv pydantic

whatsapp-chatbot-python/
├── main.py                 # FastAPI app
├── bot/
│   ├── __init__.py
│   ├── message_handler.py  # Process incoming messages
│   ├── reply_engine.py     # Generate replies
│   └── whatsapp_client.py  # API wrapper
├── config.py               # Configuration
├── .env                    # API keys
└── requirements.txt

fastapi==0.104.1
uvicorn[standard]==0.24.0
requests==2.31.0
python-dotenv==1.0.0
pydantic==2.5.0
pydantic-settings==2.1.0

# RapidAPI Credentials
RAPIDAPI_KEY=your_rapidapi_key_here
RAPIDAPI_HOST=whatsapp-messaging-bot.p.rapidapi.com

# WhatsApp Session
WHATSAPP_SESSION=my-chatbot

# Server Config
HOST=0.0.0.0
PORT=8000
DEBUG=True

# Webhook Config (for production)
WEBHOOK_URL=https://your-domain.com/webhook
WEBHOOK_SECRET=your-secret-key-here

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # RapidAPI
    rapidapi_key: str
    rapidapi_host: str
    
    # WhatsApp
    whatsapp_session: str
    
    # Server
    host: str = "0.0.0.0"
    port: int = 8000
    debug: bool = True
    
    # Webhook
    webhook_url: str = ""
    webhook_secret: str = ""
    
    # API Base URL
    @property
    def api_base_url(self) -> str:
        return f"https://{self.rapidapi_host}"
    
    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

import requests
from typing import Dict, Any, Optional
from config import get_settings

class WhatsAppClient:
    def __init__(self):
        self.settings = get_settings()
        self.base_url = self.settings.api_base_url
        self.session = self.settings.whatsapp_session
        self.headers = {
            'X-RapidAPI-Key': self.settings.rapidapi_key,
            'X-RapidAPI-Host': self.settings.rapidapi_host,
            'Content-Type': 'application/json'
        }
    
    def send_text_message(self, chat_id: str, text: str) -> Dict[str, Any]:
        """Send a text message to a WhatsApp contact."""
        url = f"{self.base_url}/v1/sendText"
        
        payload = {
            "chatId": chat_id,
            "text": text,
            "session": self.session
        }
        
        try:
            response = requests.post(url, json=payload, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error sending message: {e}")
            return {"error": str(e)}
    
    def send_image(self, chat_id: str, image_url: str, caption: str = "") -> Dict[str, Any]:
        """Send an image message."""
        url = f"{self.base_url}/v1/sendImage"
        
        payload = {
            "chatId": chat_id,
            "file": {"url": image_url, "mimetype": "image/jpeg"},
            "caption": caption,
            "session": self.session
        }
        
        try:
            response = requests.post(url, json=payload, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error sending image: {e}")
            return {"error": str(e)}
    
    def get_contacts(self) -> Dict[str, Any]:
        """Get all contacts."""
        url = f"{self.base_url}/v1/sessions/{self.session}/contacts"
        
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error getting contacts: {e}")
            return {"error": str(e)}
    
    def create_session(self) -> Dict[str, Any]:
        """Create a new WhatsApp session."""
        url = f"{self.base_url}/v1/sessions"
        
        payload = {"name": self.session}
        
        try:
            response = requests.post(url, json=payload, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error creating session: {e}")
            return {"error": str(e)}
    
    def get_qr_code(self) -> Optional[str]:
        """Get QR code for session authentication."""
        url = f"{self.base_url}/v1/sessions/{self.session}/qr"
        
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            data = response.json()
            return data.get('qr')
        except requests.exceptions.RequestException as e:
            print(f"Error getting QR code: {e}")
            return None

from typing import Dict, Optional
import re

class ReplyEngine:
    """
    Generates replies based on incoming messages.
    Supports keyword matching, command handling, and default responses.
    """
    
    def __init__(self):
        # Define your bot's responses
        self.keywords = {
            'hi': 'Hello! 👋 How can I help you today?',
            'hello': 'Hi there! 👋 Welcome to our WhatsApp bot.',
            'help': self._help_message(),
            'pricing': self._pricing_message(),
            'features': self._features_message(),
            'contact': '📞 You can reach us at support@example.com or call +1-555-0123',
            'thanks': "You're welcome! Feel free to ask anything else. 😊",
            'bye': 'Goodbye! Have a great day! 👋',
        }
        
        # Command handlers (messages starting with /)
        self.commands = {
            '/start': 'Welcome! Type "help" to see what I can do.',
            '/help': self._help_message(),
            '/subscribe': 'Great! You\'re now subscribed to updates. 🎉',
            '/unsubscribe': 'You\'ve been unsubscribed. We\'ll miss you! 😢',
            '/status': 'Bot is running smoothly! ✅',
        }
    
    def generate_reply(self, message_text: str, sender: str) -> str:
        """
        Generate a reply based on the incoming message.
        
        Args:
            message_text: The text of the incoming message
            sender: The sender's phone number or ID
            
        Returns:
            The reply text
        """
        message_text = message_text.strip().lower()
        
        # Check for commands first
        if message_text.startswith('/'):
            return self.commands.get(message_text, "Unknown command. Type /help for available commands.")
        
        # Check for keyword matches
        for keyword, reply in self.keywords.items():
            if keyword in message_text:
                return reply
        
        # Check for questions (contains ?)
        if '?' in message_text:
            return self._handle_question(message_text)
        
        # Default response
        return self._default_response()
    
    def _help_message(self) -> str:
        return """
🤖 *Available Commands:*

• Type "pricing" for pricing info
• Type "features" to see what we offer
• Type "contact" to reach our team
• Type "help" to see this message

Or just ask me anything! 💬
""".strip()
    
    def _pricing_message(self) -> str:
        return """
💰 *Our Pricing:*

• Starter: $9/month - 1,000 messages
• Pro: $29/month - 10,000 messages
• Enterprise: Custom pricing

All plans include:
✅ Unlimited contacts
✅ 24/7 support
✅ No setup fees

Reply "contact" to talk to sales!
""".strip()
    
    def _features_message(self) -> str:
        return """
✨ *Key Features:*

📱 Send messages, images, files
👥 Group management
📊 Analytics dashboard
🔒 Secure & encrypted
⚡ Lightning fast delivery
🌍 Global coverage

Want to learn more? Type "contact"
""".strip()
    
    def _handle_question(self, question: str) -> str:
        """Handle questions intelligently."""
        question_lower = question.lower()
        
        if any(word in question_lower for word in ['price', 'cost', 'pay', 'much']):
            return self._pricing_message()
        elif any(word in question_lower for word in ['do', 'can', 'feature', 'what']):
            return self._features_message()
        elif any(word in question_lower for word in ['contact', 'talk', 'speak', 'reach']):
            return self.keywords['contact']
        else:
            return "That's a great question! Type 'contact' to speak with our team who can help you better. 😊"
    
    def _default_response(self) -> str:
        return "Thanks for your message! Type 'help' to see what I can do, or 'contact' to reach our team. 💬"

from typing import Dict, Any
from .whatsapp_client import WhatsAppClient
from .reply_engine import ReplyEngine

class MessageHandler:
    """
    Handles incoming WhatsApp messages and coordinates responses.
    """
    
    def __init__(self):
        self.client = WhatsAppClient()
        self.reply_engine = ReplyEngine()
    
    def process_incoming_message(self, webhook_data: Dict[str, Any]) -> None:
        """
        Process a webhook event from WhatsApp.
        
        Args:
            webhook_data: The webhook payload from WhatsApp API
        """
        try:
            # Extract message details
            event_type = webhook_data.get('event')
            
            if event_type != 'message':
                print(f"Ignoring non-message event: {event_type}")
                return
            
            payload = webhook_data.get('payload', {})
            message = payload.get('body', '')
            sender = payload.get('from', '')
            chat_id = payload.get('chat_id', sender)
            
            # Ignore messages from yourself (bot)
            if payload.get('fromMe', False):
                print("Ignoring message from bot itself")
                return
            
            print(f"📩 Received message from {sender}: {message}")
            
            # Generate reply
            reply_text = self.reply_engine.generate_reply(message, sender)
            
            # Send reply
            result = self.client.send_text_message(chat_id, reply_text)
            
            if 'error' in result:
                print(f"❌ Failed to send reply: {result['error']}")
            else:
                print(f"✅ Sent reply to {sender}: {reply_text[:50]}...")
                
        except Exception as e:
            print(f"❌ Error processing message: {e}")

from fastapi import FastAPI, Request, HTTPException, Header
from fastapi.responses import JSONResponse
from typing import Optional
import uvicorn

from bot.message_handler import MessageHandler
from bot.whatsapp_client import WhatsAppClient
from config import get_settings

app = FastAPI(
    title="WhatsApp Chatbot API",
    description="A production-ready WhatsApp chatbot built with Python and FastAPI",
    version="1.0.0"
)

settings = get_settings()
message_handler = MessageHandler()
whatsapp_client = WhatsAppClient()

@app.get("/")
async def root():
    """Health check endpoint."""
    return {
        "status": "running",
        "message": "WhatsApp Chatbot API is live!",
        "version": "1.0.0"
    }

@app.post("/webhook")
async def webhook(request: Request, x_webhook_secret: Optional[str] = Header(None)):
    """
    Webhook endpoint to receive incoming WhatsApp messages.
    
    In production, validate the webhook secret for security.
    """
    # Validate webhook secret (production)
    if settings.webhook_secret and x_webhook_secret != settings.webhook_secret:
        raise HTTPException(status_code=401, detail="Invalid webhook secret")
    
    try:
        # Get webhook payload
        data = await request.json()
        
        # Process the message
        message_handler.process_incoming_message(data)
        
        return JSONResponse(
            status_code=200,
            content={"status": "success", "message": "Webhook processed"}
        )
    except Exception as e:
        print(f"Webhook error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/send-message")
async def send_message(chat_id: str, text: str):
    """
    Manually send a message to a WhatsApp contact.
    Useful for testing and admin operations.
    """
    result = whatsapp_client.send_text_message(chat_id, text)
    
    if 'error' in result:
        raise HTTPException(status_code=500, detail=result['error'])
    
    return {"status": "success", "data": result}

@app.get("/contacts")
async def get_contacts():
    """Get all WhatsApp contacts."""
    result = whatsapp_client.get_contacts()
    
    if 'error' in result:
        raise HTTPException(status_code=500, detail=result['error'])
    
    return {"status": "success", "data": result}

@app.post("/setup")
async def setup_session():
    """
    Create a WhatsApp session and get QR code for authentication.
    Scan the QR code with your WhatsApp to link the bot.
    """
    # Create session
    session_result = whatsapp_client.create_session()
    
    if 'error' in session_result:
        raise HTTPException(status_code=500, detail=session_result['error'])
    
    # Get QR code
    qr_code = whatsapp_client.get_qr_code()
    
    if not qr_code:
        raise HTTPException(status_code=500, detail="Failed to get QR code")
    
    return {
        "status": "success",
        "message": "Session created! Scan the QR code with WhatsApp.",
        "qr_code": qr_code,
        "instructions": "1. Open WhatsApp on your phone\n2. Go to Settings > Linked Devices\n3. Tap 'Link a Device'\n4. Scan the QR code"
    }

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host=settings.host,
        port=settings.port,
        reload=settings.debug
    )

python main.py

INFO:     Uvicorn running on http://0.0.0.0:8000
INFO:     Application startup complete.

curl -X POST http://localhost:8000/setup

pip install openai

import openai
from config import get_settings

class ReplyEngine:
    def __init__(self):
        self.settings = get_settings()
        openai.api_key = self.settings.openai_api_key  # Add to .env
        # ... existing code ...
    
    def generate_ai_reply(self, message: str, sender: str) -> str:
        """Generate AI-powered reply using OpenAI."""
        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[
                    {"role": "system", "content": "You are a helpful customer support chatbot."},
                    {"role": "user", "content": message}
                ],
                max_tokens=150,
                temperature=0.7
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"OpenAI error: {e}")
            return self._default_response()

pip install sqlalchemy

# database.py
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

Base = declarative_base()

class Conversation(Base):
    __tablename__ = 'conversations'
    
    id = Column(Integer, primary_key=True)
    chat_id = Column(String(50), index=True)
    sender = Column(String(50))
    message = Column(Text)
    reply = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)

engine = create_engine('sqlite:///chatbot.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# Create Procfile
echo "web: uvicorn main:app --host 0.0.0.0 --port \$PORT" > Procfile

# Create runtime.txt
echo "python-3.11.0" > runtime.txt

# Deploy
heroku create my-whatsapp-bot
git push heroku main

# Install Docker
docker build -t whatsapp-chatbot .
docker run -p 8000:8000 whatsapp-chatbot

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

class CustomerSupportEngine(ReplyEngine):
    def generate_reply(self, message_text: str, sender: str) -> str:
        message_lower = message_text.lower()
        
        if 'order' in message_lower and 'status' in message_lower:
            return "Please provide your order number (e.g., #12345)"
        elif 'refund' in message_lower:
            return "I can help with refunds. Reply with your order number."
        elif 'track' in message_lower:
            return "To track your order, visit: https://example.com/track"
        
        return super().generate_reply(message_text, sender)

class LeadCaptureEngine(ReplyEngine):
    def generate_reply(self, message_text: str, sender: str) -> str:
        if message_text.startswith('/'):
            command = message_text
            
            if command == '/start':
                return "Welcome! What's your name?"
            elif command == '/email':
                return "What's your email address?"
            elif command == '/demo':
                return "Great! Our team will contact you within 24 hours."
        
        # Store lead data in database
        # ...
        
        return "Thanks! Type /demo to request a demo."

def handle_ecommerce(message: str) -> str:
    if 'price' in message.lower():
        return "Our products range from $10-$100. Browse at: https://shop.example.com"
    elif 'buy' in message.lower() or 'purchase' in message.lower():
        return "📦 To order:\n1. Visit https://shop.example.com\n2. Add to cart\n3. We'll confirm via WhatsApp!"
    else:
        return "We sell amazing products! Type 'price' to see pricing or 'buy' to order."

# Use async for better performance
from fastapi import BackgroundTasks

@app.post("/webhook")
async def webhook(request: Request, background_tasks: BackgroundTasks):
    data = await request.json()
    
    # Process in background
    background_tasks.add_task(message_handler.process_incoming_message, data)
    
    return {"status": "accepted"}

pip install slowapi

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/webhook")
@limiter.limit("100/minute")
async def webhook(request: Request):
    # ... existing code ...

# Add to message_handler.py
import time

class MessageHandler:
    def __init__(self):
        self.client = WhatsAppClient()
        self.reply_engine = ReplyEngine()
        self.stats = {
            'total_messages': 0,
            'responses_sent': 0,
            'avg_response_time': 0
        }
    
    def process_incoming_message(self, webhook_data: Dict[str, Any]) -> None:
        start_time = time.time()
        
        # ... process message ...
        
        self.stats['total_messages'] += 1
        self.stats['responses_sent'] += 1
        
        response_time = time.time() - start_time
        self.stats['avg_response_time'] = (
            (self.stats['avg_response_time'] * (self.stats['total_messages'] - 1) + response_time)
            / self.stats['total_messages']
        )