DEV Community

VolodymyrPTK
VolodymyrPTK

Posted on

Extending n8n running locally python scripts using Flask

Hi everyone!

I want to share what I've done to extend the functionality of n8n by running Python code using a Flask server, so don't need to pay for Cloudconver and similar APIs. I’m running n8n locally in Docker and have added a new image conversion function that converts images from PNG to Webp format. This is done through a Python script that I’ve integrated into my setup. I’ve also created an n8n node that calls this function, converts the image, and returns the converted image back to my workflow.

Image description
Image description

This ins json for import n8n node, just copy and paste inside n8n workflow

{ "name": "Tests", "nodes": [ { "parameters": { "method": "POST", "url": "http://host.docker.internal:5000/convert", "sendBody": true, "contentType": "multipart-form-data", "bodyParameters": { "parameters": [ { "parameterType": "formBinaryData", "name": "file", "inputDataFieldName": "data" }, { "name": "original_filename", "value": "={{ $binary.data.fileName || 'image.png' }}" } ] }, "options": {} }, "name": "Flask Converter", "type": "n8n-nodes-base.httpRequest", "typeVersion": 4.1, "position": [ -820, 20 ], "id": "483881d3-6574-4eec-91e9-4715e185657c", "alwaysOutputData": false, "notes": "After importing, you must manually set this node to send the file using the instructions provided." } ], "pinData": {}, "connections": {}, "active": false, "settings": { "executionOrder": "v1" }, "versionId": "6ace63f3-e1f3-48ad-a929-c0ca4ba0887c", "meta": { "instanceId": "44c216c02c6f01f2fef8eb2ad71aa75c1545e5519733c0a1e6fb1f2fc1b2fa36" }, "id": "mKN69KiJtr7j68C3", "tags": [] } 
Enter fullscreen mode Exit fullscreen mode

this is python script

import sys import os from PIL import Image from flask import Flask, request, send_file, jsonify import io def convert_png_to_webp(input_path, output_path=None, quality=80): if not input_path.lower().endswith('.png'): raise ValueError('Input file must be a PNG image suka.') if output_path is None: output_path = os.path.splitext(input_path)[0] + '.webp' with Image.open(input_path) as img: img.save(output_path, 'WEBP', quality=quality) print(f"Converted '{input_path}' to '{output_path}' (quality={quality})") def main(): if len(sys.argv) < 2: print("Usage: python png_to_webp_converter.py <input_png> [output_webp] [quality]") sys.exit(1) input_path = sys.argv[1] output_path = sys.argv[2] if len(sys.argv) > 2 else None quality = int(sys.argv[3]) if len(sys.argv) > 3 else 80 try: convert_png_to_webp(input_path, output_path, quality) except Exception as e: print(f"Error: {e}") sys.exit(1) app = Flask(__name__) @app.route('/convert', methods=['POST']) def convert(): if 'file' not in request.files: return jsonify({'error': 'No file part'}), 400 file = request.files['file'] if not file or not getattr(file, 'filename', None): return jsonify({'error': 'No selected file'}), 400 original_filename = request.form.get('original_filename', 'image.png') base_name = os.path.splitext(original_filename)[0] webp_filename = f"{base_name}.webp" try: img = Image.open(file.stream) if img.format != 'PNG': return jsonify({'error': f'Only PNG files are supported. Received: {img.format}'}), 400 file.stream.seek(0) output = io.BytesIO() img.save(output, format='WEBP', quality=80) output.seek(0) return send_file(output, mimetype='image/webp', as_attachment=True, download_name=webp_filename) except Exception as e: return jsonify({'error': f'Error processing image: {str(e)}'}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=True) 
Enter fullscreen mode Exit fullscreen mode

Top comments (0)