CrewAIのマルチエージェントワークフローでのキャプチャの解き方

# CapSolverコアエンジン（必須の依存関係）
pip install git+https://github.com/capsolver-ai/capsolver-core.git

# CapSolverエージェントツール
pip install git+https://github.com/capsolver-ai/capsolver-agent.git

# CrewAIフレームワーク
pip install crewai crewai-tools

export CAPSOLVER_API_KEY="your-capsolver-api-key"
export OPENAI_API_KEY="your-openai-api-key"

import asyncio
from crewai import Agent, Task, Crew
from crewai.tools import tool
from capsolver_agent.schema import create_executor

# CapSolverエクスキューターを作成
executor = create_executor(api_key="YOUR_CAPSOLVER_API_KEY")

@tool("CAPTCHAを解決")
def solve_captcha(captcha_type: str, website_url: str, website_key: str) -> str:
    """CAPTCHAチャレンジを解決し、トークンを返します。
    ウェブサイトでCAPTCHAに遭遇したときにこのツールを使用してください。
    
    引数:
        captcha_type: CAPTCHAの種類 - 'reCaptchaV2', 'reCaptchaV3', または 'cloudflare'
        website_url: CAPTCHAが表示されているページの完全なURL
        website_key: サイトキー（data-sitekey属性の値）
    
    戻り値:
        フォームに送信する解決されたCAPTCHAトークン
    """
    result = asyncio.run(executor.execute("solve_captcha", {
        "captcha_type": captcha_type,
        "website_url": website_url,
        "website_key": website_key
    }))
    
    if result["success"]:
        return f"CAPTCHAは正常に解決されました。トークン: {result['solution']['token']}"
    else:
        return f"CAPTCHA解決に失敗しました: {result['error']}"

from crewai import Agent

# CAPTCHA解決機能を持つ研究者エージェント
researcher = Agent(
    role="ウェブリサーチ専門家",
    goal="ウェブサイトからデータを収集し、遭遇したCAPTCHAチャレンジを処理します",
    backstory="""あなたはさまざまなオンラインソースからデータを収集するエキスパートのウェブリサーチャーです。ウェブサイトでCAPTCHAチャレンジに遭遇した場合は、solve_captchaツールを使用してクリアし、研究を続けます。reCaptchaV2（チェックボックスまたは非表示）、reCaptchaV3（スコアベース）、cloudflare（Turnstileウィジェット）のCAPTCHAタイプを識別する方法を知っています。""",
    tools=[solve_captcha],
    verbose=True
)

# データ収集エージェント
data_collector = Agent(
    role="データ収集専門家",
    goal="ターゲットウェブサイトから構造化されたデータを抽出します",
    backstory="あなたはウェブソースからデータを収集し、必要に応じてsolve_captchaツールを使用してCAPTCHAチャレンジを処理します。",
    tools=[solve_captcha],
    verbose=True
)

# 検証エージェント
verifier = Agent(
    role="データ検証専門家", 
    goal="収集されたデータを公式ソースと照合します",
    backstory="あなたは公式ソースをチェックしてデータの正確性を検証し、必要に応じてCAPTCHAを解決します。",
    tools=[solve_captcha],
    verbose=True
)

from crewai import Task, Crew, Process

# CAPTCHAに遭遇する可能性のあるタスク
research_task = Task(
    description="""{url}のターゲットウェブサイトをリサーチしてください。 
    CAPTCHAチャレンジに遭遇した場合は、その種類とサイトキーを特定し、solve_captchaツールを使用してトークンを取得してください。
    このサイトではreCAPTCHA v2を使用しており、サイトキーは{site_key}です。
    CAPTCHAを解決した後、必要なデータを収集してください。""",
    expected_output="必要に応じてCAPTCHAトークンを含むターゲットウェブサイトからのデータを収集",
    agent=researcher
)

# 分析タスク（リサーチ結果に依存）
analysis_task = Task(
    description="リサーチャーが収集したデータを分析し、要約レポートを作成してください。",
    expected_output="収集されたデータの構造化された分析レポート",
    agent=data_collector
)

# クルーを作成
crew = Crew(
    agents=[researcher, data_collector],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
    verbose=True
)

# クルーを実行
result = crew.kickoff(inputs={
    "url": "https://example.com/data",
    "site_key": "6LeIxAcTAAAAAJcZVRqyHh71UMIEGNQ_MXjiZKhI"
})
print(result)

@tool("reCAPTCHA v3を解決")
def solve_recaptcha_v3(website_url: str, website_key: str, page_action: str = "verify") -> str:
    """reCAPTCHA v3チャレンジを高スコアトークンで解決します。
    サイトが非表示のreCAPTCHA v3（スコアベース、チェックボックスなし）を使用している場合に使用してください。
    
    引数:
        website_url: ページの完全なURL
        website_key: reCAPTCHAサイトキー
        page_action: v3スコアリング用のアクション名（デフォルト: 'verify'）
    """
    result = asyncio.run(executor.execute("solve_captcha", {
        "captcha_type": "reCaptchaV3",
        "website_url": website_url,
        "website_key": website_key,
        "page_action": page_action,
        "min_score": 0.7
    }))
    if result["success"]:
        return f"reCAPTCHA v3は解決されました。トークン: {result['solution']['token']}"
    return f"失敗: {result['error']}"

@tool("Cloudflare Turnstileを解決")
def solve_turnstile(website_url: str, website_key: str) -> str:
    """Cloudflare Turnstileチャレンジを解決します。
    サイトがCloudflareで保護されており、Turnstileウィジェットが表示されている場合に使用してください。
    
    引数:
        website_url: ページの完全なURL
        website_key: Turnstileサイトキー（0xで始まる）
    """
    result = asyncio.run(executor.execute("solve_captcha", {
        "captcha_type": "cloudflare",
        "website_url": website_url,
        "website_key": website_key
    }))
    if result["success"]:
        return f"Turnstileは解決されました。トークン: {result['solution']['token']}"
    return f"失敗: {result['error']}"

@tool("失敗時に自動リトライするCAPTCHAを解決")
def solve_captcha_robust(captcha_type: str, website_url: str, website_key: str) -> str:
    """失敗時に自動リトライするCAPTCHAを解決します。
    失敗した場合、最大3回まで試行します。
    """
    for attempt in range(3):
        result = asyncio.run(executor.execute("solve_captcha", {
            "captcha_type": captcha_type,
            "website_url": website_url,
            "website_key": website_key
        }))
        if result["success"]:
            return f"試行{attempt + 1}で解決しました。トークン: {result['solution']['token']}"
        if attempt < 2:
            import time
            time.sleep(3)
    return f"3回の試行後に失敗しました: {result.get('error', '不明なエラー')}"

@tool("CAPTCHAソルバー残高を確認")
def check_balance() -> str:
    """CAPTCHAソルバーの残高を確認します。"""
    result = asyncio.run(executor.execute("get_balance", {}))
    if result["success"]:
        return f"残高: ${result['balance']:.2f}"
    return "残高を確認できませんでした"