CrewAI 验证码工具：识别 reCAPTCHA 和 Turnstile

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

# 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

# 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

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

# 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

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

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