Cách giải reCAPTCHA v3 trong CrewAI Agents với 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("Observed URL does not match policy")
    if observation.observed_site_key != policy.website_key:
        raise ValueError("Observed site key does not match policy")
    if observation.observed_action != policy.page_action:
        raise ValueError("Observed page action does not match policy")
    if observation.form_state != "READY_FOR_VERIFICATION":
        raise ValueError("Application state is not ready")

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="Định danh mục tiêu đã đăng ký")
    crew_role: str = Field(description="Vai trò yêu cầu xác minh")
    observed_action: str = Field(description="Hành động quan sát trên trang trực tiếp")
    observed_site_key: str = Field(description="Khóa trang web quan sát trên trang trực tiếp")
    state_id: str = Field(description="Định danh trạng thái ứng dụng phía máy chủ không rõ")

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:
    """Hàm submit và kiểm tra do ứng dụng sở hữu."""
    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("Giải reCAPTCHA v3 đã phê duyệt", 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:
    """Giải một bước reCAPTCHA v3 đã đăng ký và xác minh trạng thái ứng dụng."""
    policy = TARGETS.get(target_id)
    if policy is None:
        return SolveResult(
            status="stopped",
            target_id=target_id,
            state_id=state_id,
            reason="Mục tiêu không biết",
            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="Vai trò không được phép gọi công cụ này",
            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="Hành động quan sát không khớp chính sách mục tiêu",
            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="Khóa trang quan sát không khớp chính sách mục tiêu",
            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="Nhiệm vụ CapSolver không hoàn thành",
            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="Kết quả nhiệm vụ không chứa mã thông báo",
            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="Trạng thái ứng dụng được xác minh" if verified else "Trạng thái ứng dụng không được xác minh",
        task_attempted=True,
    ).model_dump()

from crewai import Agent, Crew, Process, Task

verification_agent = Agent(
    role="verification_specialist",
    goal="Hoàn thành chỉ các bước xác minh đã đăng ký và báo cáo trạng thái đã xác minh",
    backstory=(
        "Bạn vận hành các công cụ xác minh được phê duyệt. Bạn không bao giờ tạo ID mục tiêu, "
        "khóa trang, hành động, thông tin xác thực hoặc trạng thái thành công."
    ),
    tools=[solve_approved_recaptcha_v3],
    allow_delegation=False,
    verbose=True,
)

verification_task = Task(
    description=(
        "Đối với mục tiêu đã đăng ký trong quan sát được cung cấp, gọi công cụ một lần "
        "chỉ khi URL, khóa trang, hành động và trạng thái được xác nhận. Trả về kết quả "
        "có cấu trúc mà không có bí mật."
    ),
    expected_output="Một kết quả có cấu trúc, được xác minh, yêu cầu xem xét hoặc dừng.",
    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": "Trạng thái ứng dụng được xác minh",
  "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