CrewAIエージェントでreCAPTCHA v3を解決する方法 – CapSolverを使って

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

export CAPSOLVER_API_KEY="CAP-..."

from dataclasses import dataclass

@dataclass(frozen=True)
class RecaptchaV3Policy:
    target_id: str
    website_url: str
    website_key: str
    page_action: str
    minimum_score: float
    enterprise: bool
    proxy_profile: str | None
    allowed_crew_role: str
    max_attempts: int = 1

TARGETS = {
    "approved_login_test": RecaptchaV3Policy(
        target_id="approved_login_test",
        website_url="https://approved.example.com/login",
        website_key="PUBLIC_SITE_KEY",
        page_action="login",
        minimum_score=0.7,
        enterprise=False,
        proxy_profile=None,
        allowed_crew_role="verification_specialist",
    )
}

@dataclass(frozen=True)
class PageObservation:
    target_id: str
    current_url: str
    observed_action: str
    observed_site_key: str
    form_state: str
    observed_at: str


def validate_observation(
    observation: PageObservation,
    policy: RecaptchaV3Policy,
) -> None:
    if observation.current_url != policy.website_url:
        raise PermissionError("観察されたURLがポリシーと一致しません")
    if observation.observed_site_key != policy.website_key:
        raise ValueError("観察されたサイトキーがポリシーと一致しません")
    if observation.observed_action != policy.page_action:
        raise ValueError("観察されたページアクションがポリシーと一致しません")
    if observation.form_state != "READY_FOR_VERIFICATION":
        raise ValueError("アプリケーション状態が準備ができていません")

import os
from typing import Literal

from crewai.tools import tool
from pydantic import BaseModel, Field
from capsolver_agent.schema import create_executor

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

class SolveRequest(BaseModel):
    target_id: str = Field(description="登録されたターゲット識別子")
    crew_role: str = Field(description="検証をリクエストする役割")
    observed_action: str = Field(description="ライブページで観察されたアクション")
    observed_site_key: str = Field(description="ライブページで観察されたサイトキー")
    state_id: str = Field(description="非透明なサーバーサイドアプリケーション状態識別子")

class SolveResult(BaseModel):
    status: Literal["verified", "review_required", "stopped"]
    target_id: str
    state_id: str
    reason: str
    task_attempted: bool

PROXY_VAULT = {
    "approved_proxy": os.environ.get("APPROVED_PROXY")
}

async def submit_token_and_verify(
    *,
    state_id: str,
    token: str,
    policy: RecaptchaV3Policy,
) -> bool:
    """アプリケーション所有の送信および検証関数。"""
    response = await application_sessions.submit_recaptcha_v3(
        state_id=state_id,
        token=token,
        expected_action=policy.page_action,
    )
    return (
        response.current_url.startswith("https://approved.example.com/account")
        and response.semantic_marker == "AUTHENTICATED_ACCOUNT_PAGE"
        and response.challenge_present is False
    )

@tool("承認されたreCAPTCHA v3を解決", result_schema=SolveResult)
async def solve_approved_recaptcha_v3(
    target_id: str,
    crew_role: str,
    observed_action: str,
    observed_site_key: str,
    state_id: str,
) -> dict:
    """登録されたreCAPTCHA v3ステップを解決し、アプリケーション状態を検証します。"""
    policy = TARGETS.get(target_id)
    if policy is None:
        return SolveResult(
            status="stopped",
            target_id=target_id,
            state_id=state_id,
            reason="不明なターゲット",
            task_attempted=False,
        ).model_dump()

    if crew_role != policy.allowed_crew_role:
        return SolveResult(
            status="stopped",
            target_id=target_id,
            state_id=state_id,
            reason="このツールを呼び出す権限がありません",
            task_attempted=False,
        ).model_dump()

    if observed_action != policy.page_action:
        return SolveResult(
            status="review_required",
            target_id=target_id,
            state_id=state_id,
            reason="観察されたアクションがターゲットポリシーと一致しません",
            task_attempted=False,
        ).model_dump()

    if observed_site_key != policy.website_key:
        return SolveResult(
            status="review_required",
            target_id=target_id,
            state_id=state_id,
            reason="観察されたサイトキーがターゲットポリシーと一致しません",
            task_attempted=False,
        ).model_dump()

    args = {
        "captcha_type": "reCaptchaV3",
        "website_url": policy.website_url,
        "website_key": policy.website_key,
        "page_action": policy.page_action,
        "min_score": policy.minimum_score,
        "enterprise": policy.enterprise,
    }
    if policy.proxy_profile:
        args["proxy"] = PROXY_VAULT[policy.proxy_profile]

    result = await executor.execute("solve_captcha", args)
    if not result.get("success"):
        return SolveResult(
            status="review_required",
            target_id=target_id,
            state_id=state_id,
            reason="CapSolverタスクが完了しませんでした",
            task_attempted=True,
        ).model_dump()

    solution = result.get("solution") or {}
    token = solution.get("token")
    if not token:
        return SolveResult(
            status="review_required",
            target_id=target_id,
            state_id=state_id,
            reason="タスク結果にトークンが含まれていません",
            task_attempted=True,
        ).model_dump()

    verified = await submit_token_and_verify(
        state_id=state_id,
        token=token,
        policy=policy,
    )
    return SolveResult(
        status="verified" if verified else "review_required",
        target_id=target_id,
        state_id=state_id,
        reason="アプリケーション状態が検証されました" if verified else "アプリケーション状態が検証されませんでした",
        task_attempted=True,
    ).model_dump()

from crewai import Agent, Crew, Process, Task

verification_agent = Agent(
    role="verification_specialist",
    goal="登録された検証ステップのみを完了し、検証された状態を報告する",
    backstory=(
        "あなたは承認された検証ツールを操作します。あなたはターゲットID、サイトキー、アクション、資格情報、または成功状態を発明しません。"
    ),
    tools=[solve_approved_recaptcha_v3],
    allow_delegation=False,
    verbose=True,
)

verification_task = Task(
    description=(
        "提供された観察に登録されたターゲットに対して、URL、サイトキー、アクション、および状態が確認された場合にのみツールを1回呼び出してください。シークレットを含まない構造化されたステータスを返してください。"
    ),
    expected_output="構造化されたverified、review_required、またはstoppedの結果。",
    agent=verification_agent,
)

crew = Crew(
    agents=[verification_agent],
    tasks=[verification_task],
    process=Process.sequential,
    verbose=True,
)

from datetime import datetime, timedelta, timezone

ATTEMPTS: dict[tuple[str, str], datetime] = {}


def claim_attempt(target_id: str, state_id: str) -> bool:
    key = (target_id, state_id)
    now = datetime.now(timezone.utc)
    prior = ATTEMPTS.get(key)
    if prior and now - prior < timedelta(minutes=2):
        return False
    ATTEMPTS[key] = now
    return True

{
  "status": "verified",
  "target_id": "approved_login_test",
  "state_id": "state_7c19",
  "reason": "アプリケーション状態が検証されました",
  "task_attempted": true
}

@dataclass(frozen=True)
class StateCheck:
    expected_path_prefix: str
    required_marker: str
    forbidden_markers: tuple[str, ...]


def is_verified(page, check: StateCheck) -> bool:
    return (
        page.url.path.startswith(check.expected_path_prefix)
        and page.has_semantic_marker(check.required_marker)
        and not any(page.contains(marker) for marker in check.forbidden_markers)
        and page.http_status == 200
    )

@dataclass(frozen=True)
class RecaptchaV3Policy:
    target_id: str
    website_url: str
    website_key: str
    page_action: str
    minimum_score: float
    enterprise: bool
    is_session: bool
    proxy_profile: str | None
    allowed_crew_role: str
    max_attempts: int = 1

SAFE_FIELDS = {
    "target_id",
    "crew_role",
    "action_match",
    "task_attempted",
    "provider_category",
    "duration_ms",
    "verified",
    "review_reason",
}


def safe_event(event: dict) -> dict:
    return {key: event[key] for key in SAFE_FIELDS if key in event}

import pytest

@pytest.mark.asyncio
async def test_unknown_target_stops_before_task():
    result = await solve_approved_recaptcha_v3.run(
        target_id="unknown",
        crew_role="verification_specialist",
        observed_action="login",
        observed_site_key="x",
        state_id="state-1",
    )
    assert result["status"] == "stopped"
    assert result["task_attempted"] is False

@pytest.mark.asyncio
async def test_action_mismatch_requests_review():
    result = await solve_approved_recaptcha_v3.run(
        target_id="approved_login_test",
        crew_role="verification_specialist",
        observed_action="checkout",
        observed_site_key="PUBLIC_SITE_KEY",
        state_id="state-2",
    )
    assert result["status"] == "review_required"
    assert result["task_attempted"] is False