如何在 AutoGen 智能体中解决 Cloudflare 人机验证

pip install pyautogen capsolver

export CAPSOLVER_API_KEY="CAP-xxxxxxxxxxxxxxxx"
export OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxx"

pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install git+https://github.com/capsolver-ai/capsolver-agent.git

import os
from typing import Annotated
from urllib.parse import urlparse

import capsolver

capsolver.api_key = os.environ["CAPSOLVER_API_KEY"]

ALLOWED_HOSTS = {
    "staging.example.com",
    "app.example.com",
}


def solve_turnstile(
    website_url: Annotated[str, "包含 Turnstile 的批准页面 URL"],
    website_key: Annotated[str, "小部件的 Turnstile 站点密钥"],
    action: Annotated[str, "可选的 data-action 值"] = "",
    cdata: Annotated[str, "可选的 data-cdata 值"] = "",
) -> dict:
    """为批准的页面求解 Turnstile 并返回令牌。"""
    parsed = urlparse(website_url)
    if parsed.scheme != "https" or parsed.hostname not in ALLOWED_HOSTS:
        return {
            "success": False,
            "error": "目标不在批准的主机允许列表中",
        }

    if not website_key.startswith("0x4"):
        return {
            "success": False,
            "error": "意外的 Turnstile 站点密钥格式",
        }

    task = {
        "type": "AntiTurnstileTaskProxyLess",
        "websiteURL": website_url,
        "websiteKey": website_key,
    }

    metadata = {}
    if action:
        metadata["action"] = action
    if cdata:
        metadata["cdata"] = cdata
    if metadata:
        task["metadata"] = metadata

    try:
        solution = capsolver.solve(task)
        token = solution.get("token")
        if not token:
            return {"success": False, "error": "未返回 Turnstile 令牌"}
        return {
            "success": True,
            "token": token,
            "solution_type": solution.get("type", "turnstile"),
        }
    except Exception as exc:
        return {"success": False, "error": str(exc)}

import os
from autogen import ConversableAgent, register_function

assistant = ConversableAgent(
    name="TurnstileCoordinator",
    system_message=(
        "仅继续批准的自动化工作流。"
        "当应用程序报告 Turnstile 小部件并提供确切的页面 URL 和站点密钥时，仅调用 solve_turnstile。"
        "永远不要虚构目标或请求凭证。"
        "如果工具失败两次，请停止并请求操作员审查。"
    ),
    llm_config={
        "config_list": [{
            "model": "gpt-4o-mini",
            "api_key": os.environ["OPENAI_API_KEY"],
        }]
    },
)

executor = ConversableAgent(
    name="TurnstileToolExecutor",
    llm_config=False,
    human_input_mode="NEVER",
)

register_function(
    solve_turnstile,
    caller=assistant,
    executor=executor,
    name="solve_turnstile",
    description=(
        "使用其确切的站点密钥和可选的 action/cdata 值，为批准的 HTTPS 页面求解 Cloudflare Turnstile。"
    ),
)

chat_result = executor.initiate_chat(
    assistant,
    message=(
        "批准的测试工作流遇到了 Cloudflare Turnstile。\n"
        "website_url=https://staging.example.com/account-check\n"
        "website_key=0x4AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\n"
        "action=account_check\n"
        "cdata=\n"
        "调用注册的工具一次并返回结构化结果。"
    ),
    max_turns=4,
)

async def apply_turnstile_token(page, token: str):
    await page.evaluate(
        """
        (token) => {
          const response = document.querySelector(
            'input[name="cf-turnstile-response"]'
          );
          if (!response) {
            throw new Error('Turnstile response field not found');
          }
          response.value = token;
          response.dispatchEvent(new Event('input', { bubbles: true }));
          response.dispatchEvent(new Event('change', { bubbles: true }));
        }
        """,
        token,
    )

import asyncio

MAX_ATTEMPTS = 2

async def solve_with_policy(params: dict) -> dict:
    last_error = "未知错误"

    for attempt in range(1, MAX_ATTEMPTS + 1):
        result = solve_turnstile(**params)
        if result.get("success"):
            return {
                **result,
                "attempt": attempt,
            }

        last_error = result.get("error", last_error)
        if "allowlist" in last_error or "site-key" in last_error:
            break
        await asyncio.sleep(2 * attempt)

    return {
        "success": False,
        "error": last_error,
        "requires_operator_review": True,
    }