Cara Menyelesaikan reCAPTCHA v3 di Agen CrewAI dengan 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 yang diamati tidak sesuai dengan kebijakan")
    if observation.observed_site_key != policy.website_key:
        raise ValueError("Kunci situs yang diamati tidak sesuai dengan kebijakan")
    if observation.observed_action != policy.page_action:
        raise ValueError("Tindakan halaman yang diamati tidak sesuai dengan kebijakan")
    if observation.form_state != "READY_FOR_VERIFICATION":
        raise ValueError("Keadaan aplikasi tidak siap")

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="Identifikasi target yang terdaftar")
    crew_role: str = Field(description="Peran yang meminta verifikasi")
    observed_action: str = Field(description="Tindakan yang diamati di halaman hidup")
    observed_site_key: str = Field(description="Kunci situs yang diamati di halaman hidup")
    state_id: str = Field(description="Identifikasi keadaan aplikasi sisi server yang tidak terlihat")

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:
    """Fungsi submit dan periksa yang dimiliki aplikasi."""
    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("Selesaikan reCAPTCHA v3 yang disetujui", 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:
    """Selesaikan satu langkah reCAPTCHA v3 yang terdaftar dan verifikasi keadaan aplikasi."""
    policy = TARGETS.get(target_id)
    if policy is None:
        return SolveResult(
            status="stopped",
            target_id=target_id,
            state_id=state_id,
            reason="Target tidak dikenal",
            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="Peran tidak diizinkan memanggil alat ini",
            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="Tindakan yang diamati tidak sesuai dengan kebijakan target",
            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="Kunci situs yang diamati tidak sesuai dengan kebijakan target",
            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="Tugas CapSolver tidak selesai",
            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="Hasil tugas tidak mengandung token",
            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="Keadaan aplikasi diverifikasi" if verified else "Keadaan aplikasi tidak diverifikasi",
        task_attempted=True,
    ).model_dump()

from crewai import Agent, Crew, Process, Task

verification_agent = Agent(
    role="verification_specialist",
    goal="Selesaikan hanya langkah verifikasi yang terdaftar dan laporkan keadaan yang diverifikasi",
    backstory=(
        "Anda mengoperasikan alat verifikasi yang disetujui. Anda tidak pernah menciptakan ID target, "
        "kunci situs, tindakan, kredensial, atau keadaan keberhasilan."
    ),
    tools=[solve_approved_recaptcha_v3],
    allow_delegation=False,
    verbose=True,
)

verification_task = Task(
    description=(
        "Untuk target yang terdaftar dalam pengamatan yang disediakan, panggil alat sekali "
        "hanya jika URL, kunci situs, tindakan, dan keadaan dikonfirmasi. Kembalikan "
        "hasil terstruktur tanpa rahasia."
    ),
    expected_output="Hasil terstruktur verified, review_required, atau 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": "Keadaan aplikasi diverifikasi",
  "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