Python Examples

Complete examples for integrating the WhatsApp API with Python.

Installation

pip install requests

Helper Class

# whatsapp_api.py
import requests
from typing import Optional, Dict, Any, List

class WhatsAppAPI:
    def __init__(self, base_url: str, api_key: str, session_id: str):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session_id = session_id
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': api_key,
            'Content-Type': 'application/json'
        })
    
    def _request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
        url = f"{self.base_url}{endpoint}"
        
        try:
            response = self.session.request(method, url, json=data)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            error_msg = e.response.json().get('error', str(e))
            raise Exception(error_msg)
    
    # Session Management
    def get_status(self) -> Dict[str, Any]:
        return self._request('GET', f'/sessions/{self.session_id}/status')
    
    def get_qr(self) -> Dict[str, Any]:
        return self._request('GET', f'/sessions/{self.session_id}/qr')
    
    # Messaging
    def send_text(self, to: str, text: str, quoted_message_id: Optional[str] = None) -> Dict[str, Any]:
        data = {'to': to, 'text': text}
        if quoted_message_id:
            data['quotedMessageId'] = quoted_message_id
        return self._request('POST', f'/sessions/{self.session_id}/send', data)
    
    def send_image(self, to: str, image: str, caption: Optional[str] = None, 
                   filename: Optional[str] = None) -> Dict[str, Any]:
        data = {'to': to, 'image': image}
        if caption:
            data['caption'] = caption
        if filename:
            data['filename'] = filename
        return self._request('POST', f'/sessions/{self.session_id}/send/image', data)
    
    def send_document(self, to: str, document: str, filename: str, 
                      caption: Optional[str] = None) -> Dict[str, Any]:
        data = {'to': to, 'document': document, 'filename': filename}
        if caption:
            data['caption'] = caption
        return self._request('POST', f'/sessions/{self.session_id}/send/document', data)
    
    def send_video(self, to: str, video: str, caption: Optional[str] = None) -> Dict[str, Any]:
        data = {'to': to, 'video': video}
        if caption:
            data['caption'] = caption
        return self._request('POST', f'/sessions/{self.session_id}/send/video', data)
    
    def send_location(self, to: str, latitude: float, longitude: float, 
                      description: Optional[str] = None) -> Dict[str, Any]:
        data = {'to': to, 'latitude': latitude, 'longitude': longitude}
        if description:
            data['description'] = description
        return self._request('POST', f'/sessions/{self.session_id}/send/location', data)
    
    # Utilities
    def check_number(self, phone: str) -> Dict[str, Any]:
        return self._request('GET', f'/sessions/{self.session_id}/check-number/{phone}')
    
    def get_contacts(self) -> Dict[str, Any]:
        return self._request('GET', f'/sessions/{self.session_id}/contacts')
    
    def get_chats(self) -> Dict[str, Any]:
        return self._request('GET', f'/sessions/{self.session_id}/chats')
    
    # Groups
    def create_group(self, name: str, participants: List[str]) -> Dict[str, Any]:
        return self._request('POST', f'/sessions/{self.session_id}/groups/create', {
            'name': name,
            'participants': participants
        })
    
    def get_group_info(self, group_id: str) -> Dict[str, Any]:
        return self._request('GET', f'/sessions/{self.session_id}/groups/{group_id}')
    
    def add_participants(self, group_id: str, participants: List[str]) -> Dict[str, Any]:
        return self._request('POST', 
            f'/sessions/{self.session_id}/groups/{group_id}/participants/add',
            {'participants': participants}
        )

Basic Usage

Send Text Message

from whatsapp_api import WhatsAppAPI

whatsapp = WhatsAppAPI(
    'http://localhost:3000',
    'wask_your_api_key_here',
    'your-session-id'
)

try:
    result = whatsapp.send_text('919876543210', 'Hello from Python! 🚀')
    
    print('Message sent successfully!')
    print(f"Message ID: {result['messageId']}")
except Exception as e:
    print(f"Error: {e}")

Send Image

try:
    result = whatsapp.send_image(
        '919876543210',
        'https://example.com/image.jpg',
        'Check out this image!',
        'photo.jpg'
    )
    
    print('Image sent successfully!')
except Exception as e:
    print(f"Error: {e}")

Send Document

try:
    result = whatsapp.send_document(
        '919876543210',
        'https://example.com/report.pdf',
        'Monthly Report.pdf',
        'Here is your monthly report'
    )
    
    print('Document sent successfully!')
except Exception as e:
    print(f"Error: {e}")

Advanced Examples

Bulk Message Sender

import time
from typing import List, Dict

class BulkMessageSender:
    def __init__(self, whatsapp: WhatsAppAPI, delay_seconds: int = 2):
        self.whatsapp = whatsapp
        self.delay_seconds = delay_seconds
    
    def send_to_multiple(self, recipients: List[str], message: str) -> List[Dict]:
        results = []
        
        for i, recipient in enumerate(recipients):
            try:
                result = self.whatsapp.send_text(recipient, message)
                results.append({
                    'recipient': recipient,
                    'success': True,
                    'messageId': result['messageId']
                })
                
                print(f"✓ Sent to {recipient}")
            except Exception as e:
                results.append({
                    'recipient': recipient,
                    'success': False,
                    'error': str(e)
                })
                
                print(f"✗ Failed to send to {recipient}: {e}")
            
            # Delay between messages
            if i < len(recipients) - 1:
                time.sleep(self.delay_seconds)
        
        return results

# Usage
bulk_sender = BulkMessageSender(whatsapp, 2)

recipients = [
    '919876543210',
    '919876543211',
    '919876543212'
]

message = 'Hello! This is a bulk message.'
results = bulk_sender.send_to_multiple(recipients, message)

# Summary
successful = sum(1 for r in results if r['success'])
failed = len(results) - successful

print(f"\nSummary:")
print(f"Successful: {successful}")
print(f"Failed: {failed}")

Auto-Reply Bot with Flask

from flask import Flask, request, jsonify
from whatsapp_api import WhatsAppAPI
import os

app = Flask(__name__)

whatsapp = WhatsAppAPI(
    os.getenv('API_BASE_URL'),
    os.getenv('API_KEY'),
    os.getenv('SESSION_ID')
)

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.get_json()
    
    # Respond immediately
    response = jsonify({'success': True})
    
    event = payload.get('event')
    
    if event != 'message':
        return response
    
    message = payload.get('message', {})
    
    # Ignore messages from self
    if message.get('fromMe'):
        return response
    
    from_number = message.get('from')
    body = message.get('body', '').lower().strip()
    
    # Auto-reply logic
    reply = None
    
    if 'hello' in body or 'hi' in body:
        reply = 'Hello! How can I help you today?'
    elif 'help' in body:
        reply = 'Available commands:\n- hello - Greeting\n- help - Show this message\n- status - Check status'
    elif 'status' in body:
        reply = 'System is operational! ✅'
    
    # Send reply
    if reply:
        try:
            whatsapp.send_text(from_number, reply)
            print(f"Auto-reply sent to {from_number}")
        except Exception as e:
            print(f"Failed to send auto-reply: {e}")
    
    return response

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3001)

Message Queue with SQLite

import sqlite3
import json
import time
from datetime import datetime
from typing import Optional

class MessageQueue:
    def __init__(self, whatsapp: WhatsAppAPI, db_file: str = 'message_queue.db'):
        self.whatsapp = whatsapp
        self.db_file = db_file
        self._init_db()
    
    def _init_db(self):
        conn = sqlite3.connect(self.db_file)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS message_queue (
                id TEXT PRIMARY KEY,
                recipient TEXT NOT NULL,
                message TEXT NOT NULL,
                type TEXT DEFAULT 'text',
                media TEXT,
                status TEXT DEFAULT 'pending',
                attempts INTEGER DEFAULT 0,
                created_at TEXT,
                sent_at TEXT,
                error TEXT
            )
        ''')
        conn.commit()
        conn.close()
    
    def add(self, to: str, message: str, msg_type: str = 'text', media: Optional[str] = None):
        conn = sqlite3.connect(self.db_file)
        cursor = conn.cursor()
        
        msg_id = f"{int(time.time())}_{id(message)}"
        created_at = datetime.now().isoformat()
        
        cursor.execute('''
            INSERT INTO message_queue (id, recipient, message, type, media, created_at)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (msg_id, to, message, msg_type, media, created_at))
        
        conn.commit()
        conn.close()
        
        print(f"Added message {msg_id} to queue")
    
    def process(self, max_attempts: int = 3) -> int:
        conn = sqlite3.connect(self.db_file)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT id, recipient, message, type, media, attempts
            FROM message_queue
            WHERE status = 'pending'
        ''')
        
        messages = cursor.fetchall()
        processed = 0
        
        for msg_id, recipient, message, msg_type, media, attempts in messages:
            try:
                if msg_type == 'text':
                    result = self.whatsapp.send_text(recipient, message)
                elif msg_type == 'image':
                    result = self.whatsapp.send_image(recipient, media, message)
                
                cursor.execute('''
                    UPDATE message_queue
                    SET status = 'sent', sent_at = ?
                    WHERE id = ?
                ''', (datetime.now().isoformat(), msg_id))
                
                print(f"✓ Sent message {msg_id} to {recipient}")
                processed += 1
                
            except Exception as e:
                attempts += 1
                
                if attempts >= max_attempts:
                    cursor.execute('''
                        UPDATE message_queue
                        SET status = 'failed', attempts = ?, error = ?
                        WHERE id = ?
                    ''', (attempts, str(e), msg_id))
                    
                    print(f"✗ Failed message {msg_id}: {e}")
                else:
                    cursor.execute('''
                        UPDATE message_queue
                        SET attempts = ?
                        WHERE id = ?
                    ''', (attempts, msg_id))
                    
                    print(f"⚠ Retry {attempts}/{max_attempts} for message {msg_id}")
            
            conn.commit()
            time.sleep(2)  # Delay between messages
        
        conn.close()
        return processed
    
    def get_stats(self) -> Dict[str, int]:
        conn = sqlite3.connect(self.db_file)
        cursor = conn.cursor()
        
        cursor.execute('SELECT status, COUNT(*) FROM message_queue GROUP BY status')
        stats = dict(cursor.fetchall())
        
        conn.close()
        return stats

# Usage
queue = MessageQueue(whatsapp)

# Add messages to queue
queue.add('919876543210', 'Message 1')
queue.add('919876543211', 'Message 2')
queue.add('919876543212', 'Message 3')

# Process queue
processed = queue.process()
print(f"\nProcessed {processed} messages")

# Get stats
stats = queue.get_stats()
print("Stats:", stats)

Check Number Before Sending

def send_safe(whatsapp: WhatsAppAPI, phone: str, message: str) -> bool:
    try:
        # Check if number exists
        check = whatsapp.check_number(phone)
        
        if not check['exists']:
            print(f"Number {phone} is not on WhatsApp")
            return False
        
        # Send message
        whatsapp.send_text(phone, message)
        print(f"Message sent to {phone}")
        return True
        
    except Exception as e:
        print(f"Error: {e}")
        return False

# Usage
send_safe(whatsapp, '919876543210', 'Hello!')

Database Integration

Store Messages in PostgreSQL

import psycopg2
from datetime import datetime

class MessageDatabase:
    def __init__(self, host: str, database: str, user: str, password: str):
        self.conn = psycopg2.connect(
            host=host,
            database=database,
            user=user,
            password=password
        )
        self._init_db()
    
    def _init_db(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS messages (
                id SERIAL PRIMARY KEY,
                message_id VARCHAR(255) UNIQUE NOT NULL,
                direction VARCHAR(20) NOT NULL,
                sender VARCHAR(255),
                recipient VARCHAR(255),
                body TEXT,
                type VARCHAR(50) DEFAULT 'text',
                status VARCHAR(50) DEFAULT 'sent',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP
            )
        ''')
        self.conn.commit()
    
    def save_outgoing_message(self, to: str, message: str, message_id: str, msg_type: str = 'text'):
        cursor = self.conn.cursor()
        cursor.execute('''
            INSERT INTO messages (message_id, direction, recipient, body, type, status)
            VALUES (%s, 'outgoing', %s, %s, %s, 'sent')
        ''', (message_id, to, message, msg_type))
        self.conn.commit()
    
    def save_incoming_message(self, message_id: str, from_number: str, body: str, msg_type: str):
        cursor = self.conn.cursor()
        cursor.execute('''
            INSERT INTO messages (message_id, direction, sender, body, type)
            VALUES (%s, 'incoming', %s, %s, %s)
            ON CONFLICT (message_id) DO NOTHING
        ''', (message_id, from_number, body, msg_type))
        self.conn.commit()
    
    def update_message_status(self, message_id: str, status: str):
        cursor = self.conn.cursor()
        cursor.execute('''
            UPDATE messages
            SET status = %s, updated_at = %s
            WHERE message_id = %s
        ''', (status, datetime.now(), message_id))
        self.conn.commit()

# Webhook with database
db = MessageDatabase('localhost', 'whatsapp', 'user', 'password')

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.get_json()
    event = payload.get('event')
    
    if event == 'message':
        message = payload.get('message', {})
        db.save_incoming_message(
            message['id'],
            message['from'],
            message['body'],
            message['type']
        )
    
    if event == 'message_status':
        status_update = payload.get('statusUpdate', {})
        db.update_message_status(
            status_update['messageId'],
            status_update['status']
        )
    
    return jsonify({'success': True})

Error Handling with Retry

import time

def send_with_retry(whatsapp: WhatsAppAPI, to: str, message: str, max_retries: int = 3):
    attempt = 0
    
    while attempt < max_retries:
        try:
            return whatsapp.send_text(to, message)
        except Exception as e:
            attempt += 1
            
            if attempt >= max_retries:
                raise Exception(f"Failed after {max_retries} attempts: {e}")
            
            # Exponential backoff
            delay = 2 ** attempt
            time.sleep(delay)

# Usage
try:
    result = send_with_retry(whatsapp, '919876543210', 'Hello!')
    print(f"Sent: {result['messageId']}")
except Exception as e:
    print(f"Failed: {e}")

Async Support with aiohttp

import aiohttp
import asyncio

class AsyncWhatsAppAPI:
    def __init__(self, base_url: str, api_key: str, session_id: str):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session_id = session_id
    
    async def send_text(self, to: str, text: str):
        url = f"{self.base_url}/sessions/{self.session_id}/send"
        headers = {
            'X-API-Key': self.api_key,
            'Content-Type': 'application/json'
        }
        data = {'to': to, 'text': text}
        
        async with aiohttp.ClientSession() as session:
            async with session.post(url, headers=headers, json=data) as response:
                return await response.json()

# Usage
async def main():
    whatsapp = AsyncWhatsAppAPI('http://localhost:3000', 'wask_key', 'session-id')
    result = await whatsapp.send_text('919876543210', 'Hello!')
    print(result)

asyncio.run(main())