CrewAI-Captcha-Tool: reCAPTCHA und Turnstile lösen

Um ein CrewAI-Captcha-Tool zu bauen, leiten Sie eine Unterklasse von BaseTool, geben Sie ihr ein Schema, das ungültige Eingaben ablehnt, und rufen Sie aus _run heraus das CapSkip-Python-SDK auf. Damit der Token nicht zum Modell gelangt, setzen Sie result_as_answer, was ihn zur Antwort der Aufgabe macht. Drei CrewAI-Einstellungen entscheiden dann, ob das Tool sicher unbeaufsichtigt laufen kann: der Tool-Cache, der einen toten Token wiedergeben kann, das Nutzungslimit, das pro Lauf zählt, und der Wiederholungspfad, der ein fehlschlagendes Tool von selbst erneut ausführen kann.

Was Sie brauchen

• CapSkip installiert auf einem Windows-PC, mit aktiviertem Lösen von reCAPTCHA v2 und Turnstile.
• Python 3.10 bis 3.13. crewai 1.15.35 deklariert 3.10 oder neuer und unter 3.14, und capskip deklariert 3.10 oder neuer. Unter Python 3.14.0 löste pip crewai[anthropic] nur mit einer Warnung zum alten crewai 0.11.2 auf, daher benutzen die Befehle unten py -3.13. Wenn es sagt, 3.13 sei nicht installiert, holen Sie es zuerst von python.org. Ich habe Windows 11 mit crewai 1.15.35 (dem neuesten Release auf PyPI am 5. November 2026), capskip 1.6.0, anthropic 0.73.0 und pydantic 2.12.5 verwendet, unter Python 3.13.7 und 3.11.7, nicht unter 3.10 oder 3.12.
• Eine PowerShell-Eingabeaufforderung und eine Seite, die Ihnen gehört, mit einer reCAPTCHA-v2-Checkbox. Der Formular-Helfer in Schritt 3 erwartet ein verstecktes Feld namens _csrf und ein E-Mail-Feld, passen Sie die Namen also an Ihr Formular an.
• Ein Weg zu CapSkip. Der Local-Modus (127.0.0.1) ist nur vom selben PC aus erreichbar. Im Server-Modus nimmt CapSkip auch API-Aufrufe von anderen Rechnern an. Den Modus stellen Sie in den Verbindungseinstellungen in der Einrichtungsanleitung ein.
• Ein Anthropic-API-Key, nur für den Agent-Lauf in Schritt 3. Das Tool läuft auch ohne.

So wurde getestet: gegen ein Mock-CapSkip und eine Mock-Anmeldeseite, die ich geschrieben habe, mit dem echten CrewAI und dem echten SDK. Ein skriptgesteuerter lokaler Server antwortete im Anthropic-Messages-Format, sodass der echte Wrapper von CrewAI echte Request-Bodys sendete und ich sie lesen konnte. Es wurde kein echtes Modell aufgerufen, und die Modell-ID anthropic/claude-opus-5-5 erreichte nie eine echte API, daher zeigt nichts hier, wie ein echtes Modell Argumente wählt oder einen Fehler liest. Die Job-Zahlen sind die Anfragen an in.php, die der Mock erhielt.

Führen Sie dies in PowerShell aus. Es ruft das Python der virtuellen Umgebung über seinen Pfad auf, daher aktivieren Sie sie nie.

# pip install "crewai[anthropic]" capskip
mkdir crewai-captcha
cd crewai-captcha
py -3.13 -m venv .venv
.\.venv\Scripts\python -m pip install "crewai[anthropic]" capskip
.\.venv\Scripts\python -c "import crewai, capskip; print(crewai.__version__)"

Setzen Sie Ihre eigenen Werte im selben Fenster. CAPTCHA_ALLOWED_HOSTS ist für die Sicherheit am wichtigsten: Das Modell setzt sie nie, und ein leerer Wert lehnt jeden Aufruf ab. Ein bloßer Host erlaubt auch die Ports 80 und 443, und host:port erlaubt genau diesen Port.

# PowerShell: set these once per window, before you run any file
$env:CAPTCHA_ALLOWED_HOSTS = "your-site.example"    # the sites the tool may solve for
$env:PAGE_URL = "https://your-site.example/signup"  # your own page
$env:SITEKEY = "YOUR_SITEKEY"                       # its data-sitekey value
$env:ANTHROPIC_API_KEY = "YOUR_PROVIDER_KEY"        # only run_crew.py needs this

Schritt 1: das CrewAI-Captcha-Tool schreiben

Speichern Sie die nächsten zwei Blöcke als eine Datei, captcha_tool.py. Der erste enthält die Einstellungen und das Argumentschema, in dem CrewAI prüft, was das Modell gesendet hat. Das Feld SolveArgs.kind nimmt einen von zwei Werten an, der sitekey ist ein kurzer einfacher String, und die Seitenadresse muss zu einem Host passen, den Sie aufgelistet haben. Das Python-Paket liest CAPSKIP_HOST oder eine andere Variable nicht, daher gibt die Datei Key, Host und Port von Hand weiter.

# captcha_tool.py, part 1: settings, arguments and the host check
import os
from typing import Literal, Type
from urllib.parse import urlparse

from capskip import ApiException, CapSkip, CapSkipError
from capskip import NetworkException, TimeoutException
from crewai.tools import BaseTool
from crewai.tools.tool_failure import ToolFailure
from pydantic import BaseModel, Field, field_validator

solver = CapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
)
MINTED = set()  # every token this tool has returned
# Set by you, never by the model. Empty means every call is refused.
RAW_HOSTS = os.getenv("CAPTCHA_ALLOWED_HOSTS", "").lower().split(",")
CAPTCHA_ALLOWED_HOSTS = {h.strip() for h in RAW_HOSTS if h.strip()}
# A bare host also allows ports 80 and 443. Write host:port to allow one port.
ALLOWED_NETLOCS = CAPTCHA_ALLOWED_HOSTS | {
    h + port for h in CAPTCHA_ALLOWED_HOSTS if ":" not in h
    for port in (":80", ":443")}


class SolveArgs(BaseModel):
    kind: Literal["recaptcha_v2", "turnstile"] = Field(
        description="Which CAPTCHA the page shows")
    sitekey: str = Field(
        min_length=10, max_length=100, pattern=r"^[A-Za-z0-9_-]+$",
        description="The data-sitekey value from the page")
    page_url: str = Field(
        max_length=2048,
        description="Full http or https address of the page with the CAPTCHA")

    @field_validator("page_url")
    @classmethod
    def host_allowed(cls, value):
        # urlparse silently drops tabs and newlines, so only printable
        # ASCII without spaces may reach it
        if not (value.isascii() and value.isprintable()) or " " in value:
            raise ValueError("page_url is not a valid address")
        try:
            parts = urlparse(value)
        except ValueError:
            raise ValueError("page_url is not a valid address")
        if (parts.scheme not in ("http", "https")
                or parts.netloc.lower() not in ALLOWED_NETLOCS):
            raise ValueError("page_url is not a site this tool may solve for")
        return value

Hängen Sie den zweiten Block an. Vier Einstellungen darin sind der Kern dieses Leitfadens. result_as_answer macht den Token zur Antwort der Aufgabe. cache_function sagt, dass nie gecacht werden soll. max_usage_count begrenzt die Aufrufe pro Lauf. Und ein Fehler kommt als ToolFailure, eine CrewAI-Klasse, die einen Aufruf als fehlgeschlagen markiert, ohne eine Exception zu werfen. Das Modell liest ihre Meldung und ihren Code, und ihr ToolFailure.retryable Flag ist ein Hinweis für Sie.

# captcha_tool.py, part 2: the tool (appended to part 1)
def never_cache(_args, _result):
    return False  # a token is single use, so never replay one


def fail(message, code, retryable=False):
    return ToolFailure(message=message, code=code, retryable=retryable)


class SolveCaptchaTool(BaseTool):
    name: str = "solve_captcha"
    description: str = (
        "Solve a reCAPTCHA v2 or Turnstile on a page and return its token. "
        "The token is single use. A reCAPTCHA token expires within about "
        "two minutes and a Turnstile token within five.")
    args_schema: Type[BaseModel] = SolveArgs
    result_as_answer: bool = True
    cache_function: object = never_cache
    max_usage_count: int = 3

    def _run(self, kind: str, sitekey: str, page_url: str):
        call = solver.recaptcha if kind == "recaptcha_v2" else solver.turnstile
        try:
            token = call(sitekey, page_url)["code"]
            MINTED.add(token)
            return token
        except TimeoutException:
            return fail("CapSkip timed out. Try once more.", "timeout", True)
        except NetworkException:
            return fail("CapSkip did not answer correctly (not running, "
                        "or an HTTP error). Do not retry.",
                        "capskip_unreachable")
        except ApiException as exc:
            if "UNSOLVABLE" in str(exc):
                return fail("This CAPTCHA could not be solved. Try once more.",
                            "unsolvable", True)
            return fail("CapSkip refused the job. Do not retry.",
                        "capskip_rejected")
        except CapSkipError:
            return fail("The arguments were refused. Do not retry.",
                        "bad_arguments")

Speichern Sie den nächsten Block als try_tool.py. Starten Sie im Projektordner .\.venv\Scripts\python try_tool.py. Es ruft das Tool von Hand auf, braucht keinen Anthropic-Key und gibt die Länge des Tokens aus. Das Python-SDK setzt kein HTTP-Timeout, ein CapSkip, das die Verbindung annimmt und nie antwortet, würde diese Datei also für immer blockieren. Schritt 3 fügt einen Watchdog hinzu.

# try_tool.py: call the tool by hand, no model needed
import os

from captcha_tool import SolveCaptchaTool

tool = SolveCaptchaTool()
result = tool.run(
    kind="recaptcha_v2",
    sitekey=os.environ["SITEKEY"],
    page_url=os.environ["PAGE_URL"],
)
print("%d characters" % len(result) if isinstance(result, str) else result)

Die zweite Prüfung ist ein Aufruf, den das Tool ablehnen muss. Speichern Sie ihn als try_hostile.py. Die erste Zeile, die sie ausgibt, besagt, dass die Argumente die Validierung nicht bestanden haben, und CapSkip erhält keine Aufgabe.

# try_hostile.py: a call the tool must refuse before CapSkip hears of it
import os

from captcha_tool import SolveCaptchaTool

try:
    SolveCaptchaTool().run(
        kind="recaptcha_v2",
        sitekey=os.environ["SITEKEY"],
        page_url="https://evil.example/signup",
    )
except ValueError as exc:
    print(str(exc).splitlines()[0])

Schritt 2: was das Modell sieht und woran es vorbeikommen kann

Ich habe den Anthropic-Provider von CrewAI über die Variable ANTHROPIC_BASE_URL auf den skriptgesteuerten Server gerichtet und die Anfrage protokolliert. Sie enthielt weder Tool-Auswahl noch Temperatur noch ein Thinking-Feld. Das Tool kam an mit strict: true und einem Enum für das Feld kind, aber der sitekey hatte seine Längenlimits und sein Muster verloren und die Seitenadresse ihr Längenlimit, weil CrewAI diese Schlüsselwörter für den Strict-Modus von Anthropic entfernt. Diese Limits leben also nur in Ihrem Tool: CrewAI prüft beim Lauf des Tools die Argumente gegen args_schema noch vor dem Nutzungszähler oder dem SDK. Ich konnte nicht testen, ob die echte API diese Anfrage akzeptiert.

Ich habe etwa 100 Argumentsätze direkt an das Tool gesendet, aus zwei getrennten Angriffslisten, und einige durch die Agent-Schleife.

Nur druckbares ASCII ohne Leerzeichen erreicht den Parser, weil urlparse Tabulatoren und Zeilenumbrüche entfernt, bevor es den Host liest, und die Prüfung dann den ganzen Host samt Port abgleicht. Eine Ablehnung geht als Fehlertext ans Modell zurück, 858 Zeichen beim sitekey mit 1.000.000 Zeichen. Eine Ablehnung verbraucht nichts von max_usage_count, und ein Modell, das weiter eine ungültige Adresse sandte, wurde von der Wiederholungsprüfung von CrewAI nach fünf Aufrufen gestoppt, ohne Job. Unter der Raise-Policy in Schritt 3 beendet eine Ablehnung stattdessen den Lauf.

Schritt 3: das CrewAI-Captcha-Tool in einer Crew ausführen und den Token senden

Der Agent wählt die Argumente, und Ihr Skript erledigt den Rest, sodass das Modell nie einen Token hält. Zuerst der Formular-Helfer, submit_form.py. Er ist einfaches Python, kein Tool, und eine Variante des Absende-Helfers im LangChain-Captcha-Tool-Leitfaden. Er folgt keinen Weiterleitungen und lehnt ein Formular ab, dessen action sich in Schema, Host oder Port von der Seite unterscheidet.

# submit_form.py: post the token to your own form, in plain Python
import re
from urllib.parse import urljoin, urlsplit

import requests

CSRF = r'name="_csrf" value="([^"]+)"'
ACTION = r'<form[^>]*action="([^"]*)"'


def same_origin(a, b):
    return urlsplit(a)[:2] == urlsplit(b)[:2]  # scheme and host:port


def post_form(page_url, token, email):
    http = requests.Session()
    # A redirect is not followed, so a page that bounces you gets no token
    page = http.get(page_url, timeout=15, allow_redirects=False)
    if page.status_code != 200:
        return "page_answered_%d" % page.status_code
    csrf = re.search(CSRF, page.text)
    action = re.search(ACTION, page.text)
    if not (csrf and action):
        return "form_not_recognised"
    target = urljoin(page_url, action.group(1))
    if not same_origin(target, page_url):
        return "form_posts_elsewhere"
    fields = {"_csrf": csrf.group(1), "email": email,
              "g-recaptcha-response": token}
    reply = http.post(target, data=fields, headers={"Referer": page_url},
                      timeout=15, allow_redirects=False)
    return "form_answered_%d" % reply.status_code

Ich habe ihn gegen einen naiven Sender getestet, der Weiterleitungen folgt und der Formular-action vertraut. Sowohl über http als auch über https brachte eine Weiterleitung auf einen anderen Hostnamen oder eine action auf einem solchen einen ausgestellten Token auf diesen Host, und der Helfer sendete dort nichts. Er lehnt auch eine https-Seite ab, deren Formular-action auf demselben Host und Port http ist, sowie eine Weiterleitung innerhalb desselben Hosts, setzen Sie also die endgültige Adresse in PAGE_URL.

Nun zu run_crew.py. Die Raise-Policy lässt einen fehlgeschlagenen Tool-Aufruf den Lauf anhalten und als Exception in Ihrem Code ankommen, wo ToolFailure.retryable lesbar ist. Das Skript wiederholt einmal und akzeptiert nur einen Token, den das Tool ausgestellt hat. Starten Sie es mit .\.venv\Scripts\python run_crew.py.

# run_crew.py: one agent, one tool, one task, and your own retry loop
import os

from crewai import Agent, Crew, LLM, Task
from crewai.tools import ToolExecutionFailedError
from crewai_core.paths import db_storage_path

from captcha_tool import MINTED, SolveCaptchaTool
from submit_form import post_form

PAGE_URL = os.environ["PAGE_URL"]
SITEKEY = os.environ["SITEKEY"]

agent = Agent(
    role="CAPTCHA solver",
    goal="Get one CAPTCHA token for the page you are given",
    backstory="You call solve_captcha once and answer with its token.",
    llm=LLM(model=os.getenv("CAPTCHA_MODEL", "anthropic/claude-opus-5-5")),
    tools=[SolveCaptchaTool()],
    max_iter=6,
)
task = Task(
    description="Solve the reCAPTCHA v2 on %s. Its sitekey is %s."
    % (PAGE_URL, SITEKEY),
    expected_output="The token and nothing else.",
    agent=agent,
)
# raise: a failed tool call stops the run and comes back to this script
crew = Crew(agents=[agent], tasks=[task], tool_failure_policy="raise")


def forget_outputs():
    # CrewAI keeps the last kickoff's answer, the token, in this file
    path = os.path.join(db_storage_path(), "latest_kickoff_task_outputs.db")
    if os.path.exists(path):
        os.remove(path)


def get_token(extra_tries=1):
    for attempt in range(extra_tries + 1):
        try:
            token = crew.kickoff().raw
            forget_outputs()
        except ToolExecutionFailedError as exc:
            failure = exc.record.failure
            print("attempt", attempt, "failed:", failure.code)
            if not failure.retryable:
                raise
            continue
        if token not in MINTED:  # the agent answered in words instead
            raise RuntimeError("no token in the answer")
        return token
    raise RuntimeError("no token after %d tries" % (extra_tries + 1))


if __name__ == "__main__":
    token = get_token()
    print("token:", len(token), "characters")
    print("form:", post_form(PAGE_URL, token, "[email protected]"))

Auf dem Mock gab ein unlösbares Captcha und danach ein gutes einen einzelnen fehlgeschlagenen Versuch und einen Token aus, und der Helfer gab eine 200-Antwort zurück: 2 Modellanfragen und 2 Jobs. Ein Modell, das in Worten antwortete, und eines, das einen Token erfand, führten beide zur RuntimeError. Bei jedem fehlgeschlagenen Versuch gibt CrewAI außerdem zwei Warnungen Event pairing mismatch auf stdout und eine ERROR-Zeile auf stderr aus, die das Tool und den Fehlercode nennt. Der Wiederholungsversuch gelingt trotzdem.

Der Watchdog deckt das Hängen ab. Er führt die Crew in einem Kindprozess aus und stoppt sie nach einem Limit in Sekunden, das Sie nach dem Dateinamen übergeben können. Der Standardwert 900 erlaubt zwei Versuche, weil das SDK einen reCAPTCHA-Job jeweils bis zu 300 Sekunden pollt. Starten Sie .\.venv\Scripts\python watchdog.py statt der Crew.

# watchdog.py: stop a crew that hangs on a CapSkip that never answers
import subprocess
import sys

limit = int(sys.argv[1]) if len(sys.argv) > 1 else 900
try:
    run = subprocess.run([sys.executable, "run_crew.py"], timeout=limit)
    sys.exit(run.returncode)
except subprocess.TimeoutExpired:
    sys.exit("run_crew.py was stopped after %d seconds" % limit)

Gegen ein Mock-CapSkip, das den Job annimmt und nie antwortet, beendete ein Limit von 15 den Lauf nach 15,1 Sekunden mit der Meldung, dass run_crew.py gestoppt wurde, und Exit-Code 1. Der Mock hatte 1 Job erhalten, und kein Python-Prozess hielt die Verbindung noch fest. Das Hängen ist die eigene Regel des Mocks, ein echtes CapSkip kann also anders scheitern.

Zwei Warnungen zum Token. Mit verbose=True gab CrewAI ihn auf der Konsole aus. CrewAI speichert außerdem die Antwort des letzten kickoff, Token inklusive, in latest_kickoff_task_outputs.db unter CrewAI in Ihrem lokalen App-Data-Ordner. Ich habe im Quelltext keinen Schalter gefunden, der dieses Schreiben verhindert, daher löscht das Skript die Datei nach jedem guten Lauf. Der Token tauchte in keinem der Request-Bodys auf, die an das Modell gesendet wurden.

Wie CrewAI mit Fehlern, Wiederholungen, cache_function und max_usage_count umgeht

Die Zeilen verwenden den Anthropic-Wrapper gegen den skriptgesteuerten Server, außer den zwei als Textpfad markierten, die ein Stand-in-Modell im Prozess ohne Function Calling verwendeten, sodass CrewAI sein ReAct-Textformat benutzte.

Auf dem nativen Pfad erreicht eine entkommende Exception das Modell als Text Error executing tool plus die SDK-Meldung, und nichts führt sie erneut aus. Eine ToolFailure zeigt dem Modell nur Ihre Meldung und Ihren Code. Auf dem Textpfad führte CrewAI ein Tool, das eine Exception wirft, von selbst erneut aus und machte aus einem Fehler drei Jobs, geben Sie den Fehler also zurück.

max_usage_count begrenzt Jobs, nicht Modellanfragen: Bei einem Limit von 3 stellte das Modell 9 Anfragen, und der Mock sah 3 Jobs. Der Zähler startet bei jedem Lauf neu: Fünf kickoffs einer Crew ergaben fünf Jobs bei einem Limit von 3. Ohne Limit stoppte CrewAI die Schleife nach dem fünften identischen Aufruf und verzeichnete einen Fehler wegen eines wiederholten Aufrufs. ToolFailure.retryable hat in diesen Läufen nichts entschieden, und der Quelltext zeichnet es nur auf und reicht es an das Tracing weiter.

Das Caching ist die gefährliche Sache. In 1.15.35 lässt eine Crew ihren Tool-Cache aus, sofern Sie nicht Crew(cache=True)übergeben: Zwei kickoffs mit denselben Argumenten ergaben 2 Jobs, selbst mit der Standard-Cache-Funktion von CrewAI. Mit eingeschaltetem Cache und diesem Standard ergaben sie 1 Job und gaben denselben Token zweimal zurück, und die Mock-Website lehnte das zweite Absenden als bereits verwendet ab. Mit never_cache im Tool ergab der eingeschaltete Cache wieder 2 Jobs. Eine ToolFailure wird nie gecacht. Siehe Ablauf des reCAPTCHA-Tokens zur Gültigkeitsdauer eines echten Tokens. Die CrewAI-Dokumentation zu Tools erklärt die Cache-Funktion und die Fehlerrichtlinien.

Fehler, die auftreten können

FAQ

Die Kurzfassung

Ein gutes CrewAI-Captcha-Tool lässt das Schema die Ablehnung übernehmen, gibt Fehler zurück, statt sie zu werfen, schaltet den Cache für das Tool aus und hält Wiederholungen in Ihrem eigenen Code. Diese Schutzmaßnahmen haben in allen Tests oben gehalten. Läuft eine Captcha-Umgehung auf Ihrem eigenen Windows-PC, kostet jeder Versuch nichts pro Lösung, und die Python-Captcha-Solver-Seite listet die SDK-Aufrufe für die anderen Captcha-Typen auf.