A Deferrable Airflow Operator for CaptchaAI

This guide builds a CaptchaAISolveOperator that submits a CAPTCHA to CaptchaAI in execute(), hands the wait to an async CaptchaAITrigger through self.defer(), and resumes in execute_complete() to use the token. The worker slot is free while CaptchaAI works, instead of sleeping through the 15-second initial wait and every poll after it. Deferral needs Airflow 2.2 or later and at least one running triggerer; the code targets Airflow 3.2+, and the 2.x changes are listed near the end. Put the task in a pool with include_deferred sized to your plan's thread count, or Airflow will start more solves than CaptchaAI can run at once.

The examples assume a portal your team is licensed to query, such as a partner reporting site that puts its daily export behind reCAPTCHA.

What deferring changes during a solve

reCAPTCHA v2 typically solves in under 60 seconds, and Cloudflare Turnstile typically clears in under 10 seconds. CaptchaAI's guidance is to wait 15 to 20 seconds before the first reCAPTCHA poll, then call res.php every 5 seconds. A classic operator spends that window inside a worker slot, asleep. A deferrable operator gives the slot back after the submit, and the triggerer does the polling. The triggerer is a separate asyncio process that runs up to 1,000 triggers at once by default.

Classic operator: one worker slot for the whole solve
worker     [submit] sleep 15 s [poll] 5 s [poll] 5 s [poll: token] [post form]
           0 s                                                     ~20-60 s

Deferrable operator: two short worker visits
worker     [submit]                                                [post form]
triggerer           wait 15 s > poll > poll > poll: token > event
           0 s   ~1 s                                  (hard cap: 120 s)

Sixteen blocking solves on a worker with sixteen slots stall every other task routed to it for up to a minute. Sixteen deferred solves cost sixteen sleeping coroutines on the triggerer.

The trigger: polling res.php on the event loop

Save this as plugins/captchaai_trigger.py:

from __future__ import annotations

import asyncio
import time
from typing import Any, AsyncIterator

import httpx
from airflow.sdk import BaseHook
from airflow.triggers.base import BaseTrigger, TriggerEvent

RES_URL = "https://ocr.captchaai.com/res.php"
SERVER_ERRORS = {"ERROR_SERVER_ERROR", "ERROR_INTERNAL_SERVER_ERROR"}


def parse_body(resp: httpx.Response) -> dict[str, Any]:
    """CaptchaAI can answer in plain text even with json=1, so never trust .json() blindly."""
    try:
        body = resp.json()
        if isinstance(body, dict):
            return body
    except ValueError:
        pass
    return {"status": 0, "request": resp.text.strip()[:200]}


class CaptchaAITrigger(BaseTrigger):
    """Polls res.php for one CaptchaAI task on the triggerer's event loop."""

    def __init__(self, captcha_id: str, submitted_at: float, conn_id: str = "captchaai",
                 poll_interval: float = 5, initial_wait: float = 15, timeout: float = 120):
        super().__init__()
        self.captcha_id = captcha_id
        self.submitted_at = submitted_at  # epoch seconds: survives a triggerer restart
        self.conn_id = conn_id
        self.poll_interval = poll_interval
        self.initial_wait = initial_wait
        self.timeout = timeout

    def serialize(self) -> tuple[str, dict[str, Any]]:
        return (f"{type(self).__module__}.{type(self).__qualname__}", {
            "captcha_id": self.captcha_id, "submitted_at": self.submitted_at,
            "conn_id": self.conn_id, "poll_interval": self.poll_interval,
            "initial_wait": self.initial_wait, "timeout": self.timeout,
        })

    async def run(self) -> AsyncIterator[TriggerEvent]:
        conn = await BaseHook.aget_connection(self.conn_id)  # key never enters kwargs
        params = {"key": conn.password, "action": "get", "id": self.captcha_id, "json": 1}
        deadline = self.submitted_at + self.timeout
        await asyncio.sleep(max(0.0, self.submitted_at + self.initial_wait - time.time()))
        async with httpx.AsyncClient(timeout=30) as client:
            while time.time() < deadline:
                delay = self.poll_interval
                try:
                    body = parse_body(await client.get(RES_URL, params=params))
                except httpx.HTTPError:
                    body = {"status": 0, "request": "CAPCHA_NOT_READY"}  # network blip: poll again
                code = str(body.get("request"))
                if body.get("status") == 1:
                    yield TriggerEvent({
                        "status": "success", "captcha_id": self.captcha_id,
                        # Enterprise and Cloudflare Challenge answers arrive in "result"
                        "token": body.get("result", body.get("request")),
                        "user_agent": body.get("user_agent"),
                        "solved_at": time.time(),
                        "elapsed_s": round(time.time() - self.submitted_at, 1),
                    })
                    return
                if code in SERVER_ERRORS:
                    delay = 10  # documented advice: retry after about 10 s
                elif code.startswith("ERROR_") or code == "IP_BANNED":
                    yield TriggerEvent({"status": "error", "code": code, "captcha_id": self.captcha_id})
                    return
                # CAPCHA_NOT_READY, or an HTML 502 page: wait and poll again
                await asyncio.sleep(min(delay, max(0.0, deadline - time.time())))
        yield TriggerEvent({"status": "error", "code": "POLL_TIMEOUT", "captcha_id": self.captcha_id})

serialize() returns the importable classpath plus the constructor kwargs, the tuple Airflow's deferrable operators documentation asks for. Airflow stores that tuple and rebuilds the trigger on whichever triggerer claims it. The same page is explicit that "your trigger must not come from a Dag bundle": use the plugins/ folder or an installed package, not the DAGs folder.

The trigger takes submitted_at instead of starting a clock inside run(). A trigger may be persisted and re-created on another triggerer, and run() then starts again from the stored kwargs. Measuring the wait and the 120-second cap from the submit time means a restarted trigger keeps polling the same task ID on the original schedule instead of granting itself another two minutes.

The response handling follows CaptchaAI's API contract. CAPCHA_NOT_READY (no T after "CAP") means poll again in 5 seconds. ERROR_SERVER_ERROR and ERROR_INTERNAL_SERVER_ERROR are transient, so wait about 10 seconds. Any other ERROR_* code, or IP_BANNED, ends the trigger with an error event. parse_body() exists because the API can return a bare error string even with json=1, and occasionally an HTML 500 or 502 page. reCAPTCHA v2/v3 Enterprise and Cloudflare Challenge put the answer in result next to a user_agent you must reuse on the target site. Other types answer in request, and CaptchaFox (beta) adds a user_agent there too, which is why the event always carries the field. Per-type wait times are in the CAPCHA_NOT_READY polling strategy guide.

Two rules for trigger code:

1. No blocking calls. requests.get() or time.sleep() inside run() freezes the triggerer's single event loop, and with it every other trigger in that process. Airflow's watchdog then logs "Triggerer's async thread was blocked". httpx.AsyncClient and asyncio.sleep() yield control instead, and httpx is already a core Airflow dependency.
2. No API key in the kwargs. Airflow has encrypted trigger kwargs at rest since 2.9.0, but a key passed in would still be copied into every trigger row, and a rotated key would never reach triggers already in flight. run() reads the Connection with await BaseHook.aget_connection() (Airflow 3.1+), the async lookup that does not stall the loop.

The operator: submit, defer, resume

Save this as plugins/captchaai_operator.py:

from __future__ import annotations

import time
from datetime import timedelta
from typing import Any

import httpx
from airflow.sdk import BaseHook, BaseOperator, Context, conf
from airflow.sdk.exceptions import AirflowException, AirflowFailException

from captchaai_trigger import RES_URL, SERVER_ERRORS, CaptchaAITrigger, parse_body

IN_URL = "https://ocr.captchaai.com/in.php"
# Worth another try (Airflow retries); every other code fails the task at once.
RETRYABLE = {"ERROR_SERVER_ERROR", "ERROR_INTERNAL_SERVER_ERROR",
             "ERROR_CAPTCHA_UNSOLVABLE", "POLL_TIMEOUT"}


class CaptchaAISolveOperator(BaseOperator):
    """Submit in execute(), wait on the triggerer, use the token in execute_complete()."""

    template_fields = ("pageurl", "sitekey")

    def __init__(self, *, pageurl: str, sitekey: str, method: str = "userrecaptcha",
                 extra_params: dict[str, Any] | None = None, conn_id: str = "captchaai",
                 max_token_age: float = 90,
                 deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
                 **kwargs: Any) -> None:
        super().__init__(**kwargs)
        self.pageurl, self.sitekey, self.method = pageurl, sitekey, method
        self.extra_params = extra_params or {}
        self.conn_id, self.max_token_age, self.deferrable = conn_id, max_token_age, deferrable

    def execute(self, context: Context) -> Any:
        key = BaseHook.get_connection(self.conn_id).password
        site_field = "googlekey" if self.method == "userrecaptcha" else "sitekey"
        data = {"key": key, "method": self.method, site_field: self.sitekey,
                "pageurl": self.pageurl, "json": 1, **self.extra_params}
        with httpx.Client(timeout=30) as client:
            body = parse_body(client.post(IN_URL, data=data))
            if body.get("status") != 1:
                self._fail_submit(str(body.get("request")), client, key)
        trigger = CaptchaAITrigger(captcha_id=str(body["request"]),
                                   submitted_at=time.time(), conn_id=self.conn_id)
        self.log.info("CaptchaAI task %s submitted", trigger.captcha_id)
        if not self.deferrable:  # local runs, or no triggerer: block this worker slot
            return self.execute_complete(context, self._poll_blocking(key, trigger))
        self.defer(trigger=trigger, method_name="execute_complete",
                   timeout=timedelta(seconds=150))

    def _fail_submit(self, code: str, client: httpx.Client, key: str) -> None:
        if code == "ERROR_ZERO_BALANCE":  # "no free threads", or no active plan
            info = parse_body(client.post(RES_URL, data={"key": key, "action": "threadsinfo"}))
            if "threads" not in info:  # the diagnosis failed too; leave it to retries
                raise AirflowException(f"ERROR_ZERO_BALANCE; threadsinfo said {info.get('request')}")
            if int(info["working_threads"]) >= int(info["threads"]):
                raise AirflowException("All CaptchaAI threads busy; retrying later")
            raise AirflowFailException("ERROR_ZERO_BALANCE with free threads: check the plan")
        if code in RETRYABLE:
            raise AirflowException(f"CaptchaAI submit failed: {code}")
        raise AirflowFailException(f"CaptchaAI rejected the task: {code}")

    def _poll_blocking(self, key: str, trigger: CaptchaAITrigger) -> dict[str, Any]:
        """The trigger's loop run synchronously, returning the same event payloads."""
        time.sleep(trigger.initial_wait)
        params = {"key": key, "action": "get", "id": trigger.captcha_id, "json": 1}
        while time.time() < trigger.submitted_at + trigger.timeout:
            try:
                body = parse_body(httpx.get(RES_URL, params=params, timeout=30))
            except httpx.HTTPError:
                body = {"status": 0, "request": "CAPCHA_NOT_READY"}
            code = str(body.get("request"))
            if body.get("status") == 1:
                return {"status": "success", "token": body.get("result", body.get("request")),
                        "user_agent": body.get("user_agent"), "solved_at": time.time(),
                        "elapsed_s": round(time.time() - trigger.submitted_at, 1)}
            if (code.startswith("ERROR_") and code not in SERVER_ERRORS) or code == "IP_BANNED":
                return {"status": "error", "code": code, "captcha_id": trigger.captcha_id}
            time.sleep(10 if code in SERVER_ERRORS else trigger.poll_interval)
        return {"status": "error", "code": "POLL_TIMEOUT", "captcha_id": trigger.captcha_id}

    def execute_complete(self, context: Context, event: dict[str, Any]) -> Any:
        if event["status"] != "success":
            exc = AirflowException if event["code"] in RETRYABLE else AirflowFailException
            raise exc(f"CaptchaAI task {event.get('captcha_id')}: {event['code']}")
        age = time.time() - event["solved_at"]
        if age > self.max_token_age:  # resumed too late: the token is probably dead
            raise AirflowException(f"Token is {age:.0f}s old; solving again")
        self.log.info("Solved in %ss; token age %.0fs", event["elapsed_s"], age)
        return self.use_token(context, event["token"], event.get("user_agent"))

    def use_token(self, context: Context, token: str, user_agent: str | None) -> Any:
        """Override: submit the protected form here. Return a result, never the token."""
        raise NotImplementedError

What the less obvious lines do:

• Parameters. The operator covers the token types that take a sitekey. method="userrecaptcha" sends it as googlekey; turnstile, friendly_captcha (beta) and captchafox (beta) send sitekey. Type flags go in extra_params: {"invisible": 1} for invisible reCAPTCHA v2, {"version": "v3", "action": "..."} for v3 (the page's action name, or verify if you can't find it), and {"enterprise": 1} for Enterprise. CaptchaFox also requires proxy and proxytype in extra_params, and use_token() has to post through that same proxy. CaptchaAI's proxy guide notes that proxy use is disabled on accounts by default until support enables it. pageurl and sitekey are template fields, so both can come from Airflow Variables.
• The key lives in the password field of the captchaai Connection, so it resolves through whatever secrets backend you run.
• The defer timeout of 150 seconds is a backstop above the trigger's own 120-second cap. If no triggerer picks the trigger up, the task fails with TaskDeferralTimeout, an AirflowException, so normal retries apply.
• The deferrable flag defaults to [operators] default_deferrable (added in 2.7), the pattern the Airflow docs use. With False the operator polls inside the worker and runs the same execute_complete().

How each CaptchaAI error ends the task

AirflowFailException fails the task without spending retries; AirflowException leaves it to the task's retries. Codes the operator doesn't recognize fail fast, because resending an unchanged bad request only burns attempts. The full list is in the CaptchaAI error codes reference.

Pool sizing: one slot per CaptchaAI thread

CaptchaAI bills per concurrent thread with unlimited solves per thread. BASIC ($15/month, 5 threads) runs five CAPTCHAs at once; STANDARD ($30/month, 15 threads) runs fifteen. An Airflow pool is how the scheduler respects that number:

# One slot per plan thread; deferred tasks keep their slot
airflow pools set captchaai 5 "CaptchaAI BASIC plan threads" --include-deferred

# The key lives in a Connection's password field
airflow connections add captchaai --conn-type generic --conn-password "$CAPTCHAAI_API_KEY"

# Optional: size the pool from the plan instead of hard-coding it
threads=$(curl -s -X POST https://ocr.captchaai.com/res.php \
  -F "key=${CAPTCHAAI_API_KEY}" -F "action=threadsinfo" | jq -r '.threads')
airflow pools set captchaai "$threads" "CaptchaAI plan threads" --include-deferred

--include-deferred (Airflow 2.7+) is the flag that matters. By default, deferred tasks don't occupy pool slots. Five solves defer, the pool reads empty, the scheduler starts five more, and every one of those deferred tasks still holds a CaptchaAI thread. The documented response to that oversubscription is ERROR_ZERO_BALANCE, which the site describes as "You don't have free threads". Saturation can also show up only as longer CAPCHA_NOT_READY polling that eats into the 120-second budget. Hitting your CaptchaAI thread limit covers both symptoms.

threadsinfo returns {"threads": "5", "working_threads": 2}: the plan cap and how many threads are busy right now. If the same key also serves scrapers or CI jobs elsewhere, size the pool to Airflow's share, not the whole plan.

The slot is taken before the submit and released when the trigger fires. By then CaptchaAI has already freed the thread, so the pool never undercounts busy threads. But the resumed task moves from deferred to scheduled, which doesn't count against a pool, so it needs a free slot again before execute_complete() can run. The scheduler sorts candidates by priority_weight, then logical date, then map index. A resumed instance usually sorts ahead of siblings that haven't started, because those belong to a later DAG run or carry a higher map index. The token ages while it waits.

State that does not survive the deferral

When the trigger fires, the task goes back through the scheduler, and a worker runs execute_complete() in a new process, possibly on another machine. Anything set on self in execute(), any open HTTP session and any cookie jar are gone. That has three consequences.

Use the token inside execute_complete(). Google states that each reCAPTCHA response token "is valid for two minutes, and can only be verified once". A Turnstile token lasts 300 seconds and is also single-use, per Cloudflare's validation docs. So use_token() opens its own session, loads the form and posts within that window. The resume is not instant: the scheduler has to notice the event, and if workers are saturated the resumed task waits in the queue like any other. max_token_age (90 seconds by default for reCAPTCHA; raise it toward 240 for Turnstile) turns a stale token into a retry with a fresh solve instead of a post the site will reject. A higher priority_weight puts the task ahead of other DAGs' work when executor slots run short. solved_at is stamped on the triggerer and checked on the worker, so keep their clocks in sync. The token was generated a few seconds before the poll saw it, which is why the default stops at 90 seconds rather than 120; the reCAPTCHA token lifecycle breaks that margin down.

Never hand the token to a downstream task through XCom. The downstream task can sit queued past expiry, and a single-use token cannot be retried. Return what the form post produced, as the example DAG does with a file path.

Carry small values explicitly. If a subclass needs something from execute(), such as a report ID it looked up before the submit, pass it with self.defer(..., kwargs={...}) and add a matching parameter to execute_complete(). Those kwargs are stored with the task instance in the metadata database, so keep them small and non-secret. That rules out session cookies, and with them anything tied to the old session, such as a CSRF value: use_token() fetches both fresh.

When not to defer

Image CAPTCHAs are the clearest case. CaptchaAI's image OCR typically answers in under 0.5 seconds, and solve.php returns the answer in the same request with no polling. The image is also usually bound to the session that loaded it, so the solve and the form post belong in one task and one cookie jar:

from __future__ import annotations

import base64

import httpx
from airflow.sdk import BaseHook, task
from airflow.sdk.exceptions import AirflowException

from captchaai_trigger import parse_body


@task
def submit_form_with_image_captcha(form_url: str, image_url: str) -> int:
    key = BaseHook.get_connection("captchaai").password
    with httpx.Client(timeout=30, follow_redirects=True) as http:
        http.get(form_url)  # the session cookie that the CAPTCHA image is tied to
        image = http.get(image_url).content
        answer = parse_body(httpx.post(
            "https://ocr.captchaai.com/solve.php", timeout=30,
            data={"key": key, "method": "base64", "json": 1,
                  "body": base64.b64encode(image).decode()}))
        if answer.get("status") != 1:
            raise AirflowException(f"CaptchaAI: {answer.get('request')}")
        resp = http.post(form_url, data={"captcha_text": answer["request"]})  # site's field name
        resp.raise_for_status()
        return resp.status_code

Images must be between 100 bytes and 100 KB, in jpg, png or gif. Deferral also isn't worth it for a DAG that solves two CAPTCHAs a day on a deployment without a triggerer; there, deferrable=False keeps the same error handling with a blocking poll. For schedulers outside Airflow, see scheduled CAPTCHA solving with cron and task schedulers.

The example DAG

Save this as dags/portal_export.py. The DAG file can live in the DAGs folder; only the trigger module has to stay out of it.

from __future__ import annotations

from datetime import datetime, timedelta

import httpx
from airflow.sdk import ObjectStoragePath, Variable, dag, task

from captchaai_operator import CaptchaAISolveOperator


class PortalExportOperator(CaptchaAISolveOperator):
    """Downloads the daily export from a partner portal we are licensed to query."""

    def use_token(self, context, token, user_agent):
        headers = {"User-Agent": user_agent} if user_agent else {}
        with httpx.Client(timeout=30, headers=headers, follow_redirects=True) as http:
            http.get(self.pageurl)  # new session, opened inside the token lifetime
            resp = http.post(self.pageurl, data={"g-recaptcha-response": token,
                                                 "date": context["ds"]})
            resp.raise_for_status()
        out = ObjectStoragePath(Variable.get("export_root")) / f"portal-{context['ds']}.csv"
        with out.open("wb") as fh:
            fh.write(resp.content)
        return str(out)  # the XCom carries the file path, never the token


@dag(schedule="@daily", start_date=datetime(2026, 9, 1), catchup=False,
     default_args={"retries": 3, "retry_delay": timedelta(seconds=10),
                   "retry_exponential_backoff": 2.0})
def portal_export():
    export = PortalExportOperator(
        task_id="download_export",
        pageurl="{{ var.value.portal_export_url }}",
        sitekey="{{ var.value.portal_export_sitekey }}",
        deferrable=True,
        pool="captchaai",
        priority_weight=10,
    )

    @task
    def check_export(path: str) -> int:
        with ObjectStoragePath(path).open("rb") as fh:
            rows = sum(1 for _ in fh) - 1
        if rows <= 0:
            raise ValueError(f"{path} has no data rows")
        return rows

    check_export(export.output)


portal_export()

Since Airflow 3.2, retry_exponential_backoff is a float multiplier: 2.0 doubles the 10-second base on each try (Airflow adds a deterministic jitter of up to the same amount again), and 0 disables it. Earlier versions take True. Set export_root to shared storage such as an s3:// URL when workers run on more than one machine. If you add execution_timeout, remember that Airflow measures it across the total runtime, deferrals included, so it must exceed 150 seconds plus the form post. The same submit-then-wait shape appears in calling CaptchaAI from a Dagster asset and, with durable timers in place of a triggerer, in wrapping CaptchaAI as a Temporal activity.

Running the triggerer

With the official docker-compose.yaml, the airflow-triggerer service (command: triggerer) already exists, and ./plugins is mounted into every Airflow container. Airflow puts the plugins folder on sys.path, so dropping both modules there makes them importable on workers and on the triggerer. After adding them, check docker compose logs airflow-triggerer for import errors.

On Kubernetes with the official Helm chart, the triggerer is enabled by default. Bake the two modules into the image the chart deploys and run a second replica for failover:

# values.yaml for the official apache-airflow Helm chart
images:
  airflow:
    repository: registry.example.com/airflow-captchaai  # official image plus plugins/
    tag: "3.3.2"
triggerer:
  enabled: true
  replicas: 2  # if one triggerer dies, the other adopts its triggers
config:
  triggerer:
    capacity: "1000"  # the default; this is where you would change it

[triggerer] capacity defaults to 1,000 triggers per process (airflow triggerer --capacity N sets it per process). Each in-flight solve is one trigger, and the pool caps in-flight solves at your plan's threads. One triggerer therefore covers every plan up to VIP-1 ($1,500/month, 1,000 threads), minus whatever other deferrable tasks share it. Only VIP-2 ($4,500/month, 3,000 threads) and VIP-3 ($7,500/month, 5,000 threads) need more triggerers, or a higher capacity, for throughput. Add the second triggerer for high availability much earlier.

Testing the trigger

The trigger is plain async code, so pytest-asyncio and respx can drive run() against a mocked res.php without a triggerer:

import importlib
import time
from types import SimpleNamespace
from unittest.mock import AsyncMock

import httpx
import pytest
import respx
from airflow.sdk import BaseHook

from captchaai_trigger import RES_URL, CaptchaAITrigger


@pytest.fixture(autouse=True)
def fake_connection(monkeypatch):
    conn = SimpleNamespace(password="0" * 32)
    monkeypatch.setattr(BaseHook, "aget_connection", AsyncMock(return_value=conn))


def make_trigger(timeout=5.0):
    return CaptchaAITrigger(captcha_id="7412", submitted_at=time.time(),
                            poll_interval=0.01, initial_wait=0, timeout=timeout)


async def first_event(trigger):
    async for event in trigger.run():
        return event.payload


@pytest.mark.asyncio
@respx.mock
async def test_ready_after_not_ready():
    respx.get(RES_URL).mock(side_effect=[
        httpx.Response(200, json={"status": 0, "request": "CAPCHA_NOT_READY"}),
        httpx.Response(200, json={"status": 1, "request": "03AGdBq2..."}),
    ])
    payload = await first_event(make_trigger())
    assert payload["status"] == "success" and payload["token"] == "03AGdBq2..."


@pytest.mark.asyncio
@respx.mock
async def test_ready_on_first_poll():
    respx.get(RES_URL).mock(return_value=httpx.Response(200, json={"status": 1, "request": "tok"}))
    assert (await first_event(make_trigger()))["token"] == "tok"


@pytest.mark.asyncio
@respx.mock
async def test_error_code_ends_the_trigger():
    respx.get(RES_URL).mock(return_value=httpx.Response(200, text="ERROR_CAPTCHA_UNSOLVABLE"))
    payload = await first_event(make_trigger())
    assert payload == {"status": "error", "code": "ERROR_CAPTCHA_UNSOLVABLE", "captcha_id": "7412"}


@pytest.mark.asyncio
@respx.mock
async def test_timeout():
    respx.get(RES_URL).mock(return_value=httpx.Response(200, json={"status": 0, "request": "CAPCHA_NOT_READY"}))
    assert (await first_event(make_trigger(timeout=0.1)))["code"] == "POLL_TIMEOUT"


def test_serialize_round_trip():
    classpath, kwargs = make_trigger().serialize()
    module, name = classpath.rsplit(".", 1)
    rebuilt = getattr(importlib.import_module(module), name)(**kwargs)
    assert rebuilt.serialize() == (classpath, kwargs)
    assert "key" not in kwargs and "password" not in kwargs

Run the unit tests, then the whole DAG:

pip install pytest pytest-asyncio respx
PYTHONPATH=plugins pytest tests/test_captchaai_trigger.py
airflow dags test portal_export 2026-09-28

airflow dags test runs the trigger in-process (its log says [DAG TEST] running trigger in line), so you don't need a triggerer on your laptop. It does make real calls to CaptchaAI and to the portal. This page's code was unit-tested against apache-airflow-core 3.3.2 with Task SDK 1.3.2, httpx 0.28.1, respx 0.23.1 and pytest-asyncio 1.4.0: the five trigger tests above, plus operator tests covering the submit, defer, ERROR_ZERO_BALANCE and blocking paths against a mocked in.php and res.php. The example DAG also ran to success under dag.test(), with CaptchaAI and the portal mocked, through the defer, the inline trigger, the resume and check_export.

On Airflow 2.x, 3.0 and 3.1

Airflow 2 reached end of life on April 22, 2026, according to the project's supported versions table; the last release was 2.11.2. On 2.7 through 2.11, the code should run with these swaps, which are taken from the 2.11 source but were not re-tested:

• BaseOperator from airflow.models.baseoperator, BaseHook from airflow.hooks.base, conf from airflow.configuration, Context from airflow.utils.context, and both exceptions from airflow.exceptions.
• dag and task from airflow.decorators, Variable from airflow.models, ObjectStoragePath from airflow.io.path (2.8+).
• In the trigger, replace the aget_connection line with await sync_to_async(BaseHook.get_connection)(self.conn_id), importing sync_to_async from asgiref.sync, which ships with Airflow. The test fixture then patches get_connection with a plain function instead of aget_connection with an AsyncMock.
• retry_exponential_backoff=True. The pool CLI line is the same.
• The defer timeout surfaces as TaskDeferralError instead of TaskDeferralTimeout. Both are AirflowExceptions, so retries still apply.

On 3.0 and 3.1, import conf from airflow.configuration and the exceptions from airflow.exceptions, and pass retry_exponential_backoff=True. 3.0 has no airflow.sdk.BaseHook, so it also needs the 2.x hook import and the sync_to_async line. These swaps come from the 3.0 and 3.1 sources and weren't re-tested either. What never changed: airflow.triggers.base, the serialize()/run() contract, and self.defer(trigger=, method_name=, kwargs=, timeout=).

Troubleshooting

FAQ

Why not submit from the trigger too, with start_from_trigger?

Airflow can start a task directly on the triggerer, and a trigger could POST to in.php itself. But the Airflow docs say the same trigger can, rarely, run in two places at once in an HA setup, and that this is expected behavior. A duplicated poll is a harmless extra GET; a duplicated submit spends a second thread. Keeping the submit in execute() also fails bad parameters immediately with a readable AirflowFailException, at the cost of about one second of worker time.

Can one DAG run solve several CAPTCHAs at once?

Yes, with dynamic task mapping: PortalExportOperator.partial(task_id="download_export", sitekey=..., deferrable=True, pool="captchaai").expand(pageurl=[...]). Each mapped instance takes its own pool slot while it runs and while it is deferred, so the pool still caps the total at your plan's threads, however many pages you map. Add context["ti"].map_index to the file name in use_token(), or every instance writes the same portal-{ds}.csv.

Size the pool, then ship the DAG

Create the captchaai Connection, set the pool to your plan's thread count with --include-deferred, and drop the two modules into plugins/. Plans and thread counts are on the CaptchaAI pricing page.