Testing Webhooks using the Nylas CLI and Python

Webhooks are the best way to get notifications without having to constantly ask the server. We’re using Python to demonstrate how this works.

Testing Webhooks using the Nylas CLI and Python

If you are more of a Ruby fan, you can read our post Testing Webhooks with the Nylas CLI and Ruby.

The Nylas CLI is an amazing tool that can be used for a lot of different things, but did you know we can even test Webhooks without the need of an external web server? Before we get started, let’s review what webhooks are, and why they are important.

What are Webhooks?

When something happens in an application, a Webhook, which is an automated message, gets sent. This message which is sent to a unique URL, will have a message or payload, containing important information. They are almost always faster than polling, and don’t require much work from the developers end. It’s like getting an SMS message on your cell phone.

In our environment, Webhooks are notifications triggered by certain events like receiving an email, opening a link inside an email, when an event has been created or deleted and more. Webhooks are important because they are submitted to our applications when something important happens, without the need of our application having the need of pulling information every “x” amount of time.

Why are webhooks important?

A webhook is triggered by the server and it’s delivered to your application without your application having to request it. Instead of having to make multiple requests, your application can wait until it receives a webhook. This makes applications more efficient and most important, faster.

When sending an email or creating an event, it’s important to know if the message was opened, a click inside the messages was clicked or if an event was modified or deleted. Having all this information will help us to make better decisions.

Using the Nylas CLI

Before getting started with the Nylas CLI, we need to install it.

For details on how to install and use the CLI, we can read the blog post Working with the Nylas CLI.

Setting up a Tunnel

In order to test Webhooks on the CLI, we need to create a Webhook Tunnel. This will register a temporary Webhook for us and will keep it open for around 2 hours. It will also provide us with a Web Server and will manage all the Webhook details so we don’t have to worry about anything.

We need to type (The print flag is just so that we can see the output on the CLI)

$ nylas webhook tunnel --print
Open CLI Tunnel

And just like that, our Webhook is created, our Web Server is ready and we only need to trigger some events and see what happens. We can just send an email to ourselves.

Draft Email

We can confirm that the email was received by us

Check your inbox

And check the CLI for the webhook notification

Print CLI Tunnel notifications

Awesome, huh? We got one notification because we got a new email, and another one because we opened the email. What will happen if we delete the email? Yes, we will get another notification.

But, the information is a little bit hard to read, especially if we get a lot of notifications. What we can do is get some help from a Python Web Application.

Using the Nylas CLI and Python

If you already have the Nylas Python SDK installed and your environment is configured, then process with the blog, otherwise, I would recommend you to read the post How to Send Emails with the Nylas Python SDK where everything is clearly explained.

Installing the Flask Package

As we want to create a Python Web Application, our best option is to use Flask, one of the most popular Micro Frameworks in the Python world.

To install it, we can type the following command on the terminal

$ pip3 install flask

Once installed, we’re ready to go.

Creating our Flask Web Application

We’re going to create a small script called WebHooksTest.py just to print out some information coming from the CLI.

# Import your dependencies
from flask import Flask,request

# Create the Flask app and load the configuration
app = Flask(__name__)

# Create a home page and print out the information
@app.route("/", methods=['GET', 'POST'])
def index():
	data = request.get_json()
# Print on the console
	for delta in data["deltas"]:
		print(json.dumps(delta, indent=4, sort_keys=True))
# Send information back to the caller 
	return data

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

When we run this script, it will receive input from the Nylas CLI and print it in a more readable way.

We can run this script from the terminal by typing:

$ python3 WebHooksTest.py
Run Flask Server

Our Web Server is alive on http://127.0.0.1:5000/ but it doesn’t have anything to show and it might give us an error if we open it on the browser as we build it in order to work with the CLI.

NOTE: If you are using Flask on a Mac that has Monterey or higher, you may have trouble accessing localhost. You can solve this by disabling the Airplay Receiver in your Sharing folder. (Airplay Receiver uses port 5000.) Follow these instructions: How to disable Airplay Receiver.

Calling our Flask app from the Nylas CLI

Now, we have everything ready and we can call our Flask Web Application from the Nylas CLI. We can type on the terminal:

$ nylas webhook tunnel -f http://127.0.0.1:5000/
Run CLI Tunnel with Flask Server

The Webhook tunnel will now forward the results to our Flask Web Application console.

If we send ourselves a new message, then we will see this on the console.

Print Flask Server results

And it will also look nicer on the CLI.

Print CLI and Flask Server results

And while this might not look like much, it’s simply the foundation of what’s coming next.

Using Python

We may be wondering, “Haven’t we used Python already?” and the answer would be “Yes, we did. But we did it locally”. Now it’s time to move to the outside world.

What we are going to do is create a Flask Web Application and upload it to Heroku, so it’s freely accessible. This web application is going to use SQLite to store the Webhook information and display it on the browser. While using SQLite on Heroku is not recommended because it will reset itself after a day or every time there’s a new deployment, it’s the easiest way to do some testing and it can be migrated later on to Postgresql if needed.

Setting up a Heroku account and CLI

If we don’t have a Heroku account, we can sign up for one here. Once that’s done, we will need to install the Heroku CLI. I found that the best option is to use Brew.

$ brew tap heroku/brew && brew install heroku

Once that’s done, simply log into the Heroku CLI.

$ heroku login

Creating the Flask Web Application

As we’re going to deploy our application to Heroku, we need to perform some additional steps.

First, let’s create a new folder and call it Webhooks and create the Virtual Environment.

$ mkdir webhooks && cd webhooks
$ python3 -m venv env
$ source venv/bin/activate
$ cd venv

Next, we need to install Flask and create a requirements.txt file that will be used by Heroku in order to know what to install.

$ python3 -m pip install Flask==2.0.3
$ python3 -m pip freeze > requirements.txt

Create a file called app.py

# Import your dependencies
from flask import Flask, request, json, render_template
import hmac
import hashlib
import datetime
import sqlite3
import os
from nylas import APIClient

# Create the Flask app and load the configuration
app = Flask(__name__)

# Connect to the database
def get_db_connection():
    conn = sqlite3.connect('database.db')
    conn.row_factory = sqlite3.Row
    return conn

# Page for the Webhook to send the information to
@app.route("/webhook", methods=['GET', 'POST'])
def webhook():
# We are connected to Nylas, let’s give back the challenge
	if request.method == "GET" and "challenge" in request.args:
		print(" * Nylas connected to the webhook!")
		return request.args["challenge"]
        
# Is it really coming from Nylas?
	is_genuine = verify_signature(
		message=request.data,
		key=os.environ['CLIENT_SECRET'].encode("utf8"),
		signature=request.headers.get("X-Nylas-Signature")
	)
	if not is_genuine:
		return "Signature verification failed!", 401

# Connect to the Nylas API 
	nylas = APIClient(
		os.environ['CLIENT_ID'],
		os.environ['CLIENT_SECRET'],
		os.environ['ACCESS_TOKEN']
	)
                
# Read the webhook information and store it on the database
    data = request.get_json()
    conn = get_db_connection()
    for delta in data["deltas"]:
# Get information about the message
        if delta["type"] == "message.created" or delta["type"] == "message.updated":
            message = nylas.messages.get(delta["object_data"]["id"])	
        elif delta["type"] == "message.opened" or delta["type"] == "message.link_clicked":
	message = nylas.messages.get(delta["object_data"]["metadata"]["message_id"])
#Insert into the database
        for emails in message.to:
            email_to = email_to + ", " + emails["email"]
        conn.execute('INSERT INTO webhooks (id, date, title, from_, to_, type) VALUES (?, ?, ?, ?, ?, ?)', (delta["object_data"]["id"],datetime.datetime.utcfromtimestamp(delta["date"]), message.subject,message.from_[0]["email"],email_to.lstrip(", "),delta["type"]))
        conn.commit()
        conn.close()
        return "Webhook received", 200

# Verify the signature of the webhook sent by Nylas		
def verify_signature(message, key, signature):
    digest = hmac.new(key, msg=message, digestmod=hashlib.sha256).hexdigest()
    return hmac.compare_digest(digest, signature)

# Main page where we can query the database and visualize the results
@app.route("/")
def index():
    conn = get_db_connection()
    webhooks = conn.execute('SELECT * FROM webhooks').fetchall()
    conn.close()
    return render_template('index.html', webhooks=webhooks)

We will notice right away that the code that we are using is more complex than our initial application, and that’s because when Nylas sent us a Webhook, it expects our application to send a response back in order to validate that the service is working properly. Also, we want to validate the signature to make sure that it’s coming from Nylas and not someone else.

We also need to create a subfolder called templates with two files. We can start with base.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>{% block title %} {% endblock %}- Nylas Webhooks</title>
    <style>
        .webhook {
            padding: 10px;
            margin: 5px;
            background-color: #f3f3f3
        }
    </style>
</head>
<body>
    <div class="content">
        {% block content %} {% endblock %}
    </div>
</body>
</html>

and then continue with index.html

{% extends 'base.html' %}

{% block content %}
    <h1>{% block title %} Webhooks {% endblock %}</h1>
    {% for webhook in webhooks %}
        <div class='webhook'>
           <p><b>Id:</b> {{ webhook['id'] }} | <b>Date:</b> {{ webhook['date'] }} |
           <b>Subject:</b> {{ webhook['title'] }} | <b>From:</b> {{ webhook['from_'] }} |  
           <b>To:</b> {{ webhook['to_'] }} | <b>Event:</b> {{ webhook['type'] }}</p>
        </div>
    {% endfor %}
{% endblock %}

As we’re going to use SQLite we need to create a Schema and a Table.

DROP TABLE IF EXISTS webhooks;

CREATE TABLE webhooks (
    id TEXT PRIMARY KEY,
    date TEXT NOT NULL,
    title TEXT NOT NULL, 
    from_ TEXT NOT NULL,
    to_ TEXT NOT NULL,
    type TEXT NOT NULL
);

Running this script using python3 init_db.py will generate the database.db file we need for our application.

import sqlite3

connection = sqlite3.connect('database.db')

with open('schema.sql') as f:
    connection.executescript(f.read())

cur = connection.cursor()

connection.commit()
connection.close()

Deploying our application to Heroku

The first thing we need to do is to initialize our git repository and add our files.

$ git init
$ echo env > .gitignore
$ echo __pycache__ >> .gitignore
$ echo .env >> .gitignore
$ git add .gitignore app.py requirements.txt templates config.json database.db
$ git commit -m “Initializing Git repository” 

We need to tell Heroku how to run our application, by creating a Procfile.

$ echo “web: gunicorn app:app” > Procfile

We need to install it in our folder and update the requirements file.

$ python3 -m pip install gunicorn=20.0.4
$ python3 -m pip freeze > requirements.txt

We must not forget to commit our changes. Otherwise, they will not be taken into account

$ git add Procfile requirements.txt
$ git commit -m “Add deployment files”

Finally, we’re going to create our Heroku application.

$ heroku create blag-webhooks

Application names in Heroku must be unique, so you can change blag with your own name.

Since we’re going to need to access our Nylas Tokens, we don’t want to use/push the .env file into Heroku, so we’re going to use Heroku Environment Variables instead.

We need to go to our Heroku Dashboard and select our application. Then go to Settings,

Heroku Settings

And select Reveal Config Vars.

Heroku Config Vars

This information can be found on our Nylas Dashboard.

With all that ready, we can push our changes and deploy our application.

$ git push heroku master
Pushing Heroku Master

We can now open our web application by typing:

$ heroku open
Running Webhooks app

It looks somehow empty, but that’s because we don’t have any Webhook notifications yet. For that, we need to create a Webhook first.

Creating a Webhook

If we go to our Nylas Dashboard and choose Webhooks

Selecting Webhooks

We can press Create a webhook

Creating a Webhook

Here is the most important piece, as we need to provide the URL of our Heroku Flask Server and choose the Triggers that we want to use.

Passing Webhook URL

For this example we’re going to choose the first three Message Triggers.

Choose Webhook triggers

Notice that message.link_clicked only works when an email is sent from a paid account, so the free trial will not work.

When we press Create Webhook, we can see our webhook listed.

Selected Webhook triggers

Now, if we send an email to our connected account, we will see that the event is saved and displayed on our Flask application. Notice that I haven’t implemented auto refresh, so we need to manually refresh the page.

Sending a test email

We will see the first webhook reflected on the webpage.

Checking email on Webhooks app

Awesome! A new email was received.

So the next step is to send an email from a paid account so that we can track when the email is opened and more important when a link is clicked. We will call this script SendEmailTracking.py

# Load your env variables
from dotenv import load_dotenv
load_dotenv()

# Import your dependencies
import os
from nylas import APIClient

# Initialize your Nylas API client
nylas = APIClient(
    os.environ.get("CLIENT_ID"),
    os.environ.get("CLIENT_SECRET"),
    os.environ.get("ACCESS_TOKEN")
)

# Add tracking capabilities
tracking = {
        "links": "true",
        "opens": "true",
        "thread_replies": "true",
        "payload": "new-payload",
}

# Create the draft email
draft = nylas.drafts.create()
draft.subject = "With Love, from Nylas"
draft.body = "This email was sent using the Python SDK for the  Nylas Email API. Visit <a href='https://nylas.com'>Nylas</a> for details."
draft.to = [{'name': 'Blag', 'email': [email protected]'}]
draft.tracking = tracking

# Send the email!
draft.send()

Notice that the link needs to be between link tags.

Here’s the email that we received.

Checking inbox for new messages

We open it.

Open email with tracking link

And click on the link.

Check all incoming messages on the Webhooks app

All events are correctly logged and displayed.

If you want to learn more about Webhooks, please go to our documentation.

You May Also Like

Why it’s mission-critical to build bi-directional email in your CRM with an email API… now
Why it’s mission-critical to build bi-directional email in your CRM with an email API… now
hero banner
Everything you need to know about Microsoft’s basic authentication deprecation
How we secure APIs at Nylas using JSON Web Tokens