CrewAI कैप्चा tool: reCAPTCHA और Turnstile हल करें

CrewAI कैप्चा tool बनाने के लिए BaseTool को subclass करें, उसे ऐसा schema दें जो गलत input को ठुकरा दे, और CapSkip Python SDK को _run से call करें। token को model से दूर रखने के लिए result_as_answer सेट करें, जो उसे task का जवाब बना देता है। फिर CrewAI की तीन settings तय करती हैं कि tool को चालू छोड़ना सुरक्षित है या नहीं: tool cache, जो मरा हुआ token दोबारा चला सकता है, usage limit, जो हर run के हिसाब से गिनती है, और retry path, जो नाकाम tool को अपने आप दोबारा चला सकता है।

आपको क्या चाहिए

• Windows PC पर इंस्टॉल किया हुआ CapSkip, जिसमें reCAPTCHA v2 और Turnstile हल करना चालू हो।
• Python 3.10 से 3.13। crewai 1.15.35 Python 3.10 या बाद का और 3.14 से नीचे का घोषित करता है, और capskip 3.10 या बाद का। Python 3.14.0 पर pip ने crewai[anthropic] को सिर्फ़ एक warning के साथ पुराने crewai 0.11.2 पर resolve कर दिया, इसलिए नीचे के commands में py -3.13 इस्तेमाल हुआ है। अगर वह कहे कि 3.13 इंस्टॉल नहीं है, तो पहले python.org से लें। मैंने Windows 11 पर crewai 1.15.35 (5 नवंबर 2026 को PyPI पर सबसे नया रिलीज़), capskip 1.6.0, anthropic 0.73.0 और pydantic 2.12.5 के साथ Python 3.13.7 और 3.11.7 पर चलाया, 3.10 या 3.12 पर नहीं।
• एक PowerShell prompt, और आपका अपना पेज जिस पर reCAPTCHA v2 checkbox हो। चरण 3 का form helper _csrf नाम का hidden field और एक email field मानकर चलता है, इसलिए नाम अपने form से मिला लें।
• CapSkip तक पहुँचने का रास्ता। Local mode (127.0.0.1) सिर्फ़ उसी PC से पहुँचा जा सकता है। Server mode CapSkip को दूसरी मशीनों से भी API calls लेने देता है। mode सेटअप गाइड की कनेक्शन सेटिंग्स में सेट होता है।
• एक Anthropic API key, सिर्फ़ चरण 3 के agent run के लिए। tool इसके बिना चलता है।

इसे कैसे टेस्ट किया गया: मेरे लिखे mock CapSkip और mock sign-up पेज के सामने, असली CrewAI और असली SDK के साथ। एक scripted local server ने Anthropic Messages format में जवाब दिया, इसलिए CrewAI के असली wrapper ने असली request bodies भेजीं और मैंने उन्हें पढ़ा। किसी असली model को call नहीं किया गया, और model id anthropic/claude-opus-5-5 कभी किसी असली API तक नहीं पहुँचा, इसलिए यहाँ कुछ नहीं दिखाता कि असली model arguments कैसे चुनता है या error कैसे पढ़ता है। Jobs की गिनती वे in.php रिक्वेस्ट हैं जो mock को मिलीं।

इसे PowerShell में चलाएँ। यह virtual environment के Python को उसके path से call करता है, इसलिए आप उसे कभी activate नहीं करते।

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

अपनी values उसी window में सेट करें। CAPTCHA_ALLOWED_HOSTS सुरक्षा के लिए सबसे ज़रूरी है: model इसे कभी सेट नहीं करता, और खाली value हर call को ठुकरा देती है। सिर्फ़ host देने पर port 80 और 443 भी अनुमत होते हैं, और host:port ठीक उसी 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

चरण 1: CrewAI कैप्चा tool लिखें

अगले दो ब्लॉक को एक ही फ़ाइल captcha_tool.py के रूप में सेव करें। पहले में settings और argument schema हैं, जहाँ CrewAI जाँचता है कि model ने क्या भेजा। SolveArgs.kind field दो में से एक value लेता है, sitekey एक छोटी सादी string है, और पेज का पता आपके सूचीबद्ध किए गए किसी host से मेल खाना चाहिए। Python package CAPSKIP_HOST या किसी और variable को नहीं पढ़ता, इसलिए फ़ाइल key, host और port हाथ से पास करती है।

# 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

दूसरा ब्लॉक जोड़ें। इसकी चार settings इस गाइड का असली मुद्दा हैं। result_as_answer token को task का जवाब बनाता है। cache_function कभी cache न करने को कहता है। max_usage_count हर run में calls की सीमा तय करता है। और नाकामी एक ToolFailure के रूप में लौटती है, जो CrewAI की एक class है और बिना raise किए किसी call को failed चिह्नित करती है। model उसका संदेश और code पढ़ता है, और उसका ToolFailure.retryable flag आपके लिए एक नोट है।

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

अगले ब्लॉक को try_tool.py नाम से सेव करें। project folder में यह चलाएँ .\.venv\Scripts\python try_tool.py। यह tool को हाथ से call करता है, इसे Anthropic key नहीं चाहिए और यह token की लंबाई प्रिंट करता है। Python SDK कोई HTTP timeout सेट नहीं करता, इसलिए जो CapSkip कनेक्शन ले ले पर कभी जवाब न दे, वह इस फ़ाइल को हमेशा के लिए block कर देगा। चरण 3 में एक 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)

दूसरी जाँच ऐसी call है जिसे tool को ठुकराना ही चाहिए। इसे try_hostile.py नाम से सेव करें। यह जो पहली लाइन प्रिंट करती है वह कहती है कि arguments validation में फेल हुए, और CapSkip को कोई टास्क नहीं मिलता।

# 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: model क्या देखता है, और वह क्या पार कर सकता है

मैंने CrewAI के Anthropic provider को scripted server की ओर ANTHROPIC_BASE_URL variable के ज़रिए मोड़ा और request लॉग की। उसमें कोई tool choice, temperature या thinking field नहीं था। tool strict: true के साथ आया, और kind field के लिए एक enum के साथ, लेकिन sitekey की length limits और pattern गायब थीं, और पेज के पते की length limit भी, क्योंकि Anthropic के strict mode के लिए CrewAI वे keywords हटा देता है। इसलिए वे limits सिर्फ़ आपके tool में रहती हैं: tool चलते समय CrewAI arguments को args_schema के विरुद्ध जाँचता है, usage count या SDK से पहले। मैं टेस्ट नहीं कर सका कि असली API यह request स्वीकार करती है या नहीं।

मैंने करीब 100 argument sets सीधे tool को भेजे, दो अलग attack lists से, और कुछ agent loop के ज़रिए।

parser तक सिर्फ़ बिना spaces वाला printable ASCII पहुँचता है, क्योंकि urlparse host पढ़ने से पहले tabs और newlines हटा देता है, और जाँच फिर पूरे host और port से मिलान करती है। ठुकराई गई call model को error text के रूप में लौटती है, 1,000,000 अक्षरों वाले sitekey के लिए 858 अक्षर। ठुकराई गई call max_usage_count को खर्च नहीं करती, और जो model बार-बार खराब पता भेजता रहा उसे CrewAI की repeat check ने पाँच calls बाद बिना किसी job के रोक दिया। चरण 3 की raise policy के तहत ठुकराई गई call की जगह run खत्म हो जाता है।

चरण 3: crew में CrewAI कैप्चा tool चलाएँ और token पोस्ट करें

agent arguments चुनता है और बाकी सब आपकी script करती है, इसलिए model के पास token कभी नहीं होता। पहले form helper, submit_form.py। यह सादा Python है, tool नहीं, और यह उस poster का एक रूप है जो LangChain कैप्चा tool गाइड में है। यह redirects फॉलो नहीं करता और ऐसा form ठुकरा देता है जिसकी action scheme, host या 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

मैंने इसे एक naive poster के सामने आज़माया जो redirects फॉलो करता है और form action पर भरोसा करता है। http और https दोनों पर, किसी दूसरे host name पर redirect, या उस पर form action, जारी किया गया token उस host पर रख देता था, और helper ने वहाँ कुछ नहीं भेजा। वह ऐसे https पेज को भी ठुकराता है जिसका form action उसी host और port पर http है, और उसी host के भीतर के redirect को भी, इसलिए अंतिम पता इस variable में रखें: PAGE_URL.

अब run_crew.py। raise policy किसी नाकाम tool call को run रोकने देती है और वह आपके code तक exception के रूप में पहुँचता है, जहाँ ToolFailure.retryable पढ़ा जा सकता है। script एक बार retry करती है और सिर्फ़ वही token स्वीकार करती है जो tool ने बनाया हो। इसे इस command से शुरू करें: .\.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 पर, एक unsolvable CAPTCHA और फिर एक अच्छे CAPTCHA ने एक नाकाम कोशिश और एक token प्रिंट किया, और helper ने 200 जवाब लौटाया: 2 model requests और 2 jobs। जो model शब्दों में जवाब देता है, और जो model token गढ़ लेता है, दोनों RuntimeError से टकराए। हर नाकाम कोशिश पर CrewAI stdout पर दो Event pairing mismatch warnings और stderr पर एक ERROR लाइन भी प्रिंट करता है, जो tool और failure code का नाम बताती है। retry फिर भी सफल होता है।

watchdog hang को सँभालता है। यह crew को एक child process में चलाता है और seconds में दी गई सीमा के बाद उसे रोक देता है, जिसे आप फ़ाइल नाम के बाद पास कर सकते हैं। 900 का डिफ़ॉल्ट दो कोशिशें देता है, क्योंकि SDK हर बार reCAPTCHA job को 300 सेकंड तक poll करता है। crew की जगह .\.venv\Scripts\python watchdog.py चलाएँ।

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

एक ऐसे mock CapSkip के सामने जो job स्वीकार करता है और कभी जवाब नहीं देता, 15 की सीमा ने run को 15.1 सेकंड बाद इस संदेश के साथ खत्म किया कि run_crew.py रोक दिया गया, और exit code 1 के साथ। mock को 1 job मिली थी, और कोई Python process कनेक्शन थामे नहीं रहा। hang mock का अपना नियम है, इसलिए असली CapSkip अलग तरह से फेल हो सकता है।

token को लेकर दो चेतावनियाँ। verbose=True के साथ CrewAI ने उसे console पर प्रिंट किया। CrewAI आख़िरी kickoff का जवाब, token समेत, इस फ़ाइल में भी सेव करता है: latest_kickoff_task_outputs.db यह फ़ाइल आपके local app data folder में CrewAI के नीचे रहती है। मुझे source में इस write को रोकने वाला कोई switch नहीं मिला, इसलिए script हर सफल run के बाद फ़ाइल हटा देती है। token model को भेजी गई किसी भी request body में नहीं दिखा।

CrewAI failures, retries, cache_function और max_usage_count को कैसे संभालता है

पंक्तियाँ scripted server के सामने Anthropic wrapper इस्तेमाल करती हैं, सिवाय उन दो के जिन पर text path लिखा है, जिन्होंने function calling के बिना एक in-process stand-in model इस्तेमाल किया, इसलिए CrewAI ने अपना ReAct text format इस्तेमाल किया।

native path पर बाहर निकलता exception model तक text Error executing tool और SDK संदेश के रूप में पहुँचता है, और उसे कुछ दोबारा नहीं चलाता। ToolFailure model को सिर्फ़ आपका संदेश और code दिखाता है। text path पर CrewAI ने raise करने वाले tool को अपने आप दोबारा चलाया, जिससे एक नाकामी तीन jobs बन गई, इसलिए नाकामी को लौटाएँ।

max_usage_count jobs को सीमित करता है, model requests को नहीं: 3 की सीमा के साथ model ने 9 requests कीं और mock ने 3 jobs देखीं। counter हर run पर फिर शुरू होता है: एक crew के पाँच kickoffs ने 3 की सीमा के तहत पाँच jobs दीं। सीमा न होने पर CrewAI ने पाँचवीं एक जैसी call के बाद loop रोक दिया और एक repeated call failure दर्ज की। ToolFailure.retryable इन runs में कुछ तय नहीं करता, और source उसे सिर्फ़ दर्ज करता है और tracing को पास करता है।

Caching सबसे खतरनाक है। 1.15.35 में crew अपना tool cache तब तक बंद रखता है जब तक आप Crew(cache=True) पास न करें। एक जैसे arguments वाले दो kickoffs ने CrewAI के डिफ़ॉल्ट cache function के बावजूद 2 jobs बनाईं। cache चालू और वही डिफ़ॉल्ट होने पर उन्होंने 1 job बनाई और वही token दो बार लौटाया, और mock साइट ने दूसरी post को already used बताकर ठुकरा दिया। tool में never_cache होने पर, cache चालू रहते हुए भी फिर 2 jobs बनीं। ToolFailure कभी cache नहीं होता। असली token कितनी देर चलता है, इसके लिए देखें: reCAPTCHA टोकन की एक्सपायरी और cache function तथा failure policies की जानकारी CrewAI के tools दस्तावेज़ में है।

आपको ये errors मिल सकते हैं

FAQ

संक्षेप में

एक अच्छा CrewAI कैप्चा tool schema से ही ठुकराने का काम करवाता है, नाकामियों को raise करने के बजाय लौटाता है, tool के लिए cache बंद रखता है और retries आपके अपने code में रखता है। ये guards ऊपर के हर टेस्ट में टिके रहे। अपने खुद के Windows PC पर चल रहे कैप्चा बायपास के साथ हर कोशिश में प्रति-हल कुछ खर्च नहीं होता, और Python कैप्चा सॉल्वर पेज में बाकी CAPTCHA प्रकारों के लिए SDK calls की सूची है।