How to Solve CAPTCHA in a Prefect Flow With Retries

A Prefect captcha step is one task with retries on it, and the whole flow is about twenty lines. The detail that surprises people coming from other hosted platforms is a pleasant one: Prefect Cloud never runs your code. It schedules work, and a worker on your own machine picks it up over an outbound connection. So the solver can sit on 127.0.0.1 and the flow reaches it, which is not true on Zapier or Make.com. The two things that actually go wrong are retry behaviour and caching, and both are one argument each.

What you need

• Prefect 3 and Python 3.10 or newer, plus the CapSkip Python SDK.
• A Prefect Cloud workspace or a self-hosted Prefect server. Either works the same way for this.
• The page URL of the protected form, and its sitekey.
• CapSkip running in Local mode when the worker and the solver share a machine, or in Server mode when they do not. Both are described under connection settings.

# pip install prefect
pip install -U prefect capskip

# Point the CLI at your workspace, then start a worker
# on the machine that should run the flows.
prefect cloud login

Where your flow actually runs

This is the question that decides your whole network setup, so answer it first. Prefect’s default is a hybrid model: the orchestration layer is hosted, and the execution layer is yours. Prefect Cloud stores metadata and coordinates runs, it does not execute your code, and it needs no inbound access to your network. A worker you start inside your own infrastructure polls outward for work and submits runs locally.

The practical consequence is worth stating plainly. Run a Process worker on the same Windows machine as CapSkip and the flow calls 127.0.0.1:8080 exactly as a script on your desk would. No tunnel, no public address, no certificate. That is the opposite of the situation on a cloud-only automation platform, and it is the main reason an orchestrator is a comfortable place to put solving work.

There is one exception and you should know it before you pick a work pool. Prefect Managed work pools run your flow on Prefect’s own infrastructure rather than yours, which is convenient and removes the loopback option entirely. In that configuration the solver needs Server mode and a reachable address. Prefect publishes six static outbound addresses that Managed runs use, so you can allow exactly those through the firewall to the solver’s port and drop everything else. Managed runs also have to use an official Prefect image and are capped at 24 hours, which is another reason a Process or Docker work pool is usually the better fit here.

Step 1: put the key in a Secret block

Do not hardcode the API key in the flow file. Prefect ships a Secret block, values are encrypted at rest in the backend, and loading one is two lines. Save it once from a Python shell.

# pip install prefect
from prefect.blocks.system import Secret

secret = Secret(value="YOUR_API_KEY")
secret.save("capskip-api-key")

# Rotating it later needs overwrite, or the save is refused.
# secret.save("capskip-api-key", overwrite=True)

Step 2: write the solve task

One task, one solve. Give it retries, because a solver call is a network call and network calls fail. Prefect takes a fixed delay, a list of delays, or an exponential backoff helper, and a jitter factor spreads retries out so a batch of failures does not come back in lockstep.

The important argument is the other one. A CAPTCHA token is single use and expires within a couple of minutes, so it must never come out of a cache. Prefect 3 keeps caching off unless result persistence is on, which means most people are safe by accident. If your team has turned persistence on globally, and plenty have, a retried task can hand back the token it produced the first time and the form will reject it. Set the policy explicitly and stop thinking about it.

# pip install capskip
from prefect import task
from prefect.tasks import exponential_backoff
from prefect.cache_policies import NO_CACHE
from prefect.blocks.system import Secret
from capskip import CapSkip

# NO_CACHE matters: a token is valid once and expires fast.
@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=5),
    retry_jitter_factor=0.5,
    cache_policy=NO_CACHE,
)
def solve_recaptcha(sitekey: str, page_url: str) -> str:
    key = Secret.load("capskip-api-key").get()
    solver = CapSkip(host="127.0.0.1", port=8080, apiKey=key)
    return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

One method covers reCAPTCHA v2, Invisible, Enterprise and v3. The variants are options rather than separate calls: invisible=1, enterprise=1, or version="v3" with an action. Turnstile and GeeTest have their own methods and the same shape. Full parameter lists are in the CapSkip API documentation.

Step 3: do not retry the errors that will never pass

Blind retries waste time on failures that are deterministic. A malformed argument fails identically on every attempt, and so does a sitekey that does not belong to the page. A retry condition function gets the state and decides, and returning False ends the task immediately with the original exception.

# Retry the transient ones. Fail fast on the rest.
from capskip import ValidationException, ApiException

def worth_retrying(task, task_run, state) -> bool:
    try:
        state.result()
    except (ValidationException, ApiException):
        return False   # bad arguments or a bad sitekey
    except Exception:
        return True    # solver down, or a timeout
    return True

Pass it as retry_condition_fn on the task. NetworkException means CapSkip is not running or the host is wrong, and TimeoutException means the solve outlasted recaptchaTimeout, which defaults to 300 seconds. Both are genuinely worth another attempt. Those two, plus ValidationException and ApiException, all derive from a common base, so catching CapSkipError works if you would rather handle failures in one place.

Full working example

The whole flow. Fetch the page, pull the sitekey out of it, solve, then post the token back with the form. Each step is a task, so each one gets its own retries, its own logs and its own entry in the run graph.

# pip install prefect capskip httpx
import re
import httpx
from prefect import flow, task
from prefect.tasks import exponential_backoff
from prefect.cache_policies import NO_CACHE
from prefect.blocks.system import Secret
from capskip import CapSkip

PAGE_URL = "https://example.com/page-with-recaptcha"

@task(retries=2, retry_delay_seconds=5)
def read_sitekey(page_url: str) -> str:
    html = httpx.get(page_url, timeout=30).text
    match = re.search(r'data-sitekey=["\']([^"\']+)', html)
    if not match:
        raise RuntimeError("No data-sitekey on the page.")
    return match.group(1)

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=5),
    cache_policy=NO_CACHE,
)
def solve_recaptcha(sitekey: str, page_url: str) -> str:
    key = Secret.load("capskip-api-key").get()
    solver = CapSkip(host="127.0.0.1", port=8080, apiKey=key)
    return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

@task(retries=2, cache_policy=NO_CACHE)
def submit_form(page_url: str, token: str) -> int:
    reply = httpx.post(
        page_url,
        data={"g-recaptcha-response": token},
        timeout=30,
    )
    return reply.status_code

@flow(name="captcha-protected-submit")
def run():
    sitekey = read_sitekey(PAGE_URL)
    token = solve_recaptcha(sitekey, PAGE_URL)
    return submit_form(PAGE_URL, token)

if __name__ == "__main__":
    print(run())

Solve immediately before submitting, never in an earlier scheduled step. A token that sits in a result store for ten minutes while an upstream task finishes is a dead token by the time the form sees it.

Solving a batch without flooding the solver

Prefect runs tasks concurrently by default through a thread pool, so a hundred solves is a list comprehension over submit and no task runner configuration at all. That is more parallelism than you probably want pointed at one machine.

A global concurrency limit is the control. Create the limit once with the CLI, then occupy a slot inside the task, and any run over the cap waits rather than piling on.

# Create the limit once. Six solves in flight at a time.
prefect gcl create capskip --limit 6

# The limit is enforced across every flow run, not per flow.
from prefect import flow, task
from prefect.cache_policies import NO_CACHE
from prefect.concurrency.sync import concurrency
from prefect.futures import wait
from capskip import CapSkip

@task(retries=3, cache_policy=NO_CACHE)
def solve_one(sitekey: str, page_url: str) -> str:
    with concurrency("capskip", occupy=1):
        solver = CapSkip(host="127.0.0.1", port=8080)
        return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

@flow
def solve_many(sitekey: str, urls):
    # submit, not map: map would iterate the sitekey string.
    futures = [solve_one.submit(sitekey, u) for u in urls]
    wait(futures)

The limit applies across every flow run in the workspace, which is exactly what you want when three schedules point at the same solver. If you would rather do the fan out inside a single process, the Python SDK’s AsyncCapSkip is a genuine asyncio client and that approach is covered in the guide to solving CAPTCHAs in parallel.

Running the solver on another machine

Workers move around. A Process worker on your desk becomes a Docker worker on a server, then a Kubernetes work pool, and at some point the flow is no longer on the machine the solver runs on. Nothing in the code changes except the host.

CapSkip has two connection modes. Local binds to 127.0.0.1 and answers that device only. Server binds to your network or public IP, so a worker on a VM, a container host or a Managed work pool calls the same Windows machine over the API. A static public IP keeps that address stable. It is still your hardware and still unmetered either way, so the cost of a busy day does not change with the mode.

# Same SDK, same call. Only the host moves.
solver = CapSkip(host="10.0.0.12", port=8080, apiKey=key)

Turn on key validation once the solver listens on a network address, and give each worker its own key so one can be revoked without touching the others. Both modes are walked through in the CapSkip setup guide.

Common errors and what they mean

FAQ

The short version

Store the key in a Secret block, wrap the solve in a task with retries and NO_CACHE, and call it in the step directly before the submit. Run a worker on the machine CapSkip is on and the host stays 127.0.0.1. Add a global concurrency limit before you fan a batch out. The Python side of all this is covered on the Python CAPTCHA solver page. The same three calls exist in Node.js, PHP and C# as well, and they are listed on the CAPTCHA solving SDK page.

One thing is worth knowing before you schedule this hourly. CapSkip does captcha bypass on hardware you already own, so a flow that solves ten thousand a day costs the same as one that solves ten.