Write Your Endpoint To Process Webhook Events

In this tutorial, we will create a webhook endpoint with Python and Flask, then we will expose this local endpoint to the internet, then we will register it to HeyGen and start receiving video information to our endpoint. Let's get started! 🙌

Let's Start Building Your Endpoint

We'll begin by crafting a simple endpoint application using Python and Flask. To do this, make sure you have the Flask library installed. You can install it using the following command:

pip install Flask

Next, create your webhook_test endpoint. We will accept POST as a method because HeyGen will make a POST request to our endpoint.

from flask import Flask, request

app = Flask(__name__)

@app.route("/webhook_test", methods=["POST"])
def webhook_callback_test():
    print(request.json)
    return "Success!", 200

if __name__ == "__main__":
    app.run(debug=True)

After creating this Python file, save it and run your Flask application using the following command:

python app.py
 * Serving Flask app
 * Debug mode: on
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000

Now, your local endpoint is up and running. When you make a POST request to 127.0.0.1:5000/webhook_test with a JSON body, you'll see the output in your console.

curl --location 'http://127.0.0.1:5000/webhook_test' \
--header 'Content-Type: application/json' \
--data '{"test": "test"}'
Success!
127.0.0.1 - "POST /webhook_test HTTP/1.1" 200 -
{'test': 'test'}

It works successfully in local, but we need to expose it to the internet so we can receive requests from HeyGen to our endpoint in the later stages.

Exposing Your Endpoint to the Internet

In a development environment, your Flask application likely runs on localhost by default, which is not accessible from the internet. To receive webhooks from external services, you can use tools like Ngrok, LocalTunnel, Cloudflare Tunnel or Smee.io to expose your local development server to the internet.

We will use Smee.io for this purpose, as it's free and ensures a consistent endpoint address. This is an advantage as we will save our endpoint in HeyGen.

npm install -g smee-client

Now, head over to https://smee.io/new and obtain a new endpoint address. Then, use the following command to expose your local endpoint:

smee -t http://127.0.0.1:5000/webhook_test -u https://smee.io/<id>

That's it, with this setup, your local endpoint can now receive requests from anywhere on the internet.

Register Endpoint to HeyGen

To complete the integration, you'll need to register your webhook endpoint with HeyGen. Here's how you can do it:

curl --location 'https://api.heygen.com/v1/webhook/endpoint.add' \
--header 'Content-Type: application/json' \
--header 'X-Api-Key: <your-api-key>' \
--data '
{
  "url": "https://smee.io/<id>",
  "events": ["avatar_video.success"]
}'
{
  "code": 100,
  "data": {
    "endpoint_id": "<endpoint_id>",
    "username": "<username>",
    "url": "<your-endpoint>",
    "status": "enabled",
    "events": ["avatar_video.success"],
    "secret": "<secret>",
    "created_at": "2023-06-28T13:34:51.293528",
    "space_id": "<space_id>"
  },
  "msg": null,
  "message": null
}

Note down the information returned from this request, as you'll need it to secure your endpoint.

Finalizing Your Code

To wrap up the implementation, it's essential to add a layer of security to your endpoint by verifying HeyGen's requests using the provided secret key.

If the request is successful we will print the event_type and event_data that HeyGen has sent to our endpoint.

from flask import Flask, request
import hmac
from hashlib import sha256

app = Flask(__name__)

@app.route("/webhook_test", methods=["POST"])
def webhook_callback_test():
    # Secret key received from HeyGen for verifying the request
    wh_secret = "<secret>"

    # Extracting the content of the request
    content_str = (request.data or b"").decode()

    # Extracting the signature from the request headers
    signature = request.headers.get("signature", "")

    # Calculating the HMAC of the content with the secret key
    mac = hmac.new(
        wh_secret.encode("utf-8"),
        msg=content_str.encode("utf-8"),
        digestmod=sha256,
    )
    computed_signature = mac.hexdigest()

    # Checking if the computed signature matches the received signature
    if computed_signature != signature:
        raise Exception("Invalid request")

    # Processing the event data from the request JSON
    req_json = request.json
    event_type = req_json.get('event_type')
    event_data = req_json.get('event_data')

    # Printing event information
    print(f"Event Type: {event_type}; Event Data: {event_data}")
    return "Success!", 200

if __name__ == "__main__":
    app.run(debug=True)
Event Type: avatar_video.success; Event Data: {'video_id': '<video_id>', 'url': '<url>', 'callback_id': '<callback_id>'}
127.0.0.1 - "POST /webhook_test HTTP/1.1" 200 -

Now, you're ready to process webhook events from HeyGen and unlock a world of possibilities for your application. Congratulations on completing this integration journey!

Conclusion

In this guide, you've created a robust webhook endpoint for HeyGen using Python and Flask. With this endpoint, you can seamlessly receive and process video information, opening doors to interactive and data-rich applications. Your innovative integration with HeyGen's capabilities promises enhanced user experiences, making your projects dynamic and engaging. 🌟