Инструмент CrewAI для капчи: решаем reCAPTCHA и Turnstile

Чтобы создать инструмент CrewAI для капчи, унаследуйте BaseTool, задайте ему схему, которая отклоняет плохой ввод, и вызывайте Python SDK CapSkip из _run. Чтобы токен не попадал к модели, задайте result_as_answer, что делает его ответом задачи. Затем три настройки CrewAI решают, безопасно ли оставлять инструмент работать: кэш инструмента, который может воспроизвести мёртвый токен, лимит использований, который считается на каждый запуск, и путь повтора, который может сам перезапустить упавший инструмент.

Что понадобится

• CapSkip, установленный на ПК с Windows, с включённым решением reCAPTCHA v2 и Turnstile.
• Python от 3.10 до 3.13. crewai 1.15.35 объявляет 3.10 или новее и ниже 3.14, а capskip объявляет 3.10 или новее. На Python 3.14.0 pip сопоставил crewai[anthropic] со старым crewai 0.11.2 лишь с предупреждением, поэтому команды ниже используют py -3.13. Если там сказано, что 3.13 не установлен, сначала скачайте его с python.org. Я работал на Windows 11 с crewai 1.15.35 (самым новым выпуском на PyPI на 5 ноября 2026), capskip 1.6.0, anthropic 0.73.0 и pydantic 2.12.5, на Python 3.13.7 и 3.11.7, но не на 3.10 и 3.12.
• Окно PowerShell и страница, которой вы владеете, с флажком reCAPTCHA v2. Вспомогательная функция формы из Шага 3 ожидает скрытое поле с именем _csrf и поле email, поэтому поменяйте имена под свою форму.
• Маршрут к CapSkip. Режим Local (127.0.0.1) доступен только с того же ПК. Режим Server заставляет CapSkip принимать вызовы API и с других машин. Режим задаётся в разделе Настройки подключения в руководстве по настройке.
• Ключ Anthropic API, только для запуска агента в Шаге 3. Инструмент работает и без него.

Как проверялось: на имитации CapSkip и имитации страницы регистрации, которые я написал, с настоящим CrewAI и настоящим SDK. Сценарный локальный сервер отвечал в формате Anthropic Messages, поэтому настоящая обёртка CrewAI отправляла настоящие тела запросов, и я их читал. Настоящая модель не вызывалась, а идентификатор модели anthropic/claude-opus-5-5 так и не дошёл до настоящего API, поэтому ничто здесь не показывает, как настоящая модель выбирает аргументы или читает ошибку. Число заданий считается по запросам in.php, полученным имитацией.

Выполните это в PowerShell. Оно вызывает Python виртуального окружения по пути, поэтому активировать окружение не нужно.

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

Задайте свои значения в том же окне. CAPTCHA_ALLOWED_HOSTS важнее всего для безопасности: модель его никогда не задаёт, а пустое значение отклоняет каждый вызов. Хост без порта разрешает ещё и порты 80 и 443, а host: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 для капчи

Сохраните следующие два блока как один файл captcha_tool.py. Первый содержит настройки и схему аргументов, где CrewAI проверяет то, что отправила модель. Поле SolveArgs.kind принимает одно из двух значений, sitekey должен быть короткой простой строкой, а адрес страницы должен совпадать с хостом из вашего списка. Пакет Python не читает CAPSKIP_HOST и вообще ни одну другую переменную, поэтому файл передаёт ключ, хост и порт вручную.

# 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

Добавьте второй блок. Четыре настройки в нём составляют суть этого руководства. result_as_answer делает токен ответом задачи. cache_function велит никогда не кэшировать. max_usage_count ограничивает число вызовов за запуск. А сбой возвращается как ToolFailure, класс CrewAI, который помечает вызов как неудачный, не выбрасывая исключения. Модель читает его сообщение и код, а флаг ToolFailure.retryable остаётся пометкой для вас.

# 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. В папке проекта выполните .\.venv\Scripts\python try_tool.py. Он вызывает инструмент вручную, не требует ключа Anthropic и печатает длину токена. Python SDK не задаёт HTTP-таймаут, поэтому CapSkip, который принимает соединение и никогда не отвечает, заблокировал бы этот файл навсегда. Шаг 3 добавляет сторожевой процесс.

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

Вторая проверка: вызов, который инструмент обязан отклонить. Сохраните его как try_hostile.py. Первая строка, которую он печатает, сообщает, что аргументы не прошли проверку, и 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: что видит модель и что она может обойти

Я направил провайдер Anthropic в CrewAI на сценарный сервер через переменную ANTHROPIC_BASE_URL и записал запрос в журнал. В нём не было выбора инструмента, температуры и поля thinking. Инструмент пришёл с strict: true и enum для поля kind, но у sitekey пропали ограничения длины и шаблон, а у адреса страницы ограничение длины, потому что CrewAI убирает эти ключевые слова для строгого режима Anthropic. Поэтому эти ограничения живут только в вашем инструменте: CrewAI проверяет аргументы по args_schema при запуске инструмента, до учёта использований и до SDK. Проверить, примет ли настоящий API такой запрос, я не смог.

Я отправил напрямую в инструмент около 100 наборов аргументов из двух отдельных списков атак и несколько через цикл агента.

До парсера доходит только печатный ASCII без пробелов, потому что urlparse отбрасывает табуляции и переводы строк до чтения хоста, а затем проверка сравнивает хост и порт целиком. Отказ возвращается модели как текст ошибки, 858 символов для sitekey из 1 000 000 символов. Отказ не расходует max_usage_count, а модель, которая продолжала отправлять плохой адрес, была остановлена проверкой повторов CrewAI после пяти вызовов, без задания. При политике raise из Шага 3 отказ вместо этого завершает запуск.

Шаг 3: запускаем инструмент CrewAI для капчи в команде и отправляем токен

Агент выбирает аргументы, а остальное делает ваш скрипт, поэтому модель никогда не держит токен. Сначала вспомогательная функция формы, submit_form.py. Это обычный Python, а не инструмент, и вариант отправителя из руководства по инструменту капчи в LangChain. Она не следует перенаправлениям и отклоняет форму, у которой action отличается от страницы по схеме, хосту или порту.

# 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

Я проверил её на наивном отправителе, который следует перенаправлениям и доверяет action формы. Как по http, так и по https перенаправление на другое имя хоста или action формы на нём отправляли выданный токен на этот хост, а вспомогательная функция туда ничего не отправила. Она также отклоняет страницу https, у формы которой action указывает на http с тем же хостом и портом, и перенаправление внутри того же хоста, поэтому укажите конечный адрес в PAGE_URL.

Теперь run_crew.py. Политика raise останавливает запуск при неудачном вызове инструмента, и сбой доходит до вашего кода как исключение, где ToolFailure.retryable доступен для чтения. Скрипт повторяет один раз и принимает только токен, который выпустил инструмент. Запустите его командой .\.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]"))

На имитации одна неразрешимая CAPTCHA и затем хорошая дали одну неудачную попытку и токен, а вспомогательная функция вернула ответ 200: 2 запроса к модели и 2 задания. Модель, ответившая словами, и модель, выдумавшая токен, обе упёрлись в RuntimeError. При каждой неудачной попытке CrewAI также печатает в stdout два предупреждения Event pairing mismatch и одну строку ERROR в stderr с именем инструмента и кодом сбоя. Повтор всё равно проходит успешно.

Сторожевой процесс закрывает случай зависания. Он запускает команду в дочернем процессе и останавливает её по истечении лимита в секундах, который можно передать после имени файла. Значение по умолчанию 900 допускает две попытки, потому что SDK опрашивает задание reCAPTCHA до 300 секунд каждый раз. Выполните .\.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)

Против имитации CapSkip, которая принимает задание и никогда не отвечает, лимит 15 завершил запуск через 15.1 секунды с сообщением, что run_crew.py остановлен, и кодом выхода 1. Имитация получила 1 задание, и ни один процесс Python не остался держать соединение. Зависание задано правилом самой имитации, поэтому настоящий CapSkip может сбоить иначе.

Два предупреждения о токене. При verbose=True, CrewAI печатал его в консоль. CrewAI также сохраняет ответ последнего kickoff, токен в том числе, в latest_kickoff_task_outputs.db в папке CrewAI внутри вашей локальной папки данных приложений. Я не нашёл в исходном коде переключателя, который останавливает эту запись, поэтому скрипт удаляет файл после каждого удачного запуска. Токен не появился ни в одном теле запроса, отправленного модели.

Как CrewAI обрабатывает сбои, повторы, cache_function и max_usage_count

Строки используют обёртку Anthropic против сценарного сервера, кроме двух с пометкой «текстовый путь», где использовалась встроенная в процесс модель-заглушка без вызова функций, поэтому CrewAI применил текстовый формат ReAct.

На нативном пути вырвавшееся исключение доходит до модели как текст Error executing tool плюс сообщение SDK, и ничто его не перезапускает. ToolFailure показывает модели только ваше сообщение и код. На текстовом пути CrewAI сам перезапускал инструмент, который выбрасывал исключение, превращая один сбой в три задания, поэтому возвращайте сбой.

max_usage_count ограничивает число заданий, а не запросов к модели: при лимите 3 модель сделала 9 запросов, а имитация увидела 3 задания. Счётчик сбрасывается при каждом запуске: пять kickoff одной команды дали пять заданий при лимите 3. Без лимита CrewAI остановил цикл после пятого одинакового вызова и записал сбой повторного вызова. ToolFailure.retryable ничего не решал в этих запусках, а исходный код лишь записывает его и передаёт в трассировку.

Кэширование опаснее всего. В 1.15.35 команда оставляет кэш инструментов выключенным, если не передать Crew(cache=True): два kickoff с одинаковыми аргументами дали 2 задания даже с функцией кэша CrewAI по умолчанию. При включённом кэше и этой функции они дали 1 задание и вернули один и тот же токен дважды, а имитация сайта отклонила вторую отправку как уже использованную. С never_cache в инструменте включённый кэш снова дал 2 задания. ToolFailure никогда не кэшируется. Смотрите срок действия токена reCAPTCHA (сколько живёт настоящий токен). В документации CrewAI по инструментам описаны функция кэша и политики сбоев.

Ошибки, с которыми вы можете столкнуться

FAQ

Коротко

Хороший инструмент CrewAI для капчи заставляет схему отклонять плохой ввод, возвращает сбои вместо выбрасывания исключений, отключает кэш для инструмента и оставляет повторы вашему коду. Эти защиты выдержали все тесты выше. Когда обход капчи работает на вашем ПК с Windows, каждая попытка не стоит ничего за решение, а на странице сервиса распознавания капчи для Python перечислены вызовы SDK для других типов капчи.