How to Write a Custom QGIS Processing Script Algorithm in Python

There is a large gap between a script that works and a tool other people can use. A script has hard-coded paths, no validation, no progress bar, and no way to appear in a model. A Processing algorithm has all four, plus a generated dialog you did not write, a command-line interface you did not write, and a stable id you can call from anywhere. Turning the first into the second is about eighty lines of well-understood boilerplate, and it is the highest-leverage refactor in QGIS automation.

Problem statement

Your one-off script has become something colleagues ask for, and the requests are all the same shape:

• "Can you make it work on my layer?" — the paths are baked into the file.
• "Can I run it without opening a code editor?" — there is no dialog, and there should not have to be a hand-written one.
• "Can we put it in the nightly job?" — it has no id, so nothing can call it but a human.
• "Can it go in the model I built?" — the modeler only accepts registered algorithms.
• "It stopped and I don't know why." — no progress, no cancel, no error reporting beyond a traceback.
• "Which version did we run last quarter?" — no declared parameters, so no record of what was passed.

Wrap the same logic as a QgsProcessingAlgorithm and it appears in the toolbox, the modeler, processing.run() and qgis_process, with validation and progress built in.

Quick answer

Subclass QgsProcessingAlgorithm, declare parameters in initAlgorithm, do the work in processAlgorithm.

from qgis.core import (
    QgsProcessing, QgsProcessingAlgorithm, QgsProcessingException,
    QgsProcessingParameterFeatureSource, QgsProcessingParameterFeatureSink,
    QgsProcessingParameterNumber, QgsFeatureSink, QgsField, QgsFields,
)
from qgis.PyQt.QtCore import QCoreApplication, QVariant


class AddAreaColumn(QgsProcessingAlgorithm):
    INPUT = "INPUT"
    ROUND_TO = "ROUND_TO"
    OUTPUT = "OUTPUT"

    def initAlgorithm(self, config=None):
        self.addParameter(QgsProcessingParameterFeatureSource(
            self.INPUT, self.tr("Input polygons"), [QgsProcessing.TypeVectorPolygon]))
        self.addParameter(QgsProcessingParameterNumber(
            self.ROUND_TO, self.tr("Decimal places"),
            QgsProcessingParameterNumber.Integer, defaultValue=2, minValue=0, maxValue=6))
        self.addParameter(QgsProcessingParameterFeatureSink(
            self.OUTPUT, self.tr("Output layer")))

    def processAlgorithm(self, parameters, context, feedback):
        source = self.parameterAsSource(parameters, self.INPUT, context)
        if source is None:
            raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT))
        places = self.parameterAsInt(parameters, self.ROUND_TO, context)

        fields = QgsFields(source.fields())
        fields.append(QgsField("area_m2", QVariant.Double))

        sink, dest_id = self.parameterAsSink(
            parameters, self.OUTPUT, context, fields, source.wkbType(), source.sourceCrs())

        count = source.featureCount() or 1
        for i, feature in enumerate(source.getFeatures()):
            if feedback.isCanceled():
                break
            attrs = feature.attributes()
            attrs.append(round(feature.geometry().area(), places))
            feature.setAttributes(attrs)
            sink.addFeature(feature, QgsFeatureSink.FastInsert)
            feedback.setProgress(int(100 * i / count))

        return {self.OUTPUT: dest_id}

    def name(self):        return "addareacolumn"
    def displayName(self): return self.tr("Add area column")
    def group(self):       return self.tr("Parcels")
    def groupId(self):     return "parcels"
    def createInstance(self): return AddAreaColumn()
    def tr(self, string):  return QCoreApplication.translate("Processing", string)

Save it through Processing → Scripts → Create New Script from Template and it appears in the toolbox immediately, callable as script:addareacolumn.

out/parcels_area.gpkg 21253 parcels | largest 7,149,833.7 m2 | smallest 0.1 m2

Declare parameters, write to a sink, report through feedback

Declare parameters instead of reading them

initAlgorithm is where a script stops being a script. Every parameter you declare gets a widget in the dialog, a validation rule, a slot in the model builder, and a key in the dictionary processing.run() accepts — from one line.

The parameter types you will use most:

QgsProcessingParameterField is the one worth noticing: it takes parentLayerParameterName="INPUT", so the dialog populates the field list from the layer the user just chose. That is real UI behaviour you get for a keyword argument.

Mark anything genuinely optional with optional=True and give it a defaultValue, so the algorithm can be called with a minimal dictionary.

Read parameters with the typed accessors

Never index parameters directly. The parameterAs* methods resolve values that may be layers, layer ids, paths, expressions, or defaults, and they use context to do it.

source   = self.parameterAsSource(parameters, self.INPUT, context)
distance = self.parameterAsDouble(parameters, self.DISTANCE, context)
places   = self.parameterAsInt(parameters, self.ROUND_TO, context)
field    = self.parameterAsString(parameters, self.FIELD, context)
choice   = self.parameterAsEnum(parameters, self.METHOD, context)
crs      = self.parameterAsCrs(parameters, self.TARGET_CRS, context)
flag     = self.parameterAsBool(parameters, self.DISSOLVE, context)

parameterAsSource returning None means the input could not be resolved — raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT)) so the message names the parameter rather than the traceback naming a line number.

Write through a sink, not a file

parameterAsSink returns a tuple: a QgsFeatureSink and a destination id.

sink, dest_id = self.parameterAsSink(
    parameters, self.OUTPUT, context,
    fields,               # QgsFields for the output
    source.wkbType(),     # geometry type
    source.sourceCrs(),   # CRS
)
if sink is None:
    raise QgsProcessingException(self.invalidSinkError(parameters, self.OUTPUT))

The sink abstracts where the output goes. The same code writes a GeoPackage, a scratch layer, or a PostGIS table depending on what the caller asked for — which is precisely why an algorithm composes into a model and a plain script does not. Return {self.OUTPUT: dest_id} and the framework hands the caller the right thing.

Adding a field means building a new QgsFields rather than mutating the source's:

fields = QgsFields(source.fields())
fields.append(QgsField("area_m2", QVariant.Double))

Use feedback for progress, messages, and cancellation

total = 100.0 / source.featureCount() if source.featureCount() else 0
for current, feature in enumerate(source.getFeatures()):
    if feedback.isCanceled():
        break
    ...
    feedback.setProgress(int(current * total))

feedback.pushInfo(f"processed {current + 1} features")
feedback.pushWarning("3 features had null geometry and were skipped")

Checking isCanceled() inside the loop is what makes the dialog's Cancel button work — without it the button greys out and nothing happens, which users reasonably read as a hang. In a headless run the same object routes messages to your logger, as in running Processing algorithms from Python.

Give it a name, a group, and help

def name(self):        return "addareacolumn"      # id: lowercase, no spaces, stable forever
def displayName(self): return self.tr("Add area column")
def group(self):       return self.tr("Parcels")
def groupId(self):     return "parcels"
def shortHelpString(self):
    return self.tr(
        "Adds an <b>area_m2</b> column computed from each polygon's geometry.\n\n"
        "The layer must be in a projected CRS — areas from a geographic CRS are "
        "in square degrees and meaningless."
    )

name() is the machine id and must never change once anything calls it. displayName() is the human label and can change freely. shortHelpString() becomes the panel beside the dialog and the text qgis_process help prints — the natural place to record the CRS caveat that would otherwise become a support question.

Register it so everything can find it

For interactive use, save the file through Processing → Scripts → Add Script to Toolbox; it lands in the profile's processing/scripts folder and gets the id script:addareacolumn.

For automation, ship a provider with your code and register it explicitly — no profile involved:

from qgis.core import QgsProcessingProvider

class ParcelTools(QgsProcessingProvider):
    def loadAlgorithms(self):
        self.addAlgorithm(AddAreaColumn())
    def id(self):        return "parceltools"
    def name(self):      return "Parcel tools"
    def longName(self):  return self.name()

provider = ParcelTools()                      # keep this reference alive!
QgsApplication.processingRegistry().addProvider(provider)

result = processing.run("parceltools:addareacolumn", {
    "INPUT": "data/rutland.gpkg|layername=parcels",
    "ROUND_TO": 1,
    "OUTPUT": "out/parcels_area.gpkg",
})
areas = [f["area_m2"] for f in QgsVectorLayer(result["OUTPUT"], "a", "ogr").getFeatures()]
print(result["OUTPUT"], len(areas), "parcels | largest", f"{max(areas):,.1f} m2 | smallest", f"{min(areas):,.1f} m2")

out/parcels_area.gpkg 21253 parcels | largest 7,149,833.7 m2 | smallest 0.1 m2

The comment is not decorative. addProvider does not keep the Python object alive, and example 1 shows what happens when nothing else does.

A provider of algorithms for Rutland's parcels

The data used on this page

HM Land Registry's INSPIRE Index Polygons for Rutland (© Crown copyright and database rights, HM Land Registry; OGL): all 21,253 parcels in British National Grid, and a copy of the first 200 in WGS 84 to test the CRS rule. QGIS 4.2 runs headless.

import os
import subprocess
import sys
import textwrap
from pathlib import Path

import requests
from osgeo import gdal

gdal.UseExceptions()
for folder in ("data", "out", "tests/fixtures"):
    Path(folder).mkdir(parents=True, exist_ok=True)
if not Path("data/rutland.gpkg").exists():
    Path("data/Rutland.zip").write_bytes(requests.Session().get(
        "https://use-land-property-data.service.gov.uk/datasets/inspire/download/Rutland.zip", timeout=300).content)
    gdal.VectorTranslate("data/rutland.gpkg", "/vsizip/data/Rutland.zip/Land_Registry_Cadastral_Parcels.gml",
                         layerName="parcels", SQLStatement="SELECT INSPIREID FROM PREDEFINED")
gdal.VectorTranslate("tests/fixtures/parcels_27700.gpkg", "data/rutland.gpkg", layerName="parcels",
                     SQLStatement="SELECT * FROM parcels LIMIT 200")
gdal.VectorTranslate("tests/fixtures/parcels_4326.gpkg", "tests/fixtures/parcels_27700.gpkg", layerName="parcels",
                     dstSRS="EPSG:4326", reproject=True)

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
sys.path.append("/usr/share/qgis/python/plugins")
from qgis.core import Qgis, QgsApplication, QgsProject, QgsVectorLayer
qgs = QgsApplication([], False)
qgs.initQgis()
from processing.core.Processing import Processing
import processing
Processing.initialize()
print("QGIS", Qgis.QGIS_VERSION.split("-")[0], "|", QgsVectorLayer("data/rutland.gpkg|layername=parcels", "p", "ogr").featureCount(), "parcels")

QGIS 4.2.2 | 21253 parcels

Example 1: The provider as a module

The algorithms live in a package in the repository, with one idempotent register():

package = Path("parceltools")
package.mkdir(exist_ok=True)
(package / "__init__.py").write_text(textwrap.dedent('''
    """parceltools - register the provider once, from anywhere."""
    from qgis.core import QgsApplication, QgsProcessingProvider

    _provider = None
    ALGORITHMS = []          # in a real package: [AddAreaColumn, ...] imported from its modules


    class ParcelTools(QgsProcessingProvider):
        def loadAlgorithms(self):
            for algorithm_class in ALGORITHMS:
                self.addAlgorithm(algorithm_class())
        def id(self):       return "parceltools_pkg"
        def name(self):     return "Parcel tools (package)"
        def longName(self): return self.name()


    def register(*algorithm_classes):
        """Idempotent: safe to call from a script, a test, or a plugin."""
        global _provider
        ALGORITHMS.extend(c for c in algorithm_classes if c not in ALGORITHMS)
        registry = QgsApplication.processingRegistry()
        if registry.providerById("parceltools_pkg") is None:
            _provider = ParcelTools()
            registry.addProvider(_provider)
        return registry.providerById("parceltools_pkg")
'''))

sys.path.insert(0, str(Path.cwd()))
import parceltools
first, second = parceltools.register(AddAreaColumn), parceltools.register(AddAreaColumn)
print("same provider both times:", first is second, "|", [a.id() for a in first.algorithms()])

same provider both times: True | ['parceltools_pkg:addareacolumn']

Holding the provider in a module-level global is the simplest correct way to keep the reference alive for the life of the process. Here is what happens without it — a provider registered from inside a function, with nothing holding it afterwards:

probe = textwrap.dedent('''
    import os, sys, gc
    os.environ["QT_QPA_PLATFORM"] = "offscreen"
    sys.path.append("/usr/share/qgis/python/plugins")
    from qgis.core import *
    app = QgsApplication([], False); app.initQgis()
    from processing.core.Processing import Processing; import processing; Processing.initialize()

    class Double(QgsProcessingAlgorithm):
        def initAlgorithm(self, config=None):
            self.addParameter(QgsProcessingParameterNumber("X", "x", defaultValue=1))
            self.addOutput(QgsProcessingOutputNumber("Y", "y"))
        def processAlgorithm(self, p, c, f): return {"Y": self.parameterAsDouble(p, "X", c) * 2}
        def name(self): return "double"
        def displayName(self): return "Double"
        def createInstance(self): return Double()

    class Provider(QgsProcessingProvider):
        def loadAlgorithms(self): self.addAlgorithm(Double())
        def id(self): return "probe"
        def name(self): return "Probe"

    KEEP = sys.argv[1] == "keep"
    def register():
        global kept
        provider = Provider()
        QgsApplication.processingRegistry().addProvider(provider)
        if KEEP:
            kept = provider
    register(); gc.collect()
    print(type(QgsApplication.processingRegistry().providerById("probe")).__name__, end=" | ")
    try:
        print(processing.run("probe:double", {"X": 21}))
    except Exception as exc:
        print(type(exc).__name__ + ":", " ".join(str(exc).split()))
    app.exitQgis()
''')
Path("provider_probe.py").write_text(probe)
for mode in ("keep", "drop"):
    run = subprocess.run([sys.executable, "provider_probe.py", mode], capture_output=True, text=True)
    print(f"{mode}: exit {run.returncode} |", run.stdout.strip().splitlines()[-1] if run.stdout.strip()
          else run.stderr.strip().splitlines()[-1])

keep: exit 0 | Provider | {'Y': 42.0}
drop: exit 0 | QgsProcessingProvider | QgsProcessingException: Error creating algorithm from createInstance()

With the reference dropped, the registry still listed a provider — but only the C++ base object; the Python subclass and its algorithms were gone, and the run failed with "Error creating algorithm from createInstance()". An error message that points at createInstance for a problem that is really object lifetime is exactly why the comment in the registration code is there.

Example 2: A field parameter driven by the input layer

def initAlgorithm(self, config=None):
    self.addParameter(QgsProcessingParameterFeatureSource(
        self.INPUT, self.tr("Input layer"), [QgsProcessing.TypeVectorAnyGeometry]))
    self.addParameter(QgsProcessingParameterField(
        self.GROUP_FIELD, self.tr("Group by field"),
        parentLayerParameterName=self.INPUT,
        type=QgsProcessingParameterField.Any))
    self.addParameter(QgsProcessingParameterEnum(
        self.STAT, self.tr("Statistic"),
        options=["sum", "mean", "min", "max"], defaultValue=1))

Four lines of declaration produce a dialog where choosing a layer repopulates the field dropdown and the statistic is a combo box — behaviour that would be a hundred lines of Qt by hand. (The dialog was not opened for this page; the declaration is the same one parameterDefinitions() reports.)

Example 3: Validating beyond types

Type validation is automatic; domain validation is yours. Override checkParameterValues so bad input is rejected before any work starts.

class AddAreaColumnChecked(AddAreaColumn):
    def checkParameterValues(self, parameters, context):
        source = self.parameterAsSource(parameters, self.INPUT, context)
        if source is not None and source.sourceCrs().isGeographic():
            return False, self.tr(
                "Input is in a geographic CRS; areas would be in square degrees. "
                "Reproject to a projected CRS first."
            )
        return super().checkParameterValues(parameters, context)

    def name(self):           return "addareacolumnchecked"
    def createInstance(self): return AddAreaColumnChecked()


for fixture in ("parcels_27700", "parcels_4326"):
    params = {"INPUT": f"tests/fixtures/{fixture}.gpkg", "ROUND_TO": 1, "OUTPUT": "TEMPORARY_OUTPUT"}
    for alg in (AddAreaColumn().create(), AddAreaColumnChecked().create()):     # create(): initialised copies
        try:
            layer = processing.run(alg, params)["OUTPUT"]
            print(f"{fixture:14} {alg.name():22} first area_m2 = {next(layer.getFeatures())['area_m2']}")
        except Exception as exc:
            print(f"{fixture:14} {alg.name():22} {' '.join(str(exc).split())[:70]}")

bare = processing.run(AddAreaColumn(), {"INPUT": "tests/fixtures/parcels_27700.gpkg", "ROUND_TO": 1,
                                       "OUTPUT": "TEMPORARY_OUTPUT"})["OUTPUT"]
print("a bare AddAreaColumn() instance returned:", type(bare).__name__, bare[:24] + "...")

parcels_27700  addareacolumn          first area_m2 = 321109.1
parcels_27700  addareacolumnchecked   first area_m2 = 321109.1
parcels_4326   addareacolumn          first area_m2 = 0.0
parcels_4326   addareacolumnchecked   Unable to execute algorithm Input is in a geographic CRS; areas would 
a bare AddAreaColumn() instance returned: str Output_layer_f99b252a_8a...

Note the .create(). An algorithm object you construct yourself has not been initialised — initAlgorithm runs when a provider loads it, or when create() makes a copy — so the last line, running a bare instance, returned only the id of its temporary output, a layer that had already gone with the run's own context. Run registered algorithms by id, or pass create() copies.

Without the check, the WGS 84 copy ran and wrote an area_m2 of 0.0 — the parcel's area in square degrees, rounded to one decimal and labelled as metres, which nothing downstream would question. With it, the same input was refused with the reason, before any work started. The message appears in the dialog before the user clicks Run, and in the exception text when called from Python — the same instinct as validating pipeline inputs automatically, applied at the algorithm boundary.

Example 4: An algorithm that wraps other algorithms

Child algorithms inherit your feedback and context, so progress and cancellation flow through.

BUFFER_FIXED_SOURCE = textwrap.dedent('''
    import processing
    from qgis.core import (QgsProcessingAlgorithm, QgsProcessingParameterDistance,
                           QgsProcessingParameterFeatureSink, QgsProcessingParameterFeatureSource)


    class BufferFixed(QgsProcessingAlgorithm):
        INPUT, DISTANCE, OUTPUT = "INPUT", "DISTANCE", "OUTPUT"

        def initAlgorithm(self, config=None):
            self.addParameter(QgsProcessingParameterFeatureSource(self.INPUT, "Input layer"))
            self.addParameter(QgsProcessingParameterDistance(self.DISTANCE, "Distance", 10, self.INPUT))
            self.addParameter(QgsProcessingParameterFeatureSink(self.OUTPUT, "Buffered"))

        def processAlgorithm(self, parameters, context, feedback):
            distance = self.parameterAsDouble(parameters, self.DISTANCE, context)
            feedback.pushInfo("repairing geometries")
            fixed = processing.run("native:fixgeometries",
                                   {"INPUT": parameters[self.INPUT], "OUTPUT": "TEMPORARY_OUTPUT"},
                                   context=context, feedback=feedback, is_child_algorithm=True)["OUTPUT"]
            feedback.pushInfo(f"buffering {type(fixed).__name__} {fixed[:17]}...")
            buffered = processing.run("native:buffer",
                                      {"INPUT": fixed, "DISTANCE": distance, "SEGMENTS": 8,
                                       "DISSOLVE": False, "OUTPUT": parameters[self.OUTPUT]},
                                      context=context, feedback=feedback, is_child_algorithm=True)["OUTPUT"]
            return {self.OUTPUT: buffered}

        def name(self):           return "bufferfixed"
        def displayName(self):    return "Fix and buffer"
        def createInstance(self): return BufferFixed()
''')
exec(BUFFER_FIXED_SOURCE)
from qgis.core import QgsProcessingFeedback


class Messages(QgsProcessingFeedback):
    def __init__(self):
        super().__init__()
        self.lines = []
    def pushInfo(self, info):
        self.lines.append(info)


messages = Messages()
out = processing.run(BufferFixed(), {"INPUT": "tests/fixtures/parcels_27700.gpkg", "DISTANCE": 10,
                                     "OUTPUT": "out/buffer_fixed.gpkg"}, feedback=messages)["OUTPUT"]
print([line for line in messages.lines if not line.startswith("Results")][:2])
print(out, QgsVectorLayer(out, "b", "ogr").featureCount(), "features")

['repairing geometries', 'buffering str Fixed_geometries_...']
out/buffer_fixed.gpkg 200 features

is_child_algorithm=True is essential: it keeps each temporary output in the parent's context and returns its id — the str the second step printed — so the next child can read it, and nothing is cleaned up before the parent finishes.

Example 5: Testing it without a GUI

An algorithm is an ordinary class, so it tests like one. With QGIS already started, the provider from example 1 is registered and ordinary unittest does the rest:

import unittest


class TestAddArea(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        parceltools.register()

    def test_adds_area_column(self):
        layer = processing.run("parceltools_pkg:addareacolumn", {
            "INPUT": "tests/fixtures/parcels_27700.gpkg", "ROUND_TO": 1, "OUTPUT": "TEMPORARY_OUTPUT",
        })["OUTPUT"]
        self.assertIn("area_m2", [f.name() for f in layer.fields()])
        self.assertGreater(next(layer.getFeatures())["area_m2"], 0)

    def test_rejects_geographic_crs(self):
        with self.assertRaises(Exception):
            processing.run(AddAreaColumnChecked().create(), {
                "INPUT": "tests/fixtures/parcels_4326.gpkg", "ROUND_TO": 1, "OUTPUT": "TEMPORARY_OUTPUT",
            })


suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestAddArea)
outcome = unittest.TextTestRunner(stream=open(os.devnull, "w"), verbosity=0).run(suite)
print(f"ran {outcome.testsRun} tests: {len(outcome.failures)} failures, {len(outcome.errors)} errors")

ran 2 tests: 0 failures, 0 errors

The second test is the one that matters: it proves the validation rejects something. A check that has never refused anything has not been shown to work — and example 3 showed that without it, the geographic input sails through.

The page's QGIS session ends here:

QgsProject.instance().clear()
qgs.exitQgis()
print("QGIS closed")

QGIS closed

Declarative parameters

The design idea behind Processing is declarative parameters. Your algorithm says what it needs — a polygon source, an integer between 0 and 6, a sink — and the framework does everything that follows from that declaration: builds a dialog, validates types, exposes the algorithm in the model builder, generates command-line help, records the run in the history, and accepts a dictionary from Python. Every one of those is work you would otherwise write and maintain. The eighty lines of boilerplate are not overhead; they are the interface fee for a very large amount of free infrastructure.

The parameterAs* accessors are where that abstraction pays off in a way that is easy to miss. parameterAsSource might receive a QgsVectorLayer, a layer id from a model's previous step, a file path with a |layername= suffix, a URI with a selection restriction, or a default. Your code sees a QgsFeatureSource and iterates it. The same is true at the other end: parameterAsSink hides whether the caller wanted a GeoPackage, a scratch layer, or a PostGIS table. Writing to a path directly, as a plain script would, throws all of that away and is the single change that makes an algorithm uncomposable.

Sinks also explain a performance property worth knowing. A feature source streams — getFeatures() yields features without materialising the layer — and QgsFeatureSink.FastInsert skips the per-feature id bookkeeping you do not need when writing new features. An algorithm written this way handles a layer larger than memory without any special effort, which a read everything into a list script does not. That is the same reasoning behind chunked reads in the pure-Python world, covered in fixing memory errors with large files.

The last point is about where this sits relative to the rest of the cluster. A Processing model composes existing algorithms and cannot express logic. A script algorithm is logic, wrapped so a model can use it. The two are complementary, and the productive pattern in a mature setup is: custom algorithms for the parts with real decisions in them, a model to wire those together in an analyst-editable graph, and a Python batch loop around the model for files, errors, and reporting. Each layer does the thing it is good at, and each is testable on its own.

Provider references, permanent names and new instances

Keep a reference to your provider

QgsApplication.processingRegistry().addProvider(provider) does not keep the Python object alive. If the local variable goes out of scope the Python side is collected, and in example 1 the next run failed with "Error creating algorithm from createInstance()" — a message that points at the wrong method. Store the provider in a module global or on a long-lived object.

name() is forever

The id is what models, scripts, and scheduled jobs reference. Changing name() silently breaks every caller. Change displayName() freely; treat name() as a published API.

createInstance must return a new object

The framework clones the algorithm for each run. Returning self produces state bleeding between concurrent runs and subtle wrong answers. It is always return MyAlgorithm().

Import from qgis.PyQt

from qgis.PyQt.QtCore import QVariant, never from PyQt5.QtCore import QVariant. The shim resolves to whichever Qt binding the build uses, so the same file keeps working across a Qt5-to-Qt6 QGIS transition.

Script algorithms in the profile versus a provider in your repo

The Scripts folder is right for a personal utility and wrong for anything deployed: it lives in a user profile a service account does not have. For automation, put the class in your repository and register a provider explicitly, exactly as in the examples above.

flags() for long-running or non-thread-safe work

Override flags() to return super().flags() | Qgis.ProcessingAlgorithmFlag.NoThreading (QgsProcessingAlgorithm.FlagNoThreading in QGIS 3) if your algorithm must run on the main thread — anything touching a GUI object, or a library that is not thread-safe. Without it the desktop may run it in a background task and crash.

Progress with an unknown total

source.featureCount() can return -1 for some providers. Guard the division, and fall back to feedback.setProgressText() messages instead of a percentage rather than reporting a nonsense number.

Internal links

• How to Automate QGIS with Python (PyQGIS): The Complete Workflow
• How to Run QGIS Processing Algorithms from Python
• How to Build a QGIS Processing Model and Run It from Python
• How to Batch Process Layers in QGIS with PyQGIS
• How to Run a PyQGIS Script Headless Without Opening QGIS
• How to Validate Pipeline Inputs and Outputs Automatically
• How to Chain GIS Processing Steps into a Reusable Pipeline

FAQ

What is the difference between a script algorithm and a plain PyQGIS script?

A plain script runs top to bottom with whatever paths you typed. An algorithm declares its parameters, which gives it a generated dialog, type validation, progress and cancellation, a stable id, availability in the model builder and in qgis_process, and a dictionary-based Python API. The logic is the same; the difference is that one is callable by anything and the other only by you.

Where should I save my algorithm — the Scripts folder or my repository?

The Scripts folder is fine for a personal tool you use interactively. For anything that runs on a schedule or that colleagues depend on, keep the class in your repository and register a QgsProcessingProvider explicitly, so the algorithm ships with the code and does not depend on a user profile.

Why does my algorithm fail after I register the provider?

Almost certainly because the provider object was garbage-collected while the registry still referenced it. The registry does not keep the Python object alive; in QGIS 4.2 the next run failed with "Error creating algorithm from createInstance()". Assign the provider to a module-level variable, or to an attribute of something long-lived, and the failure disappears.

How do I add a field to the output layer?

Copy the source's fields into a new QgsFields, append your QgsField, and pass that to parameterAsSink. Then for each feature take feature.attributes(), append the new value, and setAttributes before adding it to the sink. Mutating source.fields() directly does not do what it looks like.

How do I make the Cancel button work?

Check feedback.isCanceled() inside your feature loop and break out when it returns True. The framework sets the flag; only your loop can act on it. Without the check the button appears to do nothing, which users report as a freeze.

Can my algorithm call other algorithms?

Yes, and it should for anything already implemented. Call processing.run(...) with context=context, feedback=feedback, is_child_algorithm=True. That keeps progress and cancellation flowing through and stops the framework cleaning up temporary outputs before your parent algorithm has finished with them.

How do I test an algorithm without opening QGIS?

Start a headless QgsApplication, register your provider, and call the algorithm with processing.run inside ordinary unittest or pytest tests, writing to TEMPORARY_OUTPUT and asserting on the result layer. Include at least one test that feeds it invalid input and asserts that it raises — a validation rule that has never rejected anything is untested.