Cách giải Captcha trong Agno với tích hợp CapSolver

pip install agno
pip install requests

# OpenAI
from agno.models.openai import OpenAIChat

# Anthropic Claude
from agno.models.anthropic import Claude

# Google Gemini
from agno.models.google import Gemini

# Và nhiều hơn nữa...

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


@tool
def solve_captcha(
    website_url: str,
    website_key: str,
    captcha_type: str = "ReCaptchaV2TaskProxyLess"
) -> str:
    """
    Giải các thách thức CAPTCHA bằng API CapSolver.

    Args:
        website_url: URL của trang web có CAPTCHA
        website_key: Khóa trang của CAPTCHA
        captcha_type: Loại CAPTCHA (ReCaptchaV2TaskProxyLess, ReCaptchaV3TaskProxyLess, AntiTurnstileTaskProxyLess)

    Returns:
        Token giải CAPTCHA
    """
    # Tạo nhiệm vụ
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": captcha_type,
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi tạo nhiệm vụ: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # Kiểm tra kết quả
    for _ in range(60):
        time.sleep(2)

        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            solution = result.get("solution", {})
            return solution.get("gRecaptchaResponse") or solution.get("token")
        elif result.get("status") == "failed":
            return f"Nhiệm vụ thất bại: {result.get('errorDescription')}"

    return "Đợi giải CAPTCHA hết thời gian"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


@tool
def solve_recaptcha_v2(website_url: str, website_key: str) -> str:
    """
    Giải các thách thức reCAPTCHA v2 bằng CapSolver.

    Args:
        website_url: URL của trang web có reCAPTCHA v2
        website_key: Khóa trang (thuộc tính data-sitekey)

    Returns:
        Token g-recaptcha-response
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    # Tạo nhiệm vụ
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # Kiểm tra kết quả
    for attempt in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["gRecaptchaResponse"]
        if result.get("status") == "failed":
            return f"Thất bại: {result.get('errorDescription')}"

    return "Đợi giải pháp hết thời gian"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


@tool
def solve_recaptcha_v3(
    website_url: str,
    website_key: str,
    page_action: str = "submit",
    min_score: float = 0.7
) -> str:
    """
    Giải các thách thức reCAPTCHA v3 với xác minh dựa trên điểm số.

    Args:
        website_url: URL của trang web có reCAPTCHA v3
        website_key: Khóa trang của reCAPTCHA
        page_action: Tham số hành động cho reCAPTCHA v3
        min_score: Điểm số tối thiểu yêu cầu (0.1 đến 0.9)

    Returns:
        Token g-recaptcha-response
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "ReCaptchaV3TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key,
            "pageAction": page_action,
            "minScore": min_score
        }
    }

    # Tạo nhiệm vụ
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # Kiểm tra kết quả
    for attempt in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["gRecaptchaResponse"]
        if result.get("status") == "failed":
            return f"Thất bại: {result.get('errorDescription')}"

    return "Đợi giải pháp hết thời gian"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


@tool
def solve_turnstile(website_url: str, website_key: str) -> str:
    """
    Giải các thách thức Cloudflare Turnstile.

    Args:
        website_url: URL của trang web có Turnstile
        website_key: Khóa trang của widget Turnstile

    Returns:
        Token Turnstile
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AntiTurnstileTaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    # Tạo nhiệm vụ
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # Kiểm tra kết quả
    for attempt in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["token"]
        if result.get("status") == "failed":
            return f"Thất bại: {result.get('errorDescription')}"

    return "Đợi giải pháp hết thời gian"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


@tool
def solve_cloudflare_challenge(website_url: str, proxy: str) -> dict:
    """
    Giải các trang bảo vệ Cloudflare 5 giây.

    Args:
        website_url: URL của trang được bảo vệ
        proxy: Proxy theo định dạng: http://user:pass@ip:port

    Returns:
        Từ điển chứa cookie và user_agent để truy cập trang được bảo vệ
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AntiCloudflareTask",
            "websiteURL": website_url,
            "proxy": proxy
        }
    }

    # Tạo nhiệm vụ
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return {"error": result.get('errorDescription')}

    task_id = result.get("taskId")

    # Kiểm tra kết quả
    for attempt in range(60):
        time.sleep(3)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return {
                "cookies": result["solution"]["cookies"],
                "user_agent": result["solution"]["userAgent"]
            }
        if result.get("status") == "failed":
            return {"error": result.get('errorDescription')}

    return {"error": "Đợi giải pháp hết thời gian"}

import os
import requests
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool

# Cấu hình
CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"


@tool
def solve_recaptcha_v2(website_url: str, website_key: str) -> str:
    """
    Giải các thách thức reCAPTCHA v2 bằng CapSolver.

    Args:
        website_url: URL của trang web có reCAPTCHA v2
        website_key: Khóa trang (thuộc tính data-sitekey)

    Returns:
        Token g-recaptcha-response hoặc thông báo lỗi
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    for _ in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["gRecaptchaResponse"]
        if result.get("status") == "failed":
            return f"Thất bại: {result.get('errorDescription')}"

    return "Hết thời gian chờ giải pháp"


@tool
def solve_turnstile(website_url: str, website_key: str) -> str:
    """
    Giải các thách thức Cloudflare Turnstile.

    Args:
        website_url: URL của trang web có Turnstile
        website_key: Khóa trang của widget Turnstile

    Returns:
        Token Turnstile hoặc thông báo lỗi
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AntiTurnstileTaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    for _ in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["token"]
        if result.get("status") == "failed":
            return f"Thất bại: {result.get('errorDescription')}"

    return "Hết thời gian chờ giải pháp"
"https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["token"]
        if result.get("status") == "failed":
            return f"Thất bại: {result.get('errorDescription')}"

    return "Hết thời gian"

@tool
def check_capsolver_balance() -> str:
    """
    Kiểm tra số dư tài khoản CapSolver hiện tại.

    Trả về:
        Thông tin số dư hiện tại
    """
    response = requests.post(
        "https://api.capsolver.com/getBalance",
        json={"clientKey": CAPSOLVER_API_KEY}
    )
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi: {result.get('errorDescription')}"

    return f"Số dư: ${result.get('balance', 0):.4f}"


# Tạo Agent Dò tìm Trang Web
web_scraper_agent = Agent(
    name="Agent Dò tìm Trang Web",
    model=OpenAIChat(id="gpt-4o"),
    tools=[solve_recaptcha_v2, solve_turnstile, check_capsolver_balance],
    description="Chuyên gia dò tìm trang web có khả năng xử lý CAPTCHA tự động",
    instructions=[
        "Bạn là chuyên gia dò tìm trang web với khả năng xử lý CAPTCHA.",
        "Khi gặp CAPTCHA, xác định loại và sử dụng bộ giải thích hợp.",
        "Đối với reCAPTCHA v2, sử dụng solve_recaptcha_v2 với URL và site key.",
        "Đối với Turnstile, sử dụng solve_turnstile với URL và site key.",
        "Luôn kiểm tra số dư trước khi bắt đầu các công việc dò tìm lớn."
    ],
    markdown=True
)


def main():
    print("=" * 60)
    print("Demo Tích hợp Agno + CapSolver")
    print("=" * 60)

    # Nhiệm vụ: Giải quyết thách thức CAPTCHA
    task = """
    Tôi cần bạn giải quyết thách thức CAPTCHA reCAPTCHA v2.

    URL Trang web: https://www.google.com/recaptcha/api2/demo
    Site Key: 6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-

    Vui lòng giải CAPTCHA này và báo cáo 50 ký tự đầu tiên của token.
    Đồng thời kiểm tra số dư CapSolver của tôi trước khi bắt đầu.
    """

    response = web_scraper_agent.run(task)
    print("\nPhản hồi của Agent:")
    print(response.content)


if __name__ == "__main__":
    main()

from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat
from agno.tools import tool
import requests
import time

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


@tool
def solve_any_captcha(
    website_url: str,
    website_key: str,
    captcha_type: str = "ReCaptchaV2TaskProxyLess"
) -> str:
    """Bộ giải CAPTCHA toàn diện hỗ trợ nhiều loại."""
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": captcha_type,
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"Lỗi: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    for _ in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            solution = result.get("solution", {})
            return solution.get("gRecaptchaResponse") or solution.get("token")
        if result.get("status") == "failed":
            return f"Thất bại: {result.get('errorDescription')}"

    return "Hết thời gian"


# Tác nhân Chuyên gia CAPTCHA
captcha_agent = Agent(
    name="Tác nhân Chuyên gia CAPTCHA",
    model=OpenAIChat(id="gpt-4o"),
    tools=[solve_any_captcha],
    description="Chuyên gia nhận diện và giải CAPTCHA các loại",
    instructions=[
        "Nhận diện loại CAPTCHA từ phân tích trang",
        "Sử dụng bộ giải thích hợp với tham số đúng",
        "Báo cáo thành công hoặc thất bại một cách rõ ràng"
    ]
)

# Tác nhân Trích xuất Dữ liệu
data_agent = Agent(
    name="Tác nhân Trích xuất Dữ liệu",
    model=OpenAIChat(id="gpt-4o"),
    description="Trích xuất và xử lý dữ liệu từ trang web",
    instructions=[
        "Trích xuất dữ liệu có cấu trúc từ nội dung HTML",
        "Yêu cầu giải CAPTCHA khi cần",
        "Xác minh và làm sạch dữ liệu trích xuất"
    ]
)

# Tạo nhóm
scraping_team = Team(
    name="Nhóm Dò tìm Trang Web",
    agents=[captcha_agent, data_agent],
    description="Nhóm chuyên về dò tìm trang web với xử lý CAPTCHA"
)

from selenium import webdriver
from selenium.webdriver.common.by import By

def submit_recaptcha_token(driver, token: str):
    """Chèn token reCAPTCHA và gửi"""
    recaptcha_response = driver.find_element(By.ID, "g-recaptcha-response")
    driver.execute_script("arguments[0].style.display = 'block';", recaptcha_response)
    recaptcha_response.clear()
    recaptcha_response.send_keys(token)

    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

def submit_turnstile_token(driver, token: str):
    """Chèn token Turnstile và gửi"""
    turnstile_input = driver.find_element(By.NAME, "cf-turnstile-response")
    driver.execute_script("arguments[0].value = arguments[1];", turnstile_input, token)

    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

import requests

def access_cloudflare_protected_page(url: str, cf_solution: dict):
    """Sử dụng giải pháp Cloudflare Challenge để truy cập trang bị bảo vệ."""
    session = requests.Session()

    for cookie in cf_solution["cookies"]:
        session.cookies.set(cookie["name"], cookie["value"])

    headers = {"User-Agent": cf_solution["user_agent"]}

    response = session.get(url, headers=headers)
    return response.text

from agno.tools import tool
import time

@tool
def solve_with_retry(
    website_url: str,
    website_key: str,
    max_retries: int = 3
) -> str:
    """Giải CAPTCHA với thử lại tự động khi thất bại."""
    for attempt in range(max_retries):
        try:
            result = solve_recaptcha_v2(website_url, website_key)
            if not result.startswith("Lỗi") and not result.startswith("Thất bại"):
                return result
        except Exception as e:
            if attempt == max_retries - 1:
                return f"Tất cả các lần thử đều thất bại: {str(e)}"
        time.sleep(2 ** attempt)  # Backoff theo cấp số nhân

    return "Quá số lần thử tối đa"

@tool
def check_balance() -> float:
    """Kiểm tra số dư tài khoản CapSolver."""
    response = requests.post(
        "https://api.capsolver.com/getBalance",
        json={"clientKey": CAPSOLVER_API_KEY}
    )
    return response.json().get("balance", 0)

import asyncio
import aiohttp
from agno.tools import tool

@tool
async def solve_captcha_async(website_url: str, website_key: str) -> str:
    """Bộ giải CAPTCHA async để cải thiện khả năng đồng thời."""
    async with aiohttp.ClientSession() as session:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "ReCaptchaV2TaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }

        async with session.post(
            "https://api.capsolver.com/createTask",
            json=payload
        ) as response:
            result = await response.json()

        if result.get("errorId") != 0:
            return f"Lỗi: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        for _ in range(60):
            await asyncio.sleep(2)

            async with session.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ) as response:
                result = await response.json()

            if result.get("status") == "ready":
                solution = result.get("solution", {})
                return solution.get("gRecaptchaResponse") or solution.get("token")
            if result.get("status") == "failed":
                return f"Thất bại: {result.get('errorDescription')}"

        return "Hết thời gian"