如何在CrewAI代理中使用CapSolver解决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("观察到的 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、站点密钥、操作和状态确认后调用工具。返回结构化的状态，不包含秘密。"
    ),
    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