CrewAI CAPTCHA Aracı: reCAPTCHA ve Turnstile Çözme

Bir CrewAI CAPTCHA aracı oluşturmak için BaseTool sınıfından türetin, ona kötü girdiyi reddeden bir şema verin ve CapSkip Python SDK’sını _run içinden çağırın. Token’ı modelden uzak tutmak için result_as_answer ayarlayın; bu, onu görevin yanıtı yapar. Üç CrewAI ayarı aracın çalışır bırakılmasının güvenli olup olmadığına karar verir: ölü bir token’ı yeniden oynatabilen araç önbelleği, çalıştırma başına sayan kullanım sınırı ve başarısız bir aracı kendiliğinden yeniden çalıştırabilen yeniden deneme yolu.

Neye ihtiyacınız var

• Bir Windows bilgisayarına kurulu, reCAPTCHA v2 ve Turnstile çözmesi açık CapSkip.
• Python 3.10 ile 3.13. crewai 1.15.35, 3.10 veya üzeri ve 3.14 altı olduğunu bildirir; capskip ise 3.10 veya üzerini bildirir. Python 3.14.0 üzerinde pip, crewai[anthropic] paketini yalnızca bir uyarıyla eski crewai 0.11.2 sürümüne çözümledi, bu yüzden aşağıdaki komutlar py -3.13 kullanır. 3.13 kurulu değil diyorsa önce python.org adresinden edinin. Windows 11 üzerinde, crewai 1.15.35 (5 Kasım 2026 tarihinde PyPI’deki en yeni sürüm), capskip 1.6.0, anthropic 0.73.0 ve pydantic 2.12.5 ile, Python 3.13.7 ve 3.11.7 üzerinde çalıştırdım; 3.10 ya da 3.12 üzerinde değil.
• Bir PowerShell istemi ve reCAPTCHA v2 onay kutusu bulunan, sahibi olduğunuz bir sayfa. 3. Adım’daki form yardımcısı, adı _csrf olan gizli bir alan ve bir e-posta alanı bekler; bu yüzden adları kendi formunuza uyacak şekilde değiştirin.
• CapSkip için bir yol. Local modu (127.0.0.1) yalnızca aynı bilgisayardan erişilebilir. Server modu, CapSkip’in diğer makinelerden de API çağrısı almasını sağlar. Modu bağlantı ayarları bölümünden ayarlarsınız; kurulum kılavuzunda anlatılıyor.
• Yalnızca 3. Adım’daki ajan çalıştırması için bir Anthropic API anahtarı. Araç onsuz çalışır.

Bu nasıl test edildi: benim yazdığım bir mock CapSkip ve bir mock kayıt sayfasına karşı, gerçek CrewAI ve gerçek SDK ile. Senaryolu bir yerel sunucu Anthropic Messages biçiminde yanıt verdi; böylece CrewAI’ın gerçek sarmalayıcısı gerçek istek gövdeleri gönderdi ve onları okudum. Hiçbir gerçek model çağrılmadı ve model kimliği anthropic/claude-opus-5-5 hiçbir zaman gerçek bir API’ye ulaşmadı; bu yüzden buradaki hiçbir şey gerçek bir modelin argümanları nasıl seçtiğini ya da bir hatayı nasıl okuduğunu göstermez. İş sayıları, mock’un aldığı in.php isteklerinin sayısıdır.

Bunu PowerShell’de çalıştırın. Sanal ortamın Python’unu yoluyla çağırır, bu yüzden onu hiç etkinleştirmezsiniz.

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

Kendi değerlerinizi aynı pencerede ayarlayın. CAPTCHA_ALLOWED_HOSTS güvenlik açısından en çok önem taşır: model onu asla ayarlamaz ve boş bir değer her çağrıyı reddeder. Yalın bir host 80 ve 443 portlarına da izin verir, host:port ise tam olarak o porta izin verir.

# 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

1. Adım: CrewAI CAPTCHA aracını yazın

Sonraki iki bloğu tek bir dosya olarak kaydedin; adı captcha_tool.py olsun. İlki ayarları ve argüman şemasını içerir; CrewAI modelin gönderdiğini burada denetler. SolveArgs.kind alanı iki değerden birini alır, sitekey kısa ve düz bir dizedir ve sayfa adresi listelediğiniz bir host ile eşleşmelidir. Python paketi CAPSKIP_HOST ya da başka hiçbir değişkeni okumaz; bu yüzden dosya anahtarı, host’u ve portu elle geçirir.

# 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

İkinci bloğu ekleyin. İçindeki dört ayar bu rehberin asıl konusudur. result_as_answer token’ı görevin yanıtı yapar. cache_function asla önbelleğe almamasını söyler. max_usage_count çalıştırma başına çağrıları sınırlar. Ve bir başarısızlık ToolFailure olarak geri döner; bu, bir çağrıyı fırlatmadan başarısız olarak işaretleyen bir CrewAI sınıfıdır. Model onun iletisini ve kodunu okur, ToolFailure.retryable bayrağı ise sizin için bir nottur.

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

Sonraki bloğu try_tool.py olarak kaydedin. Proje klasöründe şunu çalıştırın: .\.venv\Scripts\python try_tool.py komutunu. Aracı elle çağırır, Anthropic anahtarı gerektirmez ve token uzunluğunu yazdırır. Python SDK hiçbir HTTP zaman aşımı ayarlamaz, bu yüzden bağlantıyı alıp hiç yanıt vermeyen bir CapSkip bu dosyayı sonsuza dek bloke ederdi. 3. Adım bir watchdog ekler.

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

İkinci kontrol, aracın reddetmesi gereken bir çağrıdır. Onu try_hostile.py olarak kaydedin. Yazdırdığı ilk satır argümanların doğrulamayı geçemediğini söyler ve CapSkip hiç görev almaz.

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

2. Adım: model neyi görür ve neyi aşabilir

CrewAI’ın Anthropic sağlayıcısını senaryolu sunucuya ANTHROPIC_BASE_URL değişkeni aracılığıyla yönelttim ve isteği günlüğe kaydettim. İstek bir araç seçimi, temperature ya da thinking alanı taşımıyordu. Araç strict: true ve kind alanı için bir enum ile geldi, ama sitekey uzunluk sınırlarını ve desenini, sayfa adresi de uzunluk sınırını yitirmişti, çünkü CrewAI bu anahtar sözcükleri Anthropic’in strict modu için çıkarır. Bu yüzden bu sınırlar yalnızca sizin aracınızda yaşar: araç çalışırken, kullanım sayacından ya da SDK’dan önce, CrewAI argümanları args_schema ile denetler. Gerçek API’nin bu isteği kabul edip etmediğini test edemedim.

Yaklaşık 100 argüman kümesini, iki ayrı saldırı listesinden, doğrudan araca ve birkaçını ajan döngüsünden gönderdim.

Ayrıştırıcıya yalnızca boşluksuz yazdırılabilir ASCII ulaşır, çünkü urlparse host’u okumadan önce sekmeleri ve satır sonlarını atar ve kontrol sonra host ile portun tamamını eşleştirir. Bir ret modele hata metni olarak geri gider; 1.000.000 karakterlik sitekey için 858 karakter. Bir ret max_usage_count değerini tüketmez; kötü bir adres göndermeyi sürdüren bir model, CrewAI’ın tekrar denetimiyle beş çağrıdan sonra, hiçbir iş olmadan durduruldu. 3. Adım’daki raise politikasında bir ret bunun yerine çalıştırmayı bitirir.

3. Adım: CrewAI CAPTCHA aracını bir crew içinde çalıştırın ve token’ı gönderin

Argümanları ajan seçer ve gerisini sizin script’iniz yapar, böylece model hiçbir zaman bir token tutmaz. Önce form yardımcısı, submit_form.py. Düz Python’dur, bir araç değildir ve LangChain CAPTCHA araç rehberi içindeki göndericinin bir varyantıdır. Yönlendirmeleri izlemez ve action değeri sayfadan şema, host ya da port bakımından farklı olan bir formu reddeder.

# 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

Yönlendirmeleri izleyen ve form action değerine güvenen saf bir göndericiye karşı denedim. Hem http hem https üzerinde, başka bir host adına yönlendirme ya da böyle bir host’a işaret eden form action değeri, verilmiş bir token’ı o host’a koydu; yardımcı oraya hiçbir şey göndermedi. Ayrıca action değeri aynı host ve portta http olan bir https sayfasını ve aynı host içindeki bir yönlendirmeyi de reddeder; bu yüzden son adresi şuraya yazın: PAGE_URL.

Şimdi run_crew.py dosyası. Raise politikası, başarısız bir araç çağrısının çalıştırmayı durdurmasını ve kodunuza bir istisna olarak ulaşmasını sağlar; orada ToolFailure.retryable okunabilir. Script bir kez yeniden dener ve yalnızca aracın ürettiği bir token’ı kabul eder. Şununla başlatın: .\.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]"))

Mock üzerinde, bir çözülemeyen CAPTCHA ve ardından iyi bir tane, tek bir başarısız deneme ve bir token yazdırdı; yardımcı 200 yanıtı döndürdü: 2 model isteği ve 2 iş. Sözcüklerle yanıt veren bir model ve bir token uyduran bir model, ikisi de RuntimeError hatasına takıldı. Her başarısız denemede CrewAI ayrıca stdout’a iki Event pairing mismatch uyarısı ve stderr’e araç adını ve hata kodunu veren bir ERROR satırı yazdırır. Yeniden deneme yine de başarılı olur.

Watchdog takılmayı kapsar. crew’i bir alt süreçte çalıştırır ve saniye cinsinden bir sınırdan sonra durdurur; sınırı dosya adından sonra verebilirsiniz. 900 varsayılanı iki denemeye izin verir, çünkü SDK bir reCAPTCHA işini her seferinde 300 saniyeye kadar yoklar. Crew yerine .\.venv\Scripts\python watchdog.py çalıştırın.

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

CapSkip işi kabul edip hiç yanıt vermeyen bir mock’a karşı 15 saniyelik bir sınır, çalıştırmayı 15,1 saniye sonra bitirdi; mesajda run_crew.py dosyasının durdurulduğu yazıyordu ve çıkış kodu 1 idi. Mock 1 iş almıştı ve bağlantıyı tutan hiçbir Python süreci kalmadı. Takılma, mock’un kendi kuralıdır; bu yüzden gerçek bir CapSkip farklı şekilde başarısız olabilir.

Token hakkında iki uyarı. Birincisi, verbose=True ayarlandığında CrewAI token’ı konsola yazdırdı. CrewAI ayrıca son kickoff’un yanıtını, token dahil, yerel uygulama verisi klasörünüzde CrewAI altındaki latest_kickoff_task_outputs.db dosyasına kaydeder. Bu yazmayı durduracak bir anahtarı kaynakta bulamadım, bu yüzden script her başarılı çalıştırmadan sonra dosyayı siler. Token, modele gönderilen istek gövdelerinin hiçbirinde görünmedi.

CrewAI başarısızlıkları, yeniden denemeleri, cache_function ve max_usage_count değerlerini nasıl ele alır

Satırlar, senaryolu sunucuya karşı Anthropic sarmalayıcısını kullanır; fonksiyon çağırma desteği olmayan, süreç içi bir yedek model kullanan ve metin yolu diye işaretlenen iki satır hariç. Bunlarda CrewAI ReAct metin biçimini kullandı.

Native yolda kaçan bir istisna modele Error executing tool metni artı SDK iletisi olarak ulaşır ve hiçbir şey onu yeniden çalıştırmaz. Bir ToolFailure modele yalnızca sizin iletinizi ve kodunuzu gösterir. Metin yolunda CrewAI, hata fırlatan bir aracı kendiliğinden yeniden çalıştırdı ve bir başarısızlığı üç işe dönüştürdü; bu yüzden başarısızlığı döndürün.

max_usage_count model isteklerini değil, işleri sınırlar: 3 sınırıyla model 9 istek yaptı ve mock 3 iş gördü. Sayaç her çalıştırmada sıfırlanır: bir crew’un beş kickoff’u, 3 sınırı altında beş iş verdi. Sınır yokken CrewAI, beşinci özdeş çağrıdan sonra döngüyü durdurdu ve tekrarlanan çağrı hatası kaydetti. ToolFailure.retryable bu çalıştırmalarda hiçbir şeye karar vermedi; kaynak onu yalnızca kaydeder ve izlemeye (tracing) iletir.

Önbellek tehlikeli olandır. 1.15.35 sürümünde bir crew, araç önbelleğini kapalı bırakır; açmak için şunu verin: Crew(cache=True): aynı argümanlarla iki kickoff, CrewAI’ın varsayılan önbellek fonksiyonuyla bile 2 iş yaptı. Önbellek açıkken ve bu varsayılanla 1 iş yaptılar ve aynı token’ı iki kez döndürdüler; mock site ikinci gönderimi zaten kullanılmış diye reddetti. Araçta never_cache varken, önbellek açıkken yine 2 iş oluştu. Bir ToolFailure asla önbelleğe alınmaz. Gerçek bir token’ın ne kadar süre geçerli kaldığını reCAPTCHA token süresinin dolması sayfasında bulabilirsiniz. CrewAI’ın araç belgeleri önbellek fonksiyonunu ve hata politikalarını kapsar.

Karşılaşabileceğiniz hatalar

FAQ

Kısa özet

İyi bir CrewAI CAPTCHA aracı reddetme işini şemaya yaptırır, hataları fırlatmak yerine döndürür, araç için önbelleği kapatır ve yeniden denemeleri kendi kodunuzda tutar. Bu korumalar yukarıdaki her testte tuttu. Kendi Windows bilgisayarınızda çalışan bir captcha atlatma aracı ile her deneme, çözüm başına hiçbir şeye mal olmaz. Python CAPTCHA çözücü sayfası öteki CAPTCHA türleri için SDK çağrılarını listeler.