Python Examples

Complete examples for integrating GB Chat API with Python applications.

Installation

pip install requests

Basic Usage

import os
import requests

GBCHAT_API_KEY = os.environ['GBCHAT_API_KEY']
BASE_URL = 'https://inboxapi.workmatic.in/api/v1'

def send_message(to, template, language, variables=None, external_id=None):
    """Send a WhatsApp template message."""
    payload = {
        'to': to,
        'template': template,
        'language': language,
    }

    if variables:
        payload['variables'] = variables
    if external_id:
        payload['externalId'] = external_id

    response = requests.post(
        f'{BASE_URL}/message/send',
        headers={
            'X-API-Key': GBCHAT_API_KEY,
            'Content-Type': 'application/json',
        },
        json=payload
    )

    data = response.json()

    if not data['success']:
        raise Exception(f"{data['error']['code']}: {data['error']['message']}")

    return data['data']


# Usage
result = send_message(
    to='919876543210',
    template='order_confirmation',
    language='en',
    variables={'1': 'John Doe', '2': 'ORD-12345'}
)
print(f"Message sent: {result['messageId']}")

Complete API Client

# gbchat.py
import os
import requests
from typing import Optional, Dict
from dataclasses import dataclass


@dataclass
class MessageResult:
    message_id: str
    wa_message_id: str
    status: str
    to: str
    template: str
    external_id: Optional[str] = None


class GBChatError(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message
        super().__init__(f"{code}: {message}")


class GBChatClient:
    def __init__(self, api_key: str, base_url: str = 'https://inboxapi.workmatic.in/api/v1'):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': api_key,
            'Content-Type': 'application/json',
        })

    def send_message(
        self,
        to: str,
        template: str,
        language: str,
        variables: Optional[Dict[str, str]] = None,
        external_id: Optional[str] = None,
        account_id: Optional[str] = None,
    ) -> MessageResult:
        """Send a WhatsApp template message."""
        payload = {
            'to': self._normalize_phone(to),
            'template': template,
            'language': language,
        }

        if variables:
            payload['variables'] = variables
        if external_id:
            payload['externalId'] = external_id
        if account_id:
            payload['accountId'] = account_id

        response = self.session.post(
            f'{self.base_url}/message/send',
            json=payload
        )

        data = response.json()

        if not data['success']:
            raise GBChatError(data['error']['code'], data['error']['message'])

        result = data['data']
        return MessageResult(
            message_id=result['messageId'],
            wa_message_id=result['waMessageId'],
            status=result['status'],
            to=result['to'],
            template=result['template'],
            external_id=result.get('externalId'),
        )

    def _normalize_phone(self, phone: str) -> str:
        """Remove + prefix from phone number."""
        return phone.lstrip('+')


# Usage
if __name__ == '__main__':
    client = GBChatClient(os.environ['GBCHAT_API_KEY'])

    try:
        result = client.send_message(
            to='919876543210',
            template='order_confirmation',
            language='en',
            variables={'1': 'John Doe', '2': 'ORD-12345'},
            external_id='order-123'
        )
        print(f"Message sent: {result.message_id}")
    except GBChatError as e:
        print(f"Error [{e.code}]: {e.message}")

Flask Webhook Handler

import os
import hmac
import hashlib
import time
from flask import Flask, request, jsonify

app = Flask(__name__)

WEBHOOK_SECRET = os.environ['GBCHAT_WEBHOOK_SECRET']


def verify_signature(req) -> bool:
    """Verify the webhook signature."""
    signature = req.headers.get('X-GBChat-Signature', '')
    timestamp = req.headers.get('X-GBChat-Timestamp', '')
    body = req.get_data(as_text=True)

    # Check timestamp (5 minute window)
    current_time = int(time.time())
    try:
        request_time = int(timestamp)
        if abs(current_time - request_time) > 300:
            return False
    except ValueError:
        return False

    # Verify signature
    signature_payload = f"{timestamp}.{body}"
    expected_signature = 'sha256=' + hmac.new(
        WEBHOOK_SECRET.encode(),
        signature_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected_signature)


@app.route('/webhook/gbchat', methods=['POST'])
def webhook():
    # Verify signature
    if not verify_signature(request):
        return jsonify({'error': 'Invalid signature'}), 401

    payload = request.json
    event = payload['event']
    data = payload['data']

    print(f"Received webhook: {event}")
    print(f"  Message ID: {data.get('messageId')}")
    print(f"  External ID: {data.get('externalId')}")
    print(f"  Status: {data.get('status')}")

    # Handle different event types
    if event == 'message.sent':
        handle_message_sent(data)
    elif event == 'message.delivered':
        handle_message_delivered(data)
    elif event == 'message.read':
        handle_message_read(data)
    elif event == 'message.failed':
        handle_message_failed(data)
    elif event == 'webhook.test':
        print("Test webhook received successfully!")

    return 'OK', 200


def handle_message_sent(data):
    print(f"Message {data['messageId']} was sent")
    # Update your database


def handle_message_delivered(data):
    print(f"Message {data['messageId']} was delivered")
    # Update your database


def handle_message_read(data):
    print(f"Message {data['messageId']} was read")
    # Update your database


def handle_message_failed(data):
    print(f"Message {data['messageId']} failed:")
    print(f"  Error: {data.get('error')}")
    # Send alert, retry, etc.


if __name__ == '__main__':
    app.run(port=3000, debug=True)

Django Webhook Handler

# views.py
import os
import hmac
import hashlib
import time
import json
from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

WEBHOOK_SECRET = os.environ['GBCHAT_WEBHOOK_SECRET']


def verify_signature(request) -> bool:
    """Verify the webhook signature."""
    signature = request.headers.get('X-GBChat-Signature', '')
    timestamp = request.headers.get('X-GBChat-Timestamp', '')
    body = request.body.decode('utf-8')

    # Check timestamp (5 minute window)
    current_time = int(time.time())
    try:
        request_time = int(timestamp)
        if abs(current_time - request_time) > 300:
            return False
    except ValueError:
        return False

    # Verify signature
    signature_payload = f"{timestamp}.{body}"
    expected_signature = 'sha256=' + hmac.new(
        WEBHOOK_SECRET.encode(),
        signature_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected_signature)


@csrf_exempt
@require_POST
def gbchat_webhook(request):
    if not verify_signature(request):
        return JsonResponse({'error': 'Invalid signature'}, status=401)

    payload = json.loads(request.body)
    event = payload['event']
    data = payload['data']

    # Process webhook based on event type
    if event == 'message.delivered':
        # Update order notification status
        Order.objects.filter(
            id=data.get('externalId')
        ).update(
            notification_status='delivered'
        )

    return HttpResponse('OK')

Batch Sending with Rate Limiting

import time
from typing import List, Dict, Any
from gbchat import GBChatClient, GBChatError

client = GBChatClient(os.environ['GBCHAT_API_KEY'])


def send_batch(
    messages: List[Dict[str, Any]],
    rate_limit: int = 60
) -> List[Dict[str, Any]]:
    """Send messages in batch with rate limiting."""
    results = []
    delay = 60.0 / rate_limit  # seconds between messages

    for i, message in enumerate(messages):
        try:
            result = client.send_message(**message)
            results.append({
                'success': True,
                'message_id': result.message_id,
                'to': message['to'],
            })
            print(f"Sent {i + 1}/{len(messages)}: {result.message_id}")
        except GBChatError as e:
            results.append({
                'success': False,
                'to': message['to'],
                'error': f"{e.code}: {e.message}",
            })
            print(f"Failed {i + 1}/{len(messages)}: {e.message}")

        # Wait between messages (except for the last one)
        if i < len(messages) - 1:
            time.sleep(delay)

    return results


# Usage
messages = [
    {'to': '919876543210', 'template': 'promo', 'language': 'en', 'variables': {'1': 'Alice'}},
    {'to': '919876543211', 'template': 'promo', 'language': 'en', 'variables': {'1': 'Bob'}},
    {'to': '919876543212', 'template': 'promo', 'language': 'en', 'variables': {'1': 'Charlie'}},
]

results = send_batch(messages)
print(f"Sent: {sum(1 for r in results if r['success'])}/{len(results)}")

Async Support (aiohttp)

import os
import aiohttp
import asyncio
from typing import Optional, Dict


class AsyncGBChatClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = 'https://inboxapi.workmatic.in/api/v1'

    async def send_message(
        self,
        to: str,
        template: str,
        language: str,
        variables: Optional[Dict[str, str]] = None,
        external_id: Optional[str] = None,
    ) -> dict:
        payload = {
            'to': to.lstrip('+'),
            'template': template,
            'language': language,
        }
        if variables:
            payload['variables'] = variables
        if external_id:
            payload['externalId'] = external_id

        async with aiohttp.ClientSession() as session:
            async with session.post(
                f'{self.base_url}/message/send',
                headers={
                    'X-API-Key': self.api_key,
                    'Content-Type': 'application/json',
                },
                json=payload,
            ) as response:
                data = await response.json()

                if not data['success']:
                    raise Exception(f"{data['error']['code']}: {data['error']['message']}")

                return data['data']


# Usage
async def main():
    client = AsyncGBChatClient(os.environ['GBCHAT_API_KEY'])

    result = await client.send_message(
        to='919876543210',
        template='order_confirmation',
        language='en',
        variables={'1': 'John Doe', '2': 'ORD-12345'}
    )
    print(f"Message sent: {result['messageId']}")


asyncio.run(main())