How to Integrate Camoufox with CapSolver for Seamless CAPTCHA Solving

# Install the Python package
pip install -U camoufox[geoip]

# Download the Camoufox browser
camoufox fetch

from camoufox.sync_api import Camoufox

with Camoufox(humanize=True) as browser:
    page = browser.new_page()
    page.goto("https://example.com")

pip install camoufox[geoip] httpx

import asyncio
import httpx
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def create_task(task_payload: dict) -> str:
    """Create a CAPTCHA solving task and return the task ID."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": task_payload
            }
        )
        result = response.json()
        if result.get("errorId") != 0:
            raise Exception(f"CapSolver error: {result.get('errorDescription')}")
        return result["taskId"]


async def get_task_result(task_id: str, max_attempts: int = 120) -> dict:
    """Poll for task result until solved or timeout."""
    async with httpx.AsyncClient() as client:
        for _ in range(max_attempts):
            response = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            result = response.json()

            if result.get("status") == "ready":
                return result["solution"]
            elif result.get("status") == "failed":
                raise Exception(f"Task failed: {result.get('errorDescription')}")

            await asyncio.sleep(1)

    raise TimeoutError("CAPTCHA solving timed out")


async def solve_captcha(task_payload: dict) -> dict:
    """Complete CAPTCHA solving workflow."""
    task_id = await create_task(task_payload)
    return await get_task_result(task_id)

from camoufox.sync_api import Camoufox

with Camoufox(
    addons=["/path/to/capsolver-extension"],
    headless=False  # Extensions require headed mode
) as browser:
    page = browser.new_page()
    # The extension will automatically detect and solve CAPTCHAs

import asyncio
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def solve_turnstile(site_key: str, page_url: str) -> str:
    """Solve Cloudflare Turnstile and return the token."""
    import httpx

    async with httpx.AsyncClient() as client:
        # Create the task
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": {
                    "type": "AntiTurnstileTaskProxyLess",
                    "websiteURL": page_url,
                    "websiteKey": site_key,
                }
            }
        )
        task_id = response.json()["taskId"]

        # Poll for result
        while True:
            result = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            data = result.json()

            if data.get("status") == "ready":
                return data["solution"]["token"]

            await asyncio.sleep(1)


async def main():
    target_url = "https://example.com/protected-page"
    turnstile_site_key = "0x4XXXXXXXXXXXXXXXXX"  # Find this in page source

    async with AsyncCamoufox(
        humanize=True,
        headless=False,
        os="windows"
    ) as browser:
        page = await browser.new_page()
        await page.goto(target_url)

        # Wait for Turnstile to load
        await page.wait_for_selector('input[name="cf-turnstile-response"]', timeout=10000)

        # Solve the CAPTCHA
        token = await solve_turnstile(turnstile_site_key, target_url)
        print(f"Got Turnstile token: {token[:50]}...")

        # Inject the token
        await page.evaluate(f'''
            document.querySelector('input[name="cf-turnstile-response"]').value = "{token}";

            // Also set the hidden callback if present
            const callback = document.querySelector('[data-callback]');
            if (callback) {{
                const callbackName = callback.getAttribute('data-callback');
                if (window[callbackName]) {{
                    window[callbackName]('{token}');
                }}
            }}
        ''')

        # Submit the form
        await page.click('button[type="submit"]')
        await page.wait_for_load_state("networkidle")

        print("Successfully bypassed Turnstile!")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def solve_recaptcha_v2(site_key: str, page_url: str) -> str:
    """Solve reCAPTCHA v2 and return the token."""
    import httpx

    async with httpx.AsyncClient() as client:
        # Create the task
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": {
                    "type": "ReCaptchaV2TaskProxyLess",
                    "websiteURL": page_url,
                    "websiteKey": site_key,
                }
            }
        )
        result = response.json()

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

        task_id = result["taskId"]

        # Poll for result
        while True:
            result = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            data = result.json()

            if data.get("status") == "ready":
                return data["solution"]["gRecaptchaResponse"]
            elif data.get("status") == "failed":
                raise Exception(f"Failed: {data.get('errorDescription')}")

            await asyncio.sleep(2)


async def main():
    target_url = "https://example.com/login"
    recaptcha_site_key = "6LcXXXXXXXXXXXXXXXXXXXXXXXXX"  # Find in page source

    async with AsyncCamoufox(
        humanize=True,
        headless=False,
        os=["windows", "macos"]  # Random OS selection
    ) as browser:
        page = await browser.new_page()
        await page.goto(target_url)

        # Fill in form fields with human-like delays
        await page.fill('input[name="username"]', "user@example.com")
        await asyncio.sleep(0.5)  # Human-like pause
        await page.fill('input[name="password"]', "password123")

        # Solve the CAPTCHA
        print("Solving reCAPTCHA v2...")
        token = await solve_recaptcha_v2(recaptcha_site_key, target_url)
        print(f"Got token: {token[:50]}...")

        # Inject the token into the reCAPTCHA response field
        await page.evaluate(f'''
            document.getElementById('g-recaptcha-response').innerHTML = '{token}';
            document.getElementById('g-recaptcha-response').style.display = 'block';
        ''')

        # Submit the form
        await page.click('button[type="submit"]')
        await page.wait_for_load_state("networkidle")

        print("Login successful!")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def solve_recaptcha_v3(
    site_key: str,
    page_url: str,
    action: str = "verify",
    min_score: float = 0.7
) -> str:
    """Solve reCAPTCHA v3 with specified action and minimum score."""
    import httpx

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": {
                    "type": "ReCaptchaV3TaskProxyLess",
                    "websiteURL": page_url,
                    "websiteKey": site_key,
                    "pageAction": action,
                    "minScore": min_score
                }
            }
        )
        result = response.json()

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

        task_id = result["taskId"]

        while True:
            result = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            data = result.json()

            if data.get("status") == "ready":
                return data["solution"]["gRecaptchaResponse"]
            elif data.get("status") == "failed":
                raise Exception(f"Failed: {data.get('errorDescription')}")

            await asyncio.sleep(1)


async def main():
    target_url = "https://example.com/search"
    recaptcha_v3_key = "6LcXXXXXXXXXXXXXXXXXXXXXXXXX"

    async with AsyncCamoufox(
        humanize=2.0,  # Max 2 second humanized movements
        headless=True,  # Can run headless for v3
        geoip=True,  # Auto-detect geolocation from proxy
    ) as browser:
        page = await browser.new_page()
        await page.goto(target_url)

        # Solve reCAPTCHA v3 with "search" action
        print("Solving reCAPTCHA v3...")
        token = await solve_recaptcha_v3(
            recaptcha_v3_key,
            target_url,
            action="search",
            min_score=0.9  # Request high score
        )

        # Execute the callback with the token
        await page.evaluate(f'''
            // Submit token via the site's callback
            grecaptcha.execute('{recaptcha_v3_key}', {{action: 'search'}})
                .then(function(originalToken) {{
                    // Replace with our solved token
                    submitSearch('{token}');
                }});
        ''')

        print("reCAPTCHA v3 bypassed!")


if __name__ == "__main__":
    asyncio.run(main())

async with AsyncCamoufox(
    geoip=True,  # Auto-detect from proxy IP
    proxy={
        "server": "http://proxy.example.com:8080",
        "username": "user",
        "password": "pass"
    }
) as browser:
    # Fingerprint will match proxy's geographic location
    pass

from browserforge.fingerprints import Screen

# Constrain to common screen sizes
screen = Screen(
    min_width=1280,
    max_width=1920,
    min_height=720,
    max_height=1080
)

async with AsyncCamoufox(
    os="windows",
    screen=screen,
) as browser:
    pass

import random

async def human_delay():
    """Random delay to mimic human behavior."""
    await asyncio.sleep(random.uniform(1.0, 3.0))

# Use between actions
await page.click('button')
await human_delay()
await page.fill('input', 'text')

async def solve_with_retry(task_payload: dict, max_retries: int = 3) -> dict:
    """Solve CAPTCHA with retry logic."""
    for attempt in range(max_retries):
        try:
            return await solve_captcha(task_payload)
        except TimeoutError:
            if attempt < max_retries - 1:
                print(f"Timeout, retrying... ({attempt + 1}/{max_retries})")
                await asyncio.sleep(5)
            else:
                raise
        except Exception as e:
            if "balance" in str(e).lower():
                raise  # Don't retry balance errors
            if attempt < max_retries - 1:
                await asyncio.sleep(2)
            else:
                raise