Flask

​

1. Define your task

Your task definition is where you'll perform your tasks, such as sending emails, processing files, or moving data between systems.

With Mergent, tasks are defined by HTTP request handlers inside your application.

from flask import Flask, request
import logging

app = Flask(__name__)

@app.route('/api/tasks', methods=['POST'])
def task_handler():
    body = request.data.decode()

    try:
        perform_task(body)
    except Exception as e:
        logging.error(f"Failed to perform task: {str(e)}")
        return '', 500

    return '', 200

def perform_task(body):
    # This is where you'll perform your task.
    # For now, we'll just log it.
    logging.info(f"Performing task: {body}")

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

​

2. Get your handler's URL

Before you can run your task, your handler must be accessible from the internet.

To do this, you can use a tool like ngrok, or deploy your application to a platform like Vercel or Render.

For more information, see our guide to localhost dev & webhooks.

​

3. Create your first task

Once your task handler is reachable via a URL, you can create your first task. Go to the Mergent Tasks Dashboard and click the Create button.

In this step, you'll set the Request URL to your task handler's URL. You'll also provide the required parameters (in the request body) for your handler to carry out the task.

Click Create to queue the task. Once the task executes, you should see the related log in your console, indicating a successful run. 🎊

​

4. Next Steps

At this point, you have a few options:

import requests
import json

# set the Mergent API key
api_key = "..."

# create a task
# the URL should be set to the URL of your task handler
task = {
    "request": {
        "url": "...",
        "body": "Hello, world!"
    }
}

headers = {
    "Authorization": "Bearer " + api_key
}

response = requests.post("https://api.mergent.io/tasks", headers=headers, json=task)

if response.status_code != 201:
    logging.error("Failed to create task")
else:
    res_task = response.json()
    print(res_task)

​

Learn More