CapSolver के साथ CrewAI एजेंट्स में reCAPTCHA v3 कैसे सॉल्व करें

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("अवलोकित यूआरएल नीति से मेल नहीं खाता")
    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("स्वीकृत रीकैपचा वी३ हल करें", 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:
    """एक पंजीकृत रीकैपचा वी३ चरण हल करें और एप्लिकेशन अवस्था की पुष्टि करें।"""
    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="कैपसॉल्वर कार्य पूर्ण नहीं हुआ",
            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=(
        "आप अनुमोदित सत्यापन उपकरण चलाते हैं। आप कभी लक्ष्य आईडी, साइट कुंजी, कार्य, प्रमाण पत्र या सफलता अवस्था नहीं बनाते हैं।"
    ),
    tools=[solve_approved_recaptcha_v3],
    allow_delegation=False,
    verbose=True,
)

verification_task = Task(
    description=(
        "प्रदान किए गए अवलोकन में निर्दिष्ट पंजीकृत लक्ष्य के लिए, यदि यूआरएल, साइट कुंजी, कार्य और अवस्था पुष्टि की गई है, तो केवल एक बार उपकरण का उपयोग करें। गुप्त जानकारी के बिना संरचित स्थिति लौटाएं।"
    ),
    expected_output="एक संरचित वेरिफाइड, रिव्यू आवश्यक या बंद परिणाम।",
    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