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

pip install llama-index llama-index-llms-openai capsolver

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

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

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",
    "qa.example.com",
}


def solve_turnstile(
    website_url: Annotated[str, "承認されたHTTPSページに含まれるTurnstile"],
    website_key: Annotated[str, "ページからの正確なTurnstileサイトキー"],
    action: Annotated[str, "オプションのdata-action値"] = "",
    cdata: Annotated[str, "オプションのdata-cdata値"] = "",
) -> dict:
    """承認されたページのCloudflare 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": "トークンが返されませんでした"}
        return {
            "success": True,
            "token": token,
            "solution_type": solution.get("type", "turnstile"),
        }
    except Exception as exc:
        return {"success": False, "error": str(exc)}

from llama_index.core.tools import FunctionTool

turnstile_tool = FunctionTool.from_defaults(
    fn=solve_turnstile,
    name="solve_turnstile",
    description=(
        "承認されたHTTPSページでのCloudflare Turnstileのみを解決します。"
        "信頼できるブラウザコントローラーから提供された正確なページURL、サイトキー、"
        "およびオプションのaction/cdataを使用してください。"
    ),
)

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini", temperature=0)

agent = FunctionAgent(
    tools=[turnstile_tool],
    llm=llm,
    system_prompt=(
        "あなたは承認されたブラウザワークフローのみを操作します。"
        "信頼できるアプリケーションから正確なページURLとサイトキーが提供された場合にのみ"
        "solve_turnstileを呼び出してください。ターゲット、キー、アクション、cdataを発明しないでください。"
        "ツールを1回だけ呼び出してください。失敗した場合、オペレーターのレビューを求めて停止してください。"
    ),
)

import asyncio

async def main():
    response = await agent.run(
        "承認されたステージングワークフローは、"
        "https://staging.example.com/account-checkでTurnstileを見つけ、"
        "サイトキー0x4AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAとアクション"
        "account_checkを提供しました。登録されたツールを1回呼び出し、"
        "構造化された結果を返してください。"
    )
    print(response)

asyncio.run(main())

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

import asyncio

async def solve_with_policy(params: dict) -> dict:
    last_error = "unknown error"

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

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

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

pip install "capsolver-agent[browser] @ git+https://github.com/capsolver-ai/capsolver-agent.git"