How to Solve Captcha in Agno with CapSolver Integration

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

# And many more...

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:
    """
    Solves CAPTCHA challenges using CapSolver API.

    Args:
        website_url: The URL of the website with CAPTCHA
        website_key: The site key of the CAPTCHA
        captcha_type: Type of CAPTCHA (ReCaptchaV2TaskProxyLess, ReCaptchaV3TaskProxyLess, AntiTurnstileTaskProxyLess)

    Returns:
        The CAPTCHA solution token
    """
    # Create task
    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"Error creating task: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # Poll for result
    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"Task failed: {result.get('errorDescription')}"

    return "Timeout waiting for CAPTCHA solution"

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:
    """
    Solves reCAPTCHA v2 challenges using CapSolver.

    Args:
        website_url: The URL of the website with reCAPTCHA v2
        website_key: The site key (data-sitekey attribute)

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

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

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

    task_id = result.get("taskId")

    # Poll for result
    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"Failed: {result.get('errorDescription')}"

    return "Timeout waiting for solution"

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:
    """
    Solves reCAPTCHA v3 challenges with score-based verification.

    Args:
        website_url: The URL of the website with reCAPTCHA v3
        website_key: The site key of the reCAPTCHA
        page_action: The action parameter for reCAPTCHA v3
        min_score: Minimum score required (0.1 to 0.9)

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

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

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

    task_id = result.get("taskId")

    # Poll for result
    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"Failed: {result.get('errorDescription')}"

    return "Timeout waiting for solution"

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:
    """
    Solves Cloudflare Turnstile challenges.

    Args:
        website_url: The URL of the website with Turnstile
        website_key: The site key of the Turnstile widget

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

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

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

    task_id = result.get("taskId")

    # Poll for result
    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"Failed: {result.get('errorDescription')}"

    return "Timeout waiting for solution"

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:
    """
    Solves Cloudflare 5-second challenge pages.

    Args:
        website_url: The URL of the protected page
        proxy: Proxy in format: http://user:pass@ip:port

    Returns:
        Dictionary with cookies and user_agent for accessing the protected page
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AntiCloudflareTask",
            "websiteURL": website_url,
            "proxy": proxy
        }
    }

    # Create task
    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")

    # Poll for result
    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": "Timeout waiting for solution"}

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

# Configuration
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:
    """
    Solves reCAPTCHA v2 challenges using CapSolver.

    Args:
        website_url: The URL of the website with reCAPTCHA v2
        website_key: The site key (data-sitekey attribute)

    Returns:
        The g-recaptcha-response token or error message
    """
    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"Error: {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"Failed: {result.get('errorDescription')}"

    return "Timeout"


@tool
def solve_turnstile(website_url: str, website_key: str) -> str:
    """
    Solves Cloudflare Turnstile challenges.

    Args:
        website_url: The URL of the website with Turnstile
        website_key: The site key of the Turnstile widget

    Returns:
        The Turnstile token or error message
    """
    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"Error: {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"Failed: {result.get('errorDescription')}"

    return "Timeout"


@tool
def check_capsolver_balance() -> str:
    """
    Checks the current CapSolver account balance.

    Returns:
        Current balance information
    """
    response = requests.post(
        "https://api.capsolver.com/getBalance",
        json={"clientKey": CAPSOLVER_API_KEY}
    )
    result = response.json()

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

    return f"Balance: ${result.get('balance', 0):.4f}"


# Create the Web Scraper Agent
web_scraper_agent = Agent(
    name="Web Scraper Agent",
    model=OpenAIChat(id="gpt-4o"),
    tools=[solve_recaptcha_v2, solve_turnstile, check_capsolver_balance],
    description="Expert web scraper that handles CAPTCHA challenges automatically",
    instructions=[
        "You are a web scraping specialist with CAPTCHA solving capabilities.",
        "When encountering a CAPTCHA, identify the type and use the appropriate solver.",
        "For reCAPTCHA v2, use solve_recaptcha_v2 with the URL and site key.",
        "For Turnstile, use solve_turnstile with the URL and site key.",
        "Always check the balance before starting large scraping jobs."
    ],
    markdown=True
)


def main():
    print("=" * 60)
    print("Agno + CapSolver Integration Demo")
    print("=" * 60)

    # Task: Solve a reCAPTCHA challenge
    task = """
    I need you to solve a reCAPTCHA v2 challenge.

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

    Please solve this CAPTCHA and report the first 50 characters of the token.
    Also check my CapSolver balance before starting.
    """

    response = web_scraper_agent.run(task)
    print("\nAgent Response:")
    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:
    """Universal CAPTCHA solver supporting multiple types."""
    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"Error: {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"Failed: {result.get('errorDescription')}"

    return "Timeout"


# CAPTCHA Specialist Agent
captcha_agent = Agent(
    name="CAPTCHA Specialist",
    model=OpenAIChat(id="gpt-4o"),
    tools=[solve_any_captcha],
    description="Expert at identifying and solving various CAPTCHA types",
    instructions=[
        "Identify the CAPTCHA type from page analysis",
        "Use appropriate solver with correct parameters",
        "Report success or failure clearly"
    ]
)

# Data Extraction Agent
data_agent = Agent(
    name="Data Extractor",
    model=OpenAIChat(id="gpt-4o"),
    description="Extracts and processes data from web pages",
    instructions=[
        "Extract structured data from HTML content",
        "Request CAPTCHA solving when needed",
        "Validate and clean extracted data"
    ]
)

# Create the team
scraping_team = Team(
    name="Web Scraping Team",
    agents=[captcha_agent, data_agent],
    description="Team specialized in web scraping with CAPTCHA handling"
)

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

def submit_recaptcha_token(driver, token: str):
    """Inject reCAPTCHA token and submit"""
    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):
    """Inject Turnstile token and submit"""
    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):
    """Use Cloudflare Challenge solution to access protected page."""
    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:
    """Solve CAPTCHA with automatic retry on failure."""
    for attempt in range(max_retries):
        try:
            result = solve_recaptcha_v2(website_url, website_key)
            if not result.startswith("Error") and not result.startswith("Failed"):
                return result
        except Exception as e:
            if attempt == max_retries - 1:
                return f"All retries failed: {str(e)}"
        time.sleep(2 ** attempt)  # Exponential backoff

    return "Max retries exceeded"

@tool
def check_balance() -> float:
    """Check CapSolver account balance."""
    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:
    """Async CAPTCHA solver for better concurrency."""
    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"Error: {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"Failed: {result.get('errorDescription')}"

        return "Timeout"