Tool CAPTCHA CrewAI: Memecahkan reCAPTCHA dan Turnstile

Untuk membangun tool CAPTCHA CrewAI, buat subclass dari BaseTool, beri skema yang menolak masukan buruk, dan panggil SDK Python CapSkip dari _run. Agar token tidak sampai ke model, setel result_as_answer, yang menjadikannya jawaban task. Tiga pengaturan CrewAI lalu menentukan apakah tool aman dibiarkan berjalan: cache tool, yang bisa memutar ulang token yang sudah mati, batas penggunaan, yang menghitung per run, dan jalur percobaan ulang, yang bisa menjalankan ulang tool yang gagal dengan sendirinya.

Apa yang Anda butuhkan

• CapSkip terpasang di PC Windows, dengan pemecahan reCAPTCHA v2 dan Turnstile diaktifkan.
• Python 3.10 sampai 3.13. crewai 1.15.35 menyatakan 3.10 atau lebih baru dan di bawah 3.14, dan capskip menyatakan 3.10 atau lebih baru. Di Python 3.14.0, pip me-resolve crewai[anthropic] menjadi crewai 0.11.2 yang lama hanya dengan sebuah peringatan, jadi perintah di bawah memakai py -3.13. Jika muncul pesan bahwa 3.13 belum terpasang, instal dulu dari python.org. Saya menjalankan Windows 11 dengan crewai 1.15.35 (rilis terbaru di PyPI pada 5 November 2026), capskip 1.6.0, anthropic 0.73.0, dan pydantic 2.12.5, di Python 3.13.7 dan 3.11.7, bukan di 3.10 atau 3.12.
• Prompt PowerShell, dan halaman milik Anda dengan kotak centang reCAPTCHA v2. Form helper di Langkah 3 mengharapkan field tersembunyi bernama _csrf dan sebuah field email, jadi ubah namanya agar sesuai dengan form Anda.
• Sebuah rute ke CapSkip. Mode Local (127.0.0.1) hanya dapat dijangkau dari PC yang sama. Mode Server membuat CapSkip menerima panggilan API dari mesin lain juga. Mode ini diatur di pengaturan koneksi pada panduan penyiapan.
• Sebuah API key Anthropic, hanya untuk run agent di Langkah 3. Tool berjalan tanpa key itu.

Cara pengujian: terhadap mock CapSkip dan halaman pendaftaran mock yang saya tulis, dengan CrewAI yang asli dan SDK yang asli. Server lokal berskrip menjawab dalam format Anthropic Messages, sehingga wrapper CrewAI yang asli mengirim body permintaan yang asli dan saya membacanya. Tidak ada model sungguhan yang dipanggil, dan id model anthropic/claude-opus-5-5 tidak pernah sampai ke API sungguhan, jadi tidak ada hal di sini yang menunjukkan bagaimana model sungguhan memilih argumen atau membaca error. Jumlah job adalah permintaan in.php yang diterima mock.

Jalankan ini di PowerShell. Perintah ini memanggil Python milik virtual environment lewat path-nya, jadi Anda tidak perlu mengaktifkannya.

# 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__)"

Setel nilai Anda sendiri di jendela yang sama. CAPTCHA_ALLOWED_HOSTS paling penting untuk keamanan: model tidak pernah menyetelnya, dan nilai kosong menolak setiap panggilan. Host saja juga mengizinkan port 80 dan 443, dan host:port hanya mengizinkan port itu.

# 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

Langkah 1: tulis tool CAPTCHA CrewAI

Simpan dua blok berikut sebagai satu file, captcha_tool.py. Blok pertama memuat pengaturan dan skema argumen, tempat CrewAI memeriksa apa yang dikirim model. Field SolveArgs.kind menerima salah satu dari dua nilai, sitekey adalah string polos pendek, dan alamat halaman harus cocok dengan host yang Anda daftarkan. Paket Python tidak membaca CAPSKIP_HOST atau variabel lain apa pun, jadi file ini meneruskan key, host, dan port secara manual.

# 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

Tambahkan blok kedua. Empat pengaturan di dalamnya adalah inti panduan ini. result_as_answer menjadikan token sebagai jawaban task. cache_function menyatakan agar tidak pernah di-cache. max_usage_count membatasi jumlah panggilan per run. Dan kegagalan dikembalikan sebagai ToolFailure, class CrewAI yang menandai sebuah panggilan sebagai gagal tanpa melempar exception. Model membaca pesan dan kodenya, dan flag ToolFailure.retryable itu hanyalah catatan untuk Anda.

# 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")

Simpan blok berikut sebagai try_tool.py. Di folder proyek, jalankan .\.venv\Scripts\python try_tool.py. Skrip ini memanggil tool secara manual, tidak memerlukan key Anthropic, dan mencetak panjang token. SDK Python tidak menyetel timeout HTTP, jadi CapSkip yang menerima koneksi dan tidak pernah menjawab akan memblokir file ini selamanya. Langkah 3 menambahkan watchdog.

# 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)

Pemeriksaan kedua adalah panggilan yang harus ditolak tool. Simpan sebagai try_hostile.py. Baris pertama yang dicetaknya menyatakan bahwa validasi argumen gagal, dan CapSkip tidak menerima task.

# 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])

Langkah 2: apa yang dilihat model, dan apa yang bisa ia lewati

Saya mengarahkan provider Anthropic milik CrewAI ke server berskrip melalui variabel ANTHROPIC_BASE_URL dan mencatat permintaannya. Permintaan itu tidak membawa field tool choice, temperature, atau thinking. Tool tiba dengan strict: true dan enum untuk field kind, tetapi sitekey telah kehilangan batas panjang dan polanya, dan alamat halaman kehilangan batas panjangnya, karena CrewAI membuang keyword itu untuk mode strict milik Anthropic. Jadi batas-batas itu hanya ada di tool Anda: CrewAI memeriksa argumen terhadap args_schema saat tool berjalan, sebelum penghitungan penggunaan atau SDK. Saya tidak bisa menguji apakah API sungguhan menerima permintaan ini.

Saya mengirim sekitar 100 set argumen langsung ke tool, dari dua daftar serangan yang terpisah, dan beberapa lewat loop agent.

Hanya ASCII yang dapat dicetak tanpa spasi yang sampai ke parser, karena urlparse membuang tab dan baris baru sebelum membaca host, dan pemeriksaan lalu mencocokkan seluruh host dan port. Penolakan dikembalikan ke model sebagai teks error, 858 karakter untuk sitekey sepanjang 1,000,000 karakter. Penolakan tidak menghabiskan max_usage_count, dan model yang terus mengirim alamat yang buruk dihentikan oleh pemeriksaan pengulangan CrewAI setelah lima panggilan, tanpa job. Di bawah kebijakan raise pada Langkah 3, penolakan justru mengakhiri run.

Langkah 3: jalankan tool CAPTCHA CrewAI dalam sebuah crew dan kirim tokennya

Agent memilih argumen dan skrip Anda mengerjakan sisanya, jadi model tidak pernah memegang token. Pertama form helper, submit_form.py. Ini Python biasa, bukan tool, dan merupakan varian dari poster di panduan tool CAPTCHA LangChain. Helper ini tidak mengikuti redirect dan menolak form yang action-nya berbeda dari halaman dalam skema, host, atau port.

# 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

Saya mencobanya terhadap poster naif yang mengikuti redirect dan memercayai action form. Baik lewat http maupun https, redirect ke nama host lain, atau action form yang mengarah ke sana, menaruh token yang diterbitkan di host itu, dan helper tidak mengirim apa pun ke sana. Helper juga menolak halaman https yang action form-nya http pada host dan port yang sama, dan redirect di dalam host yang sama, jadi tulis alamat akhir di PAGE_URL.

Sekarang giliran run_crew.py. Kebijakan raise membuat panggilan tool yang gagal menghentikan run dan sampai ke kode Anda sebagai exception, tempat ToolFailure.retryable dapat dibaca. Skrip mencoba ulang satu kali dan hanya menerima token yang diterbitkan oleh tool. Mulai dengan .\.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]"))

Pada mock, satu CAPTCHA yang tidak dapat dipecahkan lalu satu yang bagus mencetak satu percobaan gagal dan sebuah token, dan helper mengembalikan jawaban 200: 2 permintaan model dan 2 job. Model yang menjawab dengan kata-kata, dan model yang mengarang token, keduanya memicu RuntimeError. Pada setiap percobaan yang gagal, CrewAI juga mencetak dua peringatan Event pairing mismatch ke stdout dan satu baris ERROR ke stderr yang menyebut tool dan kode kegagalannya. Percobaan ulang tetap berhasil.

Watchdog menangani kasus hang. Watchdog menjalankan crew di proses anak dan menghentikannya setelah batas dalam detik, yang bisa Anda berikan setelah nama file. Default 900 mengizinkan dua percobaan, karena SDK melakukan polling job reCAPTCHA hingga 300 detik setiap kalinya. Jalankan .\.venv\Scripts\python watchdog.py sebagai ganti 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)

Terhadap mock CapSkip yang menerima job dan tidak pernah menjawab, batas 15 mengakhiri run setelah 15.1 detik dengan pesan bahwa run_crew.py dihentikan, dan exit code 1. Mock telah menerima 1 job, dan tidak ada proses Python yang tersisa menahan koneksi. Hang itu berasal dari aturan mock itu sendiri, jadi CapSkip sungguhan mungkin gagal dengan cara lain.

Dua peringatan tentang token. Dengan verbose=True, CrewAI mencetaknya ke konsol. CrewAI juga menyimpan jawaban kickoff terakhir, termasuk token, di latest_kickoff_task_outputs.db di bawah CrewAI dalam folder data aplikasi lokal Anda. Saya tidak menemukan sakelar di source untuk menghentikan penulisan itu, jadi skrip menghapus file tersebut setelah setiap run yang berhasil. Token tidak muncul di satu pun body permintaan yang dikirim ke model.

Cara CrewAI menangani kegagalan, percobaan ulang, cache_function, dan max_usage_count

Baris memakai wrapper Anthropic terhadap server berskrip, kecuali dua baris bertanda text path, yang memakai model pengganti dalam proses tanpa function calling, sehingga CrewAI memakai format teks ReAct-nya.

Pada native path, exception yang lolos sampai ke model sebagai teks Error executing tool ditambah pesan SDK, dan tidak ada yang menjalankannya ulang. ToolFailure hanya menampilkan pesan dan kode Anda kepada model. Pada text path, CrewAI menjalankan ulang tool yang melempar exception dengan sendirinya, mengubah satu kegagalan menjadi tiga job, jadi kembalikan kegagalan itu.

max_usage_count membatasi job, bukan permintaan model: dengan batas 3 model membuat 9 permintaan dan mock melihat 3 job. Penghitung dimulai ulang pada setiap run: lima kickoff dari satu crew menghasilkan lima job dengan batas 3. Tanpa batas, CrewAI menghentikan loop setelah panggilan identik kelima dan mencatat kegagalan panggilan berulang. ToolFailure.retryable tidak menentukan apa pun dalam run ini, dan di source ia hanya dicatat lalu diteruskan ke tracing.

Cache adalah yang paling berbahaya. Di 1.15.35, crew membiarkan cache tool mati kecuali Anda memberikan Crew(cache=True): dua kickoff dengan argumen yang sama menghasilkan 2 job bahkan dengan fungsi cache default CrewAI. Dengan cache aktif dan default itu, keduanya menghasilkan 1 job dan mengembalikan token yang sama dua kali, dan situs mock menolak post kedua karena sudah dipakai. Dengan never_cache di dalam tool, cache yang aktif kembali menghasilkan 2 job. ToolFailure tidak pernah di-cache. Lihat kedaluwarsa token reCAPTCHA untuk mengetahui berapa lama token asli bertahan, dan dokumentasi tool CrewAI mencakup fungsi cache dan kebijakan kegagalan.

Error yang mungkin Anda temui

FAQ

Versi singkatnya

Tool CAPTCHA CrewAI yang baik menyerahkan penolakan kepada skema, mengembalikan kegagalan alih-alih melemparnya, mematikan cache untuk tool, dan menyimpan percobaan ulang di kode Anda sendiri. Penjagaan ini bertahan di setiap pengujian di atas. Dengan sebuah bypass captcha yang berjalan di PC Windows Anda sendiri, setiap percobaan tidak memakan biaya per pemecahan, dan Halaman pemecah CAPTCHA Python mencantumkan panggilan SDK untuk tipe CAPTCHA lainnya.