Solve reCAPTCHA v3 with Python: Step-by-Step Tutorial

pip install httpx asyncio

import httpx
import time

API_KEY = "your-api-key"
BASE_URL = "https://api.regotcha.com"

def solve_recaptcha(website_url: str, website_key: str, action: str = "submit"):
    # Create task
    response = httpx.post(
        f"{BASE_URL}/createTask",
        json={
            "clientKey": API_KEY,
            "task": {
                "type": "ReCaptchaV3EnterpriseTaskProxyless",
                "websiteURL": website_url,
                "websiteKey": website_key,
                "pageAction": action
            }
        }
    )
    task_id = response.json()["taskId"]

    # Poll for result
    while True:
        result = httpx.post(
            f"{BASE_URL}/getTaskResult",
            json={"clientKey": API_KEY, "taskId": task_id}
        ).json()

        if result["status"] == "ready":
            return result["solution"]["gRecaptchaResponse"]

        time.sleep(2)

# Usage
token = solve_recaptcha(
    "https://example.com",
    "6LcR_RIpAAAAANGDwXhTz8nO4N_NijF_Pj2iO4L6",
    "login"
)
print(f"Token: {token}")

import httpx
import asyncio
from typing import Optional

class RecaptchaSolver:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.regotcha.com"

    async def solve(
        self,
        website_url: str,
        website_key: str,
        action: str = "submit",
        timeout: int = 120
    ) -> Optional[str]:
        async with httpx.AsyncClient() as client:
            # Create task
            response = await client.post(
                f"{self.base_url}/createTask",
                json={
                    "clientKey": self.api_key,
                    "task": {
                        "type": "ReCaptchaV3EnterpriseTaskProxyless",
                        "websiteURL": website_url,
                        "websiteKey": website_key,
                        "pageAction": action
                    }
                }
            )
            data = response.json()

            if data.get("errorId"):
                raise Exception(f"API Error: {data.get('errorDescription')}")

            task_id = data["taskId"]

            # Poll with timeout
            start = asyncio.get_event_loop().time()
            while asyncio.get_event_loop().time() - start < timeout:
                result = await client.post(
                    f"{self.base_url}/getTaskResult",
                    json={"clientKey": self.api_key, "taskId": task_id}
                )
                result_data = result.json()

                if result_data["status"] == "ready":
                    return result_data["solution"]["gRecaptchaResponse"]

                if result_data.get("errorId"):
                    raise Exception(f"Solve Error: {result_data.get('errorDescription')}")

                await asyncio.sleep(2)

            raise TimeoutError("CAPTCHA solving timed out")

# Usage
async def main():
    solver = RecaptchaSolver("your-api-key")
    token = await solver.solve(
        "https://example.com",
        "6LcR_RIpAAAAANGDwXhTz8nO4N_NijF_Pj2iO4L6",
        "checkout"
    )
    print(f"Token: {token}")

asyncio.run(main())

async def solve_batch(solver, tasks):
    """Solve multiple CAPTCHAs concurrently"""
    return await asyncio.gather(
        *[solver.solve(**task) for task in tasks],
        return_exceptions=True
    )

# Example: Solve 10 CAPTCHAs in parallel
tasks = [
    {"website_url": "https://site1.com", "website_key": "key1", "action": "login"},
    {"website_url": "https://site2.com", "website_key": "key2", "action": "signup"},
    # ... more tasks
]

results = await solve_batch(solver, tasks)