Async CAPTCHA Solving with Python asyncio and CaptchaAI

When you need to solve multiple CAPTCHAs simultaneously, Python's asyncio with aiohttp lets you submit and poll dozens of tasks concurrently without blocking. This guide shows async patterns for CaptchaAI that can solve 10-50+ CAPTCHAs in parallel.

Prerequisites

pip install aiohttp

Python 3.7+ includes asyncio in the standard library.

Why async?

With asyncio, all tasks are submitted immediately and polled concurrently. Total time ≈ the slowest single solve, not the sum.

Basic async solver

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"
API_BASE = "https://ocr.captchaai.com"


async def solve_recaptcha_async(session, sitekey, page_url):
    """Solve a single reCAPTCHA v2 asynchronously."""
    # Submit task
    async with session.post(f"{API_BASE}/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": page_url,
        "json": 1,
    }) as response:
        data = await response.json()

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

    task_id = data["request"]

    # Poll for result
    for _ in range(30):
        await asyncio.sleep(5)

        async with session.get(f"{API_BASE}/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": task_id,
            "json": 1,
        }) as response:
            result = await response.json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
            raise Exception("Unsolvable")

    raise TimeoutError("Solve timed out")


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


asyncio.run(main())

Parallel solving (multiple CAPTCHAs)

The real power of async is solving many CAPTCHAs concurrently:

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"
API_BASE = "https://ocr.captchaai.com"


async def solve_captcha(session, task_config):
    """Generic async CAPTCHA solver for any type."""
    # Submit
    submit_data = {
        "key": API_KEY,
        "json": 1,
        **task_config,
    }

    async with session.post(f"{API_BASE}/in.php", data=submit_data) as response:
        data = await response.json()

    if data.get("status") != 1:
        return {"success": False, "error": data.get("request"), "config": task_config}

    task_id = data["request"]

    # Poll
    for _ in range(30):
        await asyncio.sleep(5)

        async with session.get(f"{API_BASE}/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": task_id,
            "json": 1,
        }) as response:
            result = await response.json()

        if result.get("status") == 1:
            return {"success": True, "token": result["request"], "config": task_config}
        if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
            return {"success": False, "error": "unsolvable", "config": task_config}

    return {"success": False, "error": "timeout", "config": task_config}


async def solve_batch(tasks):
    """Solve multiple CAPTCHAs concurrently."""
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            *[solve_captcha(session, task) for task in tasks],
            return_exceptions=True,
        )
    return results


# Define batch of tasks
tasks = [
    {
        "method": "userrecaptcha",
        "googlekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
        "pageurl": "https://site-a.com/login",
    },
    {
        "method": "turnstile",
        "sitekey": "0x4AAAAAAAC3DHQhMMQ_Rxrg",
        "pageurl": "https://site-b.com/signup",
    },
    {
        "method": "userrecaptcha",
        "googlekey": "6LcR_okUAAAAAPYrPe-HK_0RULO1aZM15ENyM-Mf",
        "pageurl": "https://site-c.com/form",
    },
]

# Solve all concurrently
results = asyncio.run(solve_batch(tasks))

for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"Task {i}: Error - {result}")
    elif result["success"]:
        print(f"Task {i}: Solved - {result['token'][:40]}...")
    else:
        print(f"Task {i}: Failed - {result['error']}")

Async solver class with concurrency control

For production use, control the number of simultaneous solve tasks:

import asyncio
import aiohttp
import time

API_KEY = "YOUR_API_KEY"
API_BASE = "https://ocr.captchaai.com"


class AsyncCaptchaSolver:
    """Production async CAPTCHA solver with concurrency control."""

    def __init__(self, api_key, max_concurrent=20, max_retries=2):
        self.api_key = api_key
        self.max_concurrent = max_concurrent
        self.max_retries = max_retries
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.stats = {"submitted": 0, "solved": 0, "failed": 0}

    async def solve(self, session, task_config):
        """Solve a single CAPTCHA with semaphore-controlled concurrency."""
        async with self.semaphore:
            for attempt in range(1, self.max_retries + 1):
                try:
                    result = await self._solve_once(session, task_config)
                    if result["success"]:
                        self.stats["solved"] += 1
                        return result
                except Exception as e:
                    if attempt == self.max_retries:
                        self.stats["failed"] += 1
                        return {"success": False, "error": str(e)}

            self.stats["failed"] += 1
            return {"success": False, "error": "max retries exceeded"}

    async def _solve_once(self, session, task_config):
        """Single solve attempt."""
        self.stats["submitted"] += 1

        # Submit
        submit_data = {"key": self.api_key, "json": 1, **task_config}

        async with session.post(f"{API_BASE}/in.php", data=submit_data) as resp:
            data = await resp.json()

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

        task_id = data["request"]

        # Poll
        for _ in range(30):
            await asyncio.sleep(5)

            async with session.get(f"{API_BASE}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            }) as resp:
                result = await resp.json()

            if result.get("status") == 1:
                return {"success": True, "token": result["request"]}
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise Exception("Unsolvable")

        raise TimeoutError("Poll timed out")

    async def solve_batch(self, tasks):
        """Solve a batch of tasks with controlled concurrency."""
        async with aiohttp.ClientSession() as session:
            coroutines = [self.solve(session, task) for task in tasks]
            results = await asyncio.gather(*coroutines, return_exceptions=True)
        return results

    def get_stats(self):
        return self.stats.copy()


# Usage
async def main():
    solver = AsyncCaptchaSolver(API_KEY, max_concurrent=10)

    tasks = [
        {
            "method": "turnstile",
            "sitekey": "0x4AAAAAAAC3DHQhMMQ_Rxrg",
            "pageurl": f"https://example.com/page/{i}",
        }
        for i in range(20)
    ]

    start = time.time()
    results = await solver.solve_batch(tasks)
    elapsed = time.time() - start

    solved = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
    print(f"Solved {solved}/{len(tasks)} in {elapsed:.1f}s")
    print(f"Stats: {solver.get_stats()}")


asyncio.run(main())

Producer-consumer pattern

For continuous CAPTCHA solving (e.g., in a scraping pipeline):

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"
API_BASE = "https://ocr.captchaai.com"


async def captcha_producer(queue, urls_with_sitekeys):
    """Produce CAPTCHA tasks from URLs."""
    for url, sitekey, captcha_type in urls_with_sitekeys:
        await queue.put({
            "url": url,
            "sitekey": sitekey,
            "type": captcha_type,
        })
    # Signal end
    await queue.put(None)


async def captcha_solver(queue, result_queue, session, worker_id):
    """Consume and solve CAPTCHAs."""
    while True:
        task = await queue.get()
        if task is None:
            await queue.put(None)  # Pass sentinel to next worker
            break

        try:
            method_map = {
                "recaptcha_v2": ("userrecaptcha", "googlekey"),
                "turnstile": ("turnstile", "sitekey"),
            }

            method, key_param = method_map.get(
                task["type"], ("userrecaptcha", "googlekey")
            )

            submit_data = {
                "key": API_KEY,
                "method": method,
                key_param: task["sitekey"],
                "pageurl": task["url"],
                "json": 1,
            }

            async with session.post(f"{API_BASE}/in.php", data=submit_data) as resp:
                data = await resp.json()

            if data.get("status") != 1:
                await result_queue.put({
                    "url": task["url"],
                    "success": False,
                    "error": data.get("request"),
                })
                continue

            task_id = data["request"]

            # Poll
            token = None
            for _ in range(30):
                await asyncio.sleep(5)
                async with session.get(f"{API_BASE}/res.php", params={
                    "key": API_KEY,
                    "action": "get",
                    "id": task_id,
                    "json": 1,
                }) as resp:
                    result = await resp.json()

                if result.get("status") == 1:
                    token = result["request"]
                    break
                if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                    break

            await result_queue.put({
                "url": task["url"],
                "success": token is not None,
                "token": token,
            })

        except Exception as e:
            await result_queue.put({
                "url": task["url"],
                "success": False,
                "error": str(e),
            })


async def result_processor(result_queue, total):
    """Process solved CAPTCHAs."""
    processed = 0
    while processed < total:
        result = await result_queue.get()
        processed += 1
        status = "OK" if result["success"] else f"FAIL: {result.get('error', 'unknown')}"
        print(f"[{processed}/{total}] {result['url']}: {status}")


async def main():
    urls = [
        ("https://site1.com/login", "6Le-wvkSAAAA...", "recaptcha_v2"),
        ("https://site2.com/signup", "0x4AAAAAAAC3...", "turnstile"),
        # ... more URLs
    ]

    task_queue = asyncio.Queue()
    result_queue = asyncio.Queue()

    async with aiohttp.ClientSession() as session:
        # Start workers
        workers = [
            asyncio.create_task(
                captcha_solver(task_queue, result_queue, session, i)
            )
            for i in range(5)  # 5 concurrent workers
        ]

        # Start producer and result processor
        producer = asyncio.create_task(captcha_producer(task_queue, urls))
        processor = asyncio.create_task(result_processor(result_queue, len(urls)))

        await producer
        await asyncio.gather(*workers)
        await processor


asyncio.run(main())

Performance tips

Troubleshooting

Frequently asked questions

How many CAPTCHAs can I solve in parallel?

CaptchaAI doesn't enforce a strict concurrency limit per key. Practically, 20-50 concurrent tasks work well. Beyond 50, you may see slower response times.

Is async faster than threading?

For I/O-bound work like API polling, asyncio and threading perform similarly. Async uses less memory (no thread stacks) and avoids GIL issues with many concurrent tasks.

Can I mix CAPTCHA types in a batch?

Yes. Each task is independent. You can solve reCAPTCHA, Turnstile, and GeeTest tasks in the same batch.

Should I use aiohttp or httpx?

Both work. aiohttp is more mature for async HTTP and generally faster. httpx offers a similar API to requests if you prefer consistency.

Summary

Async CAPTCHA solving with asyncio and aiohttp lets you solve many CAPTCHAs in parallel through CaptchaAI. Use asyncio.Semaphore to control concurrency, asyncio.gather() for batch solving, and the producer-consumer pattern for continuous pipelines. Total solve time equals the slowest individual task, not the sum.

Related Articles

• How To Solve Recaptcha V2 Callback Using Api
• Python Playwright Captchaai Complete Guide
• Recaptcha V2 Turnstile Same Site Handling