Python asyncio + CaptchaAI: Concurrent Solving at Scale

Sequential CAPTCHA solving caps throughput at one solve per cycle — around 20–30 seconds each. With Python asyncio and aiohttp, you can solve dozens of CAPTCHAs concurrently while keeping resource usage flat.

Prerequisites

pip install aiohttp

• Python 3.8+
• CaptchaAI API key from captchaai.com

Basic async solver

import asyncio
import aiohttp

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
API_KEY = "YOUR_API_KEY"


async def solve_captcha(session, method, params):
    """Submit a CAPTCHA task and poll for the result."""
    data = {
        "key": API_KEY,
        "method": method,
        "json": 1,
        **params
    }

    # Submit
    async with session.post(SUBMIT_URL, data=data) as resp:
        result = await resp.json(content_type=None)

    if result.get("status") != 1:
        raise Exception(f"Submit error: {result.get('error_text', result.get('request'))}")

    task_id = result["request"]

    # Poll
    for _ in range(24):  # 24 * 5s = 120s max
        await asyncio.sleep(5)
        async with session.get(RESULT_URL, params={
            "key": API_KEY,
            "action": "get",
            "id": task_id,
            "json": 1
        }) as resp:
            result = await resp.json(content_type=None)

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise Exception(f"Poll error: {result.get('error_text', result.get('request'))}")

    raise Exception(f"Timeout: task {task_id}")


async def main():
    async with aiohttp.ClientSession() as session:
        token = await solve_captcha(session, "userrecaptcha", {
            "googlekey": "6Le-SITEKEY",
            "pageurl": "https://example.com"
        })
        print(f"Token: {token[:50]}...")


asyncio.run(main())

Expected output:

Token: 03AGdBq26ZfPxL...

Concurrent solving with semaphore

Use asyncio.Semaphore to limit concurrency and prevent rate limit errors:

import asyncio
import aiohttp

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
API_KEY = "YOUR_API_KEY"

MAX_CONCURRENT = 10


async def solve_captcha(session, semaphore, method, params, label=""):
    async with semaphore:
        data = {
            "key": API_KEY,
            "method": method,
            "json": 1,
            **params
        }

        async with session.post(SUBMIT_URL, data=data) as resp:
            result = await resp.json(content_type=None)

        if result.get("status") != 1:
            error = result.get("error_text", result.get("request"))
            print(f"[{label}] Submit failed: {error}")
            return None

        task_id = result["request"]
        print(f"[{label}] Submitted: {task_id}")

        for _ in range(24):
            await asyncio.sleep(5)
            async with session.get(RESULT_URL, params={
                "key": API_KEY,
                "action": "get",
                "id": task_id,
                "json": 1
            }) as resp:
                result = await resp.json(content_type=None)

            if result.get("status") == 1:
                print(f"[{label}] Solved: {result['request'][:40]}...")
                return result["request"]
            if result.get("request") != "CAPCHA_NOT_READY":
                print(f"[{label}] Error: {result.get('error_text', result.get('request'))}")
                return None

        print(f"[{label}] Timeout")
        return None


async def main():
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)

    # 20 tasks — only 10 run concurrently
    tasks_params = [
        {"googlekey": f"6Le-SITEKEY-{i}", "pageurl": f"https://example.com/page/{i}"}
        for i in range(20)
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [
            solve_captcha(session, semaphore, "userrecaptcha", params, label=f"task-{i}")
            for i, params in enumerate(tasks_params)
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

    solved = [r for r in results if r and not isinstance(r, Exception)]
    failed = len(results) - len(solved)
    print(f"\nResults: {len(solved)} solved, {failed} failed")


asyncio.run(main())

Expected output:

[task-0] Submitted: 72345678901
[task-1] Submitted: 72345678902
...
[task-0] Solved: 03AGdBq26ZfPxL...
[task-1] Solved: 03AGdBq27AbCdE...
...
Results: 18 solved, 2 failed

Mixed CAPTCHA types

Solve different CAPTCHA types in the same batch:

async def solve_mixed_batch(session, semaphore):
    tasks = [
        solve_captcha(session, semaphore, "userrecaptcha", {
            "googlekey": "6Le-SITEKEY-A",
            "pageurl": "https://site-a.com"
        }, label="recaptcha-1"),

        solve_captcha(session, semaphore, "turnstile", {
            "sitekey": "0x4AAAA-SITEKEY-B",
            "pageurl": "https://site-b.com"
        }, label="turnstile-1"),

        solve_captcha(session, semaphore, "userrecaptcha", {
            "googlekey": "6Le-SITEKEY-C",
            "pageurl": "https://site-c.com"
        }, label="recaptcha-2"),
    ]

    return await asyncio.gather(*tasks, return_exceptions=True)

Error handling patterns

Retry with backoff

async def solve_with_retry(session, semaphore, method, params, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            return await solve_captcha(session, semaphore, method, params,
                                        label=f"attempt-{attempt}")
        except Exception as e:
            if "ERROR_ZERO_BALANCE" in str(e):
                raise  # Don't retry permanent errors
            if attempt < max_retries:
                wait = 2 ** attempt
                print(f"Retry in {wait}s: {e}")
                await asyncio.sleep(wait)
            else:
                raise

Collect results with error tracking

async def solve_batch_with_tracking(session, semaphore, task_list):
    results = {"solved": [], "failed": [], "errors": []}

    async def tracked_solve(task_params, index):
        try:
            token = await solve_captcha(session, semaphore,
                                         task_params["method"],
                                         task_params["params"],
                                         label=f"task-{index}")
            if token:
                results["solved"].append({"index": index, "token": token})
            else:
                results["failed"].append(index)
        except Exception as e:
            results["errors"].append({"index": index, "error": str(e)})

    tasks = [tracked_solve(t, i) for i, t in enumerate(task_list)]
    await asyncio.gather(*tasks)

    return results

Performance tuning

Optimized connector

connector = aiohttp.TCPConnector(limit=100, keepalive_timeout=30)
timeout = aiohttp.ClientTimeout(total=15)

async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
    # Use session for all solves
    pass

Troubleshooting

FAQ

How many concurrent tasks should I run?

Start with 10. Monitor for ERROR_NO_SLOT_AVAILABLE and adjust. Most accounts support 20+ concurrent tasks.

Does asyncio use multiple CPU cores?

No. asyncio is single-threaded — it handles I/O concurrency, not CPU parallelism. For CAPTCHA solving, the bottleneck is network I/O, so asyncio is ideal.

Can I cancel in-progress tasks?

Yes. Use task.cancel() on any asyncio task. The pending API requests will be abandoned, but CaptchaAI will still process them (you'll be charged).

Start solving CAPTCHAs concurrently with CaptchaAI

Get your API key at captchaai.com.

Related guides

• Parallel CAPTCHA Solving
• Retry Queue with Exponential Backoff
• Reusable CAPTCHA Modules
• Rate Limits and Throttling