reCAPTCHA v3をLlamaIndexエージェントで解決する方法

LlamaIndexエージェント
    ↓ トゥールを選択
FunctionToolラッパー
    ↓ 信頼できる参照を検証
CapSolver実行者
    ↓ 短期間の解決を返す
ブラウザサービス
    ↓ 送信と検証
LlamaIndexワークフローが再開

from dataclasses import dataclass
from urllib.parse import urlparse

@dataclass(frozen=True)
class CaptchaContext:
    context_id: str
    website_url: str
    website_key: str
    page_action: str
    enterprise: bool = False
    proxy_profile: str | None = None
    session_mode: bool = False

TRUSTED_CONTEXTS: dict[str, CaptchaContext] = {}
ALLOWED_HOSTS = {"staging.example.com", "portal.example.org"}


def get_trusted_context(context_id: str) -> CaptchaContext:
    context = TRUSTED_CONTEXTS.get(context_id)
    if context is None:
        raise ValueError("Unknown CAPTCHA context")

    host = urlparse(context.website_url).hostname
    if host not in ALLOWED_HOSTS:
        raise PermissionError("Target is outside the approved host policy")

    if not context.website_key or not context.page_action:
        raise ValueError("Trusted context is missing required v3 parameters")

    return context

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

export CAPSOLVER_API_KEY="your-capsolver-api-key"

import os
from capsolver_agent.schema import create_executor

executor = create_executor(
    api_key=os.environ["CAPSOLVER_API_KEY"],
    default_timeout=120,
)

from typing import Annotated

async def solve_recaptcha_v3(
    context_id: Annotated[
        str,
        "信頼できる現在のブラウザCAPTCHAコンテキストの不透明なID"
    ],
) -> dict:
    """承認されたブラウザコンテキストのreCAPTCHA v3を解決します。

    現在のワークフローがサポートされているreCAPTCHA v3チェックポイントを報告している場合にのみ使用してください。ターゲットURL、サイトキー、またはアクションを推測または変更しないでください。
    """
    context = get_trusted_context(context_id)

    captcha_type = (
        "reCaptchaV3Enterprise"
        if context.enterprise
        else "reCaptchaV3"
    )

    args = {
        "captcha_type": captcha_type,
        "website_url": context.website_url,
        "website_key": context.website_key,
        "page_action": context.page_action,
    }

    if context.proxy_profile:
        args["proxy"] = resolve_proxy(context.proxy_profile)

    result = await executor.execute("solve_captcha", args)
    if not result.get("success"):
        return {
            "success": False,
            "context_id": context_id,
            "error": normalize_error(result.get("error")),
        }

    solution = result.get("solution") or {}
    token = solution.get("token")
    if not token:
        return {
            "success": False,
            "context_id": context_id,
            "error": "solution did not contain a token",
        }

    receipt = await submit_solution_and_verify(
        context_id=context_id,
        token=token,
        session_cookie=extract_session_cookie(solution),
    )

    return {
        "success": receipt["verified"],
        "context_id": context_id,
        "verified": receipt["verified"],
        "next_state": receipt["next_state"],
    }

from llama_index.core.tools import FunctionTool

tool = FunctionTool.from_defaults(
    async_fn=solve_recaptcha_v3,
    name="solve_recaptcha_v3",
    description=(
        "承認された現在のブラウザコンテキストのreCAPTCHA v3を解決します。 "
        "入力はブラウザサービスによって提供された不透明なcontext_idでなければなりません。 "
        "サポートされていないページや承認されていないホストでは呼び出さないでください。"
    ),
)

schema = tool.metadata.get_parameters_dict()
print(schema)

from llama_index.core.agent.workflow import FunctionAgent

agent = FunctionAgent(
    llm=llm,
    tools=[tool],
    system_prompt=(
        "承認されたブラウザワークフローのみを操作してください。ブラウザサービスがサポートされているreCAPTCHA v3チェックポイントを報告した場合、"
        "提供されたcontext_idでsolve_recaptcha_v3を呼び出してください。一度だけ呼び出してください。"
        "verified=trueのときのみ継続してください。それ以外の場合はレビューを要求してください。"
    ),
)

response = await agent.run(
    "承認されたステージングワークフローはreCAPTCHA v3チェックポイントで待機しています。"
    "context_id ctx_7f19を使用し、verifiedのときのみ継続してください。"
)

async def collect_v3_context(page, context_id: str) -> CaptchaContext:
    website_url = page.url
    host = urlparse(website_url).hostname
    if host not in ALLOWED_HOSTS:
        raise PermissionError("Unapproved target")

    values = await page.evaluate("""
    () => {
      const scripts = Array.from(document.scripts)
        .map(s => s.textContent || '')
        .join('\n');

      const siteKey =
        document.querySelector('[data-sitekey]')?.getAttribute('data-sitekey')
        || null;

      const actionMatch = scripts.match(
        /grecaptcha(?:\.enterprise)?\.execute\([^,]+,\s*\{\s*action:\s*['\"]([^'\"]+)/
      );

      return {
        siteKey,
        pageAction: actionMatch ? actionMatch[1] : null,
        enterprise: scripts.includes('grecaptcha.enterprise')
      };
    }
    """)

    if not values["siteKey"] or not values["pageAction"]:
        raise RuntimeError("Could not read required v3 parameters")

    return CaptchaContext(
        context_id=context_id,
        website_url=website_url,
        website_key=values["siteKey"],
        page_action=values["pageAction"],
        enterprise=values["enterprise"],
    )

SESSION_KEYS = {
    "recaptcha-ca-t",
    "recaptcha_ca_t",
}


def extract_session_cookie(solution: dict) -> str | None:
    raw = solution.get("raw") or {}
    for key in SESSION_KEYS:
        value = solution.get(key) or raw.get(key)
        if value:
            return value
    return None

async def submit_solution_and_verify(
    context_id: str,
    token: str,
    session_cookie: str | None,
) -> dict:
    browser_state = BROWSER_CONTEXTS[context_id]
    page = browser_state.page

    if session_cookie:
        await browser_state.context.add_cookies([{
            "name": "recaptcha-ca-t",
            "value": session_cookie,
            "domain": urlparse(page.url).hostname,
            "path": "/",
            "secure": True,
        }])

    await page.evaluate(
        """({ token }) => {
          let input = document.querySelector(
            'textarea[name="g-recaptcha-response"]'
          );
          if (!input) {
            input = document.createElement('textarea');
            input.name = 'g-recaptcha-response';
            input.style.display = 'none';
            document.body.appendChild(input);
          }
          input.value = token;
          input.dispatchEvent(new Event('change', { bubbles: true }));
        }""",
        {"token": token},
    )

    await trigger_trusted_callback(page, browser_state.callback_name)

    try:
        await page.locator(browser_state.success_selector).wait_for(
            state="visible",
            timeout=15000,
        )
        return {"verified": True, "next_state": "continue"}
    except Exception:
        return {"verified": False, "next_state": "operator_review"}

from enum import Enum

class RecoveryState(str, Enum):
    DETECTED = "detected"
    SOLVING = "solving"
    VERIFIED = "verified"
    REVIEW_REQUIRED = "review_required"

ATTEMPTS: dict[str, int] = {}

async def guarded_solve(context_id: str) -> dict:
    attempts = ATTEMPTS.get(context_id, 0)
    if attempts >= 1:
        return {
            "success": False,
            "context_id": context_id,
            "next_state": RecoveryState.REVIEW_REQUIRED,
            "error": "recovery budget exhausted",
        }

    ATTEMPTS[context_id] = attempts + 1
    return await solve_recaptcha_v3(context_id)

from datetime import datetime, timezone


def recovery_event(context: CaptchaContext, result: dict) -> dict:
    return {
        "event": "recaptcha_v3_recovery",

websiteKeyをログに記録しないでください。あなたのポリシーが設定情報として扱う場合、解決トークン、セッションクッキー、APIキー、プロキシ、または完全なプライベートページHTMLをログに記録しないでください。

[CapSolverエラーのFAQ](https://www.capsolver.com/faq/errors-and-troubleshooting)は、エラーのカテゴリを正規化するのに役立ちます。

> **ボーナスコード**: [CapSolverダッシュボード](https://dashboard.capsolver.com/dashboard/overview/?utm_source=offcial&utm_medium=blog&utm_campaign=how-to-solve-recaptcha-v3-in-llamaindex-agents)でコード **WEBS** を使用すると、毎回のチャージで追加の5%ボーナスが得られます。

## 比較概要

| 統合パターン | モデル入力 | シークレット漏洩リスク | 最適な用途 |
|---|---|---:|---|
| モデルがすべてのタスクフィールドを提供 | URL、キー、アクション、プロキシ | 高 | 本番環境では避ける |
| 検証済みフィールドを持つタイプ付きFunctionTool | 明示的なフィールド | 中 | コントロールされたプロトタイプ |
| 透過的なコンテキストIDとサーバー検証 | コンテキスト参照のみ | 低 | 本番LlamaIndexワークフロー |
| ブラウザのみのコア`solve_on_page` | モデルパラメータなし | 最低 | 確定的なPlaywright復元 |

透過的なコンテキストパターンにより、LlamaIndexエージェントは、シーケンス固有のパラメータを再設定することなく復元を要求できるようになります。

## 本番環境チェックリスト

- APIキーとプロキシプロファイルをシークレットマネージャーに保持する。
- 承認されたホストと正確なワークフローの目的のみを許可する。
- 現在のライブページから`websiteKey`と`pageAction`を読み込む。
- エンタープライズとセッション設定をターゲット統合に一致させる。
- 信頼できるブラウザコードを通じてトークンを即座に提出する。
- 継続する前に予期されるアプリケーション状態を検証する。
- 1回の解決試行後にオペレータレビューにルーティングする。
- トレースからトークン、クッキー、プロキシ、資格情報を削除する。
- SDKまたはプロンプトが変更されるたびにツールスキーマを再テストする。

[CapSolver製品ページ](https://www.capsolver.com/products)にはサポートされている解決カテゴリがリストアップされており、[CapSolver AIブログ](https://www.capsolver.com/blog/ai)では関連するエージェント統合パターンがカバーされています。

## 責任ある使用

このワークフローは、所有するアプリケーション、テストするアプリケーション、または自動化に明示的な許可を与えたアプリケーションでのみ使用してください。技術的スキルがアクセス権を保証するものではありません。ターゲットの利用規約、レートリミット、プライバーや認証の境界を尊重してください。許可なしにプライベートアカウント、制限付き記録、またはサードパーティワークフローにエージェントツールを使用しないでください。送信、支払い、予約、アカウント変更などの高影響力アクションは、別途のポリシーと確認ステップの後で行うべきです。

## 結論

本番環境用のLlamaIndex reCAPTCHA v3ソルバーは、狭いタイプの復元関数を公開する必要があります。ブラウザサービスは信頼できるコンテキストIDを提供し、サーバーサイドコードは正確なURL、サイトキー、アクション、エンタープライズモード、プロキシポリシーを保持し、CapSolverは一時的な解決を返し、エージェントが続行する前にブラウザが予期される状態を検証します。

[CapSolver](https://www.capsolver.com/?utm_source=offcial&utm_medium=blog&utm_campaign=how-to-solve-recaptcha-v3-in-llamaindex-agents)で承認されたLlamaIndex統合を開始し、制御されたステージングワークフローでテストし、本番環境に移行する前にパラメータの基盤と再試行アサーションを追加してください。

## FAQ

### reCAPTCHA v3にはチェックボックスのクリックが必要ですか？

いいえ。reCAPTCHA v3はスコアベースで、通常はバックグラウンドで動作します。ターゲットのサイトキー、URL、アクションを保持する必要があります。

### `pageAction`が重要な理由は何ですか？

アクションは評価されている操作を識別し、ログインや送信などの例があります。ライブ統合から読み取った正確なアクションを使用し、汎用的な値ではなくしてください。

### LlamaIndexエージェントにトークンを渡すべきですか？

サーバーサイドの送信を優先し、確認済みのステータスのみを返してください。トークンは一時的なランタイムデータであり、モデルコンテキストやログに含まれてはなりません。

### セッションモードを有効にするのはいつですか？

認可されたターゲットが返されたセッション値を必要とする場合のみ有効にしてください。その値を一時的な暗号化されたランタイムストレージに保持してください。

### 失敗した場合に何が起こるべきですか？

設定された試行予算後に停止し、赤字された診断イベントを記録し、適切な場合に信頼できるページパラメータを更新し、ワークフローをオペレータレビューにルーティングしてください。