Cloudflare TurnstileをAutoGenエージェントで解決する方法

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を呼び出してください。"
        "決してターゲットを発明したり、資格情報を要求したりしないでください。"
        "ツールが2回失敗した場合、停止してオペレーターのレビューを求めてください。"
    ),
    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=(
        "承認されたHTTPSページのCloudflare Turnstileを、正確なサイトキーとオプションのaction/cdata値を使用して解決します。"
    ),
)

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"
        "登録されたツールを1回呼び出し、構造化された結果を返してください。"
    ),
    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 = "unknown 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,
    }