Writing a Basic Webhook App

In this section, we'll write a basic app to get started with receiving webhook events from X.

Here's the requests that the app will need to handle:

• GET requests for security challenge checks
• POST requests for receiving events

After we make the app, we'll need to host it with a public-facing HTTPS URL. I'll show you some options for quick, free methods for getting this stood up quickly so you can test.

The Basic App

Let's start with a basic skeleton for our webhook app. We'll build it incrementally, adding features step by step.

from flask import Flask
from waitress import serve

HOST = "0.0.0.0"
PORT = 8080

app = Flask(__name__)

@app.route('/webhooks', methods=['GET', 'POST'])
def webhook_request():

    if request.method == 'GET':
        # Stub - we'll implement the security check here
        print("Got GET request")
        return '', 200

    elif request.method == 'POST':
        # Stub - we'll implement webhook event handling here
        print("Got POST request")
        return '', 200

    # Got an invalid method
    return 'Method Not Allowed', 405

def main():
    print(f"Hosting WSGI server on {HOST}:{PORT}")
    serve(app, host=HOST, port=PORT)

if __name__=='__main__':
    main()

This gives us a basic Flask app with stub endpoints for both GET and POST requests. Now let's implement each part step by step.

Implementing the Security Check (GET Request)

X uses a Challenge-Response Check (CRC) to verify that your webhook endpoint is legitimate and secure. When you register a webhook URL, X will send a GET request with a crc_token parameter. Your app must respond with an HMAC SHA-256 hash of that token using your consumer secret.

Getting Your Consumer Secret

To get your consumer secret, go to your X Developer Portal and navigate to your app's settings. You'll find the Consumer Keys section where you can view and regenerate your API keys. Copy the "API Key Secret" (also called Consumer Secret) and set it as an environment variable before running your app.

Why This Matters

The CRC check ensures that:

• Your endpoint is accessible and responding
• Only you (with your consumer secret) can validate the webhook
• The connection between X and your server is secure

Implementation

First, let's add the necessary imports and environment variable setup:

from flask import Flask, request, jsonify
from waitress import serve

import base64
import hashlib
import hmac
import os
import json
import sys

app = Flask(__name__)

# Your Twitter consumer secret - set this as an environment variable
CONSUMER_SECRET = os.environ.get("CONSUMER_SECRET")
if CONSUMER_SECRET is None:
  print("Missing consumer secret. Ensure CONSUMER_SECRET env var is set.")
  sys.exit(1)

HOST = "0.0.0.0"
PORT = 8080

Now let's implement the GET request handler:

@app.route('/webhooks', methods=['GET', 'POST'])
def webhook_request():
    # Handle GET request (CRC challenge)
    if request.method == 'GET':
        crc_token = request.args.get('crc_token')
        print(f"CRC Token received: {crc_token}")

        if crc_token is None:
            print("Error: No crc_token found in the request.")
            return json.dumps({'error': 'No crc_token'})

        # Creates HMAC SHA-256 hash from incoming token and your consumer secret
        sha256_hash_digest = hmac.new(
            CONSUMER_SECRET.encode('utf-8'),
            msg=crc_token.encode('utf-8'),
            digestmod=hashlib.sha256
        ).digest()

        # Construct response data with base64 encoded hash
        response = {
            'response_token': 'sha256=' + base64.b64encode(sha256_hash_digest).decode('utf-8')
        }

        return jsonify(response)

    elif request.method == 'POST':
        print("Got POST request")
        return 'Not implemented yet', 501

    return 'Method Not Allowed', 405

How the CRC Check Works

1. X sends a challenge: GET request with crc_token parameter
2. You create an HMAC hash: Using SHA-256 with your consumer secret and the token
3. You respond: With {"response_token": "sha256=<base64_hash>"}
4. X verifies: Compares your hash with what they expect

Implementing Webhook Events (POST Request)

Now let's implement the POST request handler to receive actual webhook events from X. Replace the POST stub with this implementation:

@app.route('/webhooks', methods=['GET', 'POST'])
def webhook_request():
    if request.method == 'GET':
        # Truncated...
        ...

    elif request.method == 'POST':
        event_data = request.get_json()
        if event_data:
            print(json.dumps(event_data, indent=2))
        else:
            print(f"Body: {request.data.decode('utf-8')}")

        return '', 200

    return 'Method Not Allowed', 405

Processing Events

The POST handler:

1. Parses the JSON payload from the request body
2. Logs the event data for debugging and processing
3. Returns 200 OK to acknowledge successful receipt
4. Handles edge cases like non-JSON payloads

Running the Complete App

To run the app:

1. Set your consumer secret:

   export CONSUMER_SECRET="your_consumer_secret_here"

2. Run the app:

   python sample_app.py

from flask import Flask, request, jsonify
from waitress import serve

import base64
import hashlib
import hmac
import os
import json
import sys

app = Flask(__name__)

CONSUMER_SECRET = os.environ.get("CONSUMER_SECRET")
if CONSUMER_SECRET is None:
  print("Missing consumer secret. Ensure CONSUMER_SECRET env var is set.")
  sys.exit(1)

HOST = "0.0.0.0"
PORT = 8080

@app.route('/webhooks', methods=['GET', 'POST'])
def webhook_request():
    if request.method == 'GET':
        crc_token = request.args.get('crc_token')
        print(f"CRC Token received: {crc_token}")

        if crc_token is None:
            print("Error: No crc_token found in the request.")
            return json.dumps({'error': 'No crc_token'})

        sha256_hash_digest = hmac.new(
            CONSUMER_SECRET.encode('utf-8'),
            msg=crc_token.encode('utf-8'),
            digestmod=hashlib.sha256
        ).digest()

        response = {
            'response_token': 'sha256=' + base64.b64encode(sha256_hash_digest).decode('utf-8')
        }
        return jsonify(response)

    elif request.method == 'POST':
        event_data = request.get_json()
        if event_data:
            print(json.dumps(event_data, indent=2))
        else:
            print(f"Body: {request.data.decode('utf-8')}")
        return '', 200

    return 'Method Not Allowed', 405

def main():
    print("--- Starting Webhook App ---")
    print(f"Running with Waitress WSGI server on {HOST}:{PORT}")
    serve(app, host=HOST, port=PORT)

if __name__ == '__main__':
    main()

Next Steps

Now that the server is running locally, you'll need to host it to generate a publicly-available HTTPS URL. This can be done for free using public tools like ngrok, deployed to a cloud environment, or any other means you wish.

In the next section, we'll look at a temporary webhook via xURL — then register the URL with X so we can start receiving events.