KonPro Logo

Write Your Endpoint To Process Webhook Events

Build a complete webhook endpoint application to handle KonPro events

Welcome to Webhook Endpoint Development! 🔧

In this comprehensive guide, we'll walk you through creating a complete webhook endpoint application using Python and Flask. This endpoint will receive and process webhook events from KonPro, allowing you to handle video generation status updates in real-time.

In this guide, we'll walk you through creating a complete webhook endpoint application using Python and Flask. This endpoint will receive and process webhook events from KonPro, allowing you to handle video generation status updates in real-time.

Prerequisites

📋

What You'll Need

  • Python 3.7 or higher installed on your system
  • Basic knowledge of Python and Flask
  • A KonPro API key
  • A way to expose your local server to the internet (we'll use ngrok)
  • Text editor or IDE of your choice

Step 1: Set Up Flask Application

First, let's create a basic Flask application that can receive webhook events.

python
from flask import Flask, request, jsonify
import json
import logging

app = Flask(__name__)

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    """Handle incoming webhook events from KonPro"""
    try:
        # Get the raw data
        data = request.get_json()
        
        if not data:
            logger.error("No JSON data received")
            return jsonify({"error": "No JSON data"}), 400
        
        # Log the received event
        logger.info(f"Received webhook event: {data.get('event_type', 'unknown')}")
        
        # Process the event based on type
        event_type = data.get('event_type')
        event_data = data.get('event_data', {})
        
        if event_type == 'avatar_video.success':
            handle_video_success(event_data)
        elif event_type == 'avatar_video.fail':
            handle_video_failure(event_data)
        elif event_type == 'video_translate.success':
            handle_translate_success(event_data)
        elif event_type == 'video_translate.fail':
            handle_translate_failure(event_data)
        else:
            logger.warning(f"Unknown event type: {event_type}")
        
        return jsonify({"status": "success"}), 200
        
    except Exception as e:
        logger.error(f"Error processing webhook: {str(e)}")
        return jsonify({"error": "Internal server error"}), 500

def handle_video_success(data):
    """Handle successful video generation"""
    video_id = data.get('video_id')
    video_url = data.get('url')
    gif_url = data.get('gif_download_url')
    
    logger.info(f"Video {video_id} generated successfully!")
    logger.info(f"Video URL: {video_url}")
    logger.info(f"GIF URL: {gif_url}")
    
    # Add your custom logic here
    # For example: save to database, send notification, etc.

def handle_video_failure(data):
    """Handle failed video generation"""
    video_id = data.get('video_id')
    error = data.get('error', 'Unknown error')
    
    logger.error(f"Video {video_id} generation failed: {error}")
    
    # Add your custom error handling logic here

def handle_translate_success(data):
    """Handle successful video translation"""
    video_id = data.get('video_id')
    translated_url = data.get('url')
    
    logger.info(f"Video {video_id} translated successfully!")
    logger.info(f"Translated URL: {translated_url}")

def handle_translate_failure(data):
    """Handle failed video translation"""
    video_id = data.get('video_id')
    error = data.get('error', 'Unknown error')
    
    logger.error(f"Video {video_id} translation failed: {error}")

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)

Step 2: Install Dependencies

Create a requirements.txt file and install the necessary dependencies.

txt
Flask==2.3.3
requests==2.31.0
bash
pip install -r requirements.txt

Step 3: Expose Your Local Server

Since KonPro needs to send webhooks to your endpoint, you need to expose your local server to the internet. We'll use ngrok for this.

Install ngrok

bash
# Download and install ngrok
# Visit https://ngrok.com/download and follow instructions for your OS

# Or use package managers:
# macOS: brew install ngrok
# Windows: choco install ngrok

Start Your Flask App

bash
python app.py

Expose with ngrok

bash
ngrok http 5000

Note: ngrok will provide you with a public URL (e.g., https://abc123.ngrok.io). Use this URL + '/webhook' as your webhook endpoint when registering with KonPro.

Step 4: Register Webhook with KonPro

Now let's register your webhook endpoint with KonPro to start receiving events.

bash
curl --location 'https://api.konpro.ai/v1/webhook/endpoint.add' \
--header 'Content-Type: application/json' \
--header 'x-api-key: <your-api-key>' \
--data '{
  "url": "https://your-ngrok-url.ngrok.io/webhook",
  "events": ["avatar_video.success", "avatar_video.fail", "video_translate.success", "video_translate.fail"]
}'

Step 5: Test Your Webhook

Let's test your webhook endpoint to make sure it's working correctly.

bash
curl -X POST https://your-ngrok-url.ngrok.io/webhook \
  -H "Content-Type: application/json" \
  -d '{
    "event_type": "avatar_video.success",
    "event_data": {
      "video_id": "test_video_123",
      "url": "https://example.com/video.mp4",
      "gif_download_url": "https://example.com/video.gif",
      "video_share_page_url": "https://example.com/share",
      "folder_id": "test_folder",
      "callback_id": "test_callback"
    }
  }'

You should see the event logged in your Flask application console.

Advanced Features

Database Integration

Store webhook events in a database for tracking and analysis.

# Add to your Flask app
import sqlite3
from datetime import datetime

def save_webhook_event(event_type, event_data):
    conn = sqlite3.connect('webhooks.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS webhook_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            event_type TEXT,
            video_id TEXT,
            status TEXT,
            created_at TIMESTAMP
        )
    ''')
    
    cursor.execute('''
        INSERT INTO webhook_events (event_type, video_id, status, created_at)
        VALUES (?, ?, ?, ?)
    ''', (event_type, event_data.get('video_id'), 'processed', datetime.now()))
    
    conn.commit()
    conn.close()

Error Handling & Retries

Implement robust error handling and retry mechanisms.

import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        logger.error(f"Failed after {max_retries} attempts: {str(e)}")
                        raise
                    logger.warning(f"Attempt {attempt + 1} failed: {str(e)}. Retrying in {delay} seconds...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry_on_failure(max_retries=3, delay=2)
def process_video_event(event_data):
    # Your processing logic here
    pass

Security Considerations

Webhook Verification

Consider implementing webhook signature verification to ensure requests are actually from KonPro.

Rate Limiting

Implement rate limiting to prevent abuse and ensure your endpoint can handle the expected load.

HTTPS Only

Always use HTTPS in production to ensure webhook data is encrypted in transit.

Conclusion

You now have a complete webhook endpoint that can receive and process events from KonPro. This foundation allows you to build sophisticated video processing workflows, implement real-time notifications, and maintain comprehensive logs of your video generation activities.

KonPro's Webhook EventsVideo Translate API