How to Solve CAPTCHA in a Dagster Pipeline (Ops and Assets)

A Dagster captcha step is one op with a retry policy on it, and the pipeline around it is about thirty lines. The part that is specific to Dagster, and that no other orchestrator will warn you about, is what happens to the value the op returns. Dagster persists op and asset outputs through an IO manager, and the default one pickles them to disk. A CAPTCHA token is a single use credential with a two minute life, so that is the last place it should end up.

What you need

• Dagster 1.9 or newer and Python 3.10 or newer, plus the CapSkip Python SDK.
• Dagster open source, or a Dagster+ deployment. The code is the same either way.
• The page URL of the protected form, and its sitekey.
• CapSkip running in Local mode when the code and the solver share a machine, or in Server mode when they do not. Both are described under connection settings.

# pip install capskip
pip install dagster dagster-webserver capskip

# Run the UI locally while you build the job.
dagster dev

Where your code actually runs

Answer this before you pick a host string, because it decides the whole network setup. Dagster open source runs entirely on your own machines, so nothing here needs discussing. Dagster+ has two shapes and they are opposites for this purpose.

Hybrid is the one that behaves like you would hope. You run an agent inside your own infrastructure and it connects out to the control plane. Dagster+ has no ingress into your network, does not see your code and does not touch your data, so a run launched from the cloud UI still executes on hardware you own. Put the agent on the same machine as CapSkip and the op calls 127.0.0.1:8080 exactly as a script on your desk would. No tunnel and no public address.

Serverless is the exception. There, your code executes in Dagster’s environment rather than yours, and loopback stops meaning anything useful: it points at Dagster’s container, where nothing is listening. In that configuration the solver needs Server mode and an address the run can reach. That is not a downgrade: it is the same solver on the same hardware with a different bind address, and it still bills nothing per solve.

Step 1: wrap the solver in a resource

Dagster’s idiom for anything external is a resource, not a module level client. Subclass ConfigurableResource, declare the connection fields, and the UI gets a launchpad entry for them, the config is validated before the run starts, and tests can swap the whole thing for a fake. Read the key from the environment rather than hardcoding it.

# pip install capskip
import dagster as dg
from capskip import CapSkip

class CapSkipResource(dg.ConfigurableResource):
    host: str = "127.0.0.1"
    port: int = 8080
    api_key: str = "capskip"

    def solve_recaptcha(self, sitekey: str, page_url: str) -> str:
        solver = CapSkip(host=self.host, port=self.port, apiKey=self.api_key)
        return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

One method covers reCAPTCHA v2, Invisible, Enterprise and v3, because the variants are keyword options rather than separate calls: invisible set to 1, enterprise set to 1, or version set to v3 with an action. Turnstile and GeeTest have their own methods and the same shape. The full parameter list is in the CapSkip API documentation.

Step 2: one op, with a retry policy

A solver call is a network call, and network calls fail. Dagster’s RetryPolicy is declarative and goes on the decorator: a maximum count, a base delay, a backoff curve and a jitter mode. Exponential backoff with plus or minus jitter is the sensible default, because it spreads a batch of simultaneous failures out instead of bringing them all back at once.

# The policy lives on the decorator, not in the body.
@dg.op(
    retry_policy=dg.RetryPolicy(
        max_retries=3,
        delay=5,
        backoff=dg.Backoff.EXPONENTIAL,
        jitter=dg.Jitter.PLUS_MINUS,
    )
)
def solve_and_submit(context, sitekey: str, capskip: CapSkipResource) -> int:
    token = capskip.solve_recaptcha(sitekey, PAGE_URL)
    context.log.info("Solved, submitting immediately.")
    return post_form(PAGE_URL, token)

Note what that op returns: a status code, not a token. That is the whole point of the next section.

Step 3: never let the token cross an op boundary

This is the Dagster specific trap and it is easy to walk into, because the obvious design is one op that solves and a second that submits. Dagster does not pass values between ops in memory. Every output goes through an IO manager, and the default is the filesystem one, which stores outputs as pickle files on local disk. A token returned from an op is therefore written to a file, read back by the next op, and left sitting there afterwards.

That is wrong twice over. It writes a live credential to disk, where nothing cleans it up. And it puts a storage round trip between the solve and the submit, which is exactly the delay a two minute expiry cannot afford. On a re-execution it gets worse: Dagster can load a previous run’s stored output instead of recomputing, and then the form is handed a token that expired yesterday.

The fix is not a clever IO manager. It is to keep the solve and the submit inside the same op, so the token lives in a local variable and never becomes an output at all. Everything upstream and downstream can stay as separate ops or assets. Only the pair that has to share the token gets merged.

# The token is a local variable. It is never an op output,
# so no IO manager ever sees it and nothing is written to disk.
@dg.op(retry_policy=dg.RetryPolicy(max_retries=3, delay=5))
def submit_protected_form(sitekey: str, capskip: CapSkipResource) -> int:
    token = capskip.solve_recaptcha(sitekey, PAGE_URL)
    return post_form(PAGE_URL, token)

If you are working in assets rather than ops, the same rule applies with one extra note: a materialised asset is a permanent record with a history in the UI, and a single use secret has no business being one. Model the solve as an op inside a graph backed asset, and let the asset materialise the result of the submission.

Full working example

A complete job. Fetch the page, pull the sitekey out of it, then solve and submit in one op. Two ops, one resource, one Definitions object.

# pip install dagster capskip httpx
import re
import httpx
import dagster as dg
from capskip import CapSkip

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

class CapSkipResource(dg.ConfigurableResource):
    host: str = "127.0.0.1"
    port: int = 8080
    api_key: str = "capskip"

    def solve_recaptcha(self, sitekey: str, page_url: str) -> str:
        solver = CapSkip(host=self.host, port=self.port, apiKey=self.api_key)
        return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

@dg.op(retry_policy=dg.RetryPolicy(max_retries=2, delay=3))
def read_sitekey() -> str:
    html = httpx.get(PAGE_URL, timeout=30).text
    match = re.search(r'data-sitekey=["\']([^"\']+)', html)
    if not match:
        raise dg.Failure("No data-sitekey on the page.")
    return match.group(1)

@dg.op(
    pool="capskip",
    retry_policy=dg.RetryPolicy(
        max_retries=3, delay=5, backoff=dg.Backoff.EXPONENTIAL
    ),
)
def submit_protected_form(sitekey: str, capskip: CapSkipResource) -> int:
    token = capskip.solve_recaptcha(sitekey, PAGE_URL)
    reply = httpx.post(
        PAGE_URL,
        data={"g-recaptcha-response": token},
        timeout=30,
    )
    return reply.status_code

@dg.job
def captcha_protected_submit():
    submit_protected_form(read_sitekey())

defs = dg.Definitions(
    jobs=[captcha_protected_submit],
    resources={
        "capskip": CapSkipResource(api_key=dg.EnvVar("CAPSKIP_API_KEY")),
    },
)

EnvVar defers the lookup to run time rather than baking the value into the definition, so the key is never in your repository and never in a serialised snapshot.

Do not retry the failures that will never pass

A declarative policy retries everything, which wastes three attempts and a minute of backoff on a malformed argument that will fail identically every time. Dagster’s escape hatch is to raise the retry yourself: catch the exceptions that are worth another go, ask for a retry explicitly, and let the rest propagate and fail the run immediately.

# Retry the transient ones. Fail fast on the rest.
from capskip import NetworkException, TimeoutException

@dg.op
def solve_with_judgement(sitekey: str, capskip: CapSkipResource) -> int:
    try:
        token = capskip.solve_recaptcha(sitekey, PAGE_URL)
    except (NetworkException, TimeoutException) as err:
        raise dg.RetryRequested(max_retries=3, seconds_to_wait=10) from err
    return post_form(PAGE_URL, token)

NetworkException means CapSkip is not running or the host is wrong, and TimeoutException means the solve outlasted recaptchaTimeout, which defaults to 300 seconds. Both deserve another attempt. ValidationException means a bad argument and ApiException means the API rejected the request, and neither improves on a second try. All four derive from a common base called CapSkipError, so you can catch that one instead when you would rather handle everything in one place.

Throttling a batch with a concurrency pool

Fan a partitioned job out across two hundred URLs and Dagster will happily try to run all of them, which is more load than you want pointed at one solver. Concurrency pools are the control. Tag the op with a pool name, set the limit once, and anything over the cap queues instead of piling on.

# Six solves in flight at a time, across every run.
dagster instance concurrency set capskip 6

The pool argument is already on the op in the example above. Limits apply across all runs rather than per run, which is exactly what you want when three schedules point at the same machine. If you would rather do the fan out inside one process instead of one op per URL, the Python SDK ships a genuine asyncio client and that approach is covered in the guide to solving CAPTCHAs in parallel.

Running the solver on another machine

Code location containers, Kubernetes agents and Serverless runs all move your op away from your desk. Nothing in the code changes except the host, and the resource already reads it from config.

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 container, a VM or a Serverless run reaches the same Windows machine over the API. A static public IP keeps the 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 resource, same call. Only the host moves.
resources={
    "capskip": CapSkipResource(
        host=dg.EnvVar("CAPSKIP_HOST"),
        api_key=dg.EnvVar("CAPSKIP_API_KEY"),
    ),
}

Turn on key validation once the solver listens on a network address, and give each code location 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

Wrap the solver in a ConfigurableResource and read the key from EnvVar. Give the op a RetryPolicy with exponential backoff, and raise RetryRequested yourself when you want to skip retries that cannot succeed. Above all, keep the solve and the submit in the same op, because Dagster persists op outputs by default and a token is not something to persist.

The Python side of all this is covered on the Python CAPTCHA solver page. The details of that CAPTCHA type live on the reCAPTCHA v2 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 to know before you schedule this against a large backfill. CapSkip is a captcha solver that runs on hardware you already own, so a job that solves fifty thousand rows costs exactly what a job that solves fifty does.