aiohttp + CaptchaAI: Resolução de CAPTCHA assíncrono

Um scraper que resolve CAPTCHAs um de cada vez passa a maior parte do tempo esperando a API responder, não processando dados. A saída é assíncrona: aiohttp e a API da CaptchaAI enviam e consultam dezenas de desafios ao mesmo tempo, no mesmo loop de eventos, sem abrir uma thread do sistema operacional por requisição. Isso muda o cálculo de qualquer pipeline de alto volume — de um único worker local a jobs distribuídos rodando em sa-east-1 (São Paulo) para reduzir a latência até os sites monitorados.

O que você precisa antes de começar

pip install aiohttp

Construindo o cliente assíncrono da CaptchaAI

Esta classe concentra os três métodos usados no restante do guia: enviar a tarefa, consultar o resultado com polling e checar o saldo — tudo sem bloquear o event loop do aiohttp. Os métodos de submissão e consulta seguem exatamente o contrato da API (in.php para enviar, res.php para consultar), então você pode reaproveitar essa classe em qualquer tipo de CAPTCHA suportado apenas trocando o method.

import aiohttp
import asyncio


class AsyncCaptchaAI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    async def submit(self, session, params):
        """Submit a CAPTCHA task and return the task ID."""
        params["key"] = self.api_key
        async with session.get(
            f"{self.base_url}/in.php", params=params
        ) as resp:
            text = await resp.text()

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        return text.split("|")[1]

    async def poll(self, session, task_id, timeout=300):
        """Poll for the result with a timeout."""
        params = {
            "key": self.api_key,
            "action": "get",
            "id": task_id,
        }
        deadline = asyncio.get_event_loop().time() + timeout

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)

            async with session.get(
                f"{self.base_url}/res.php", params=params
            ) as resp:
                text = await resp.text()

            if text == "CAPCHA_NOT_READY":
                continue
            if text.startswith("OK|"):
                return text.split("|", 1)[1]
            raise Exception(f"Solve failed: {text}")

        raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

    async def solve(self, session, params, timeout=300):
        """Submit and poll in one call."""
        task_id = await self.submit(session, params)
        return await self.poll(session, task_id, timeout)

    async def get_balance(self, session):
        """Check account balance."""
        params = {"key": self.api_key, "action": "getbalance"}
        async with session.get(
            f"{self.base_url}/res.php", params=params
        ) as resp:
            return float(await resp.text())

Testando com um único CAPTCHA

Antes de escalar para lotes, confirme que a chave de API funciona e resolva um único reCAPTCHA v2 de ponta a ponta — envio, polling e leitura do saldo.

import asyncio
import os

async def main():
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Check balance
        balance = await solver.get_balance(session)
        print(f"Balance: ${balance:.2f}")

        # Solve reCAPTCHA v2
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": "https://example.com",
        })
        print(f"Token: {token[:50]}...")

asyncio.run(main())

Resolvendo vários CAPTCHAs em paralelo

Aqui está o ganho real do aiohttp: asyncio.gather dispara todas as tarefas de uma vez e cada uma resolve seu próprio CAPTCHA de forma independente, com return_exceptions=True garantindo que uma falha isolada não derrube o lote inteiro.

async def solve_batch(urls, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        tasks = [
            solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })
            for url in urls
        ]

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

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"FAILED {url}: {result}")
            else:
                print(f"SOLVED {url}: {len(result)} chars")

        return results


urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    "https://example.com/page4",
    "https://example.com/page5",
]
asyncio.run(solve_batch(urls, "6Le-wvkS..."))

Web scraping com resolução automática de CAPTCHA

Na prática, a maioria das páginas não exibe CAPTCHA o tempo todo — só quando o site detecta um padrão de tráfego suspeito. Por isso a função abaixo verifica a presença do desafio antes de acionar o solver, economizando chamadas de API nas páginas sem CAPTCHA. Ao coletar e armazenar dados públicos em produção, vale considerar as obrigações da LGPD para retenção e finalidade do uso desses dados.

async def scrape_with_captcha(url, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Fetch the page
        async with session.get(url) as resp:
            html = await resp.text()

        # Check if page has a CAPTCHA
        if "g-recaptcha" not in html:
            return html  # No CAPTCHA, return content

        # Solve the CAPTCHA
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit with solved token
        async with session.post(url, data={
            "g-recaptcha-response": token,
        }) as resp:
            return await resp.text()

Limitando a concorrência com um semáforo

Disparar centenas de tarefas de uma só vez pode saturar a sua cota de threads simultâneas. Um asyncio.Semaphore mantém a concorrência dentro de um teto previsível — ajuste max_concurrent para ficar abaixo do número de threads do seu plano, evitando erros de limite e mantendo a fila estável:

async def solve_with_limit(urls, site_key, max_concurrent=10):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
    semaphore = asyncio.Semaphore(max_concurrent)

    async def solve_one(session, url):
        async with semaphore:
            return await solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })

    async with aiohttp.ClientSession() as session:
        tasks = [solve_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    solved = sum(1 for r in results if not isinstance(r, Exception))
    print(f"Solved {solved}/{len(urls)} CAPTCHAs")
    return results

Resolvendo Cloudflare Turnstile de forma assíncrona

O mesmo cliente serve para o Cloudflare Turnstile sem nenhuma mudança de estrutura — só o method muda para turnstile e o parâmetro sitekey substitui googlekey, já que o Turnstile não usa o modelo de site key do reCAPTCHA:

async def solve_turnstile(url, sitekey):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        token = await solver.solve(session, {
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": url,
        })
        return token

Erros comuns e como corrigir

Perguntas frequentes

Preciso reescrever todo o scraper em asyncio para usar isso?

Não. Você pode manter o restante do código síncrono e isolar só a etapa de resolução de CAPTCHA em uma função assíncrona chamada via asyncio.run(). Migrar o scraper inteiro só compensa quando o volume de requisições realmente justifica o esforço de reescrita.

Quantas requisições simultâneas a API da CaptchaAI aguenta com esse cliente?

Mais de 100 requisições em paralelo. Na prática, o teto real costuma ser o número de threads do seu plano contratado — use o semáforo para manter a concorrência dentro desse limite em vez de descobrir o teto por tentativa e erro.

O que fazer se aparecer ERROR_ZERO_BALANCE no meio de um lote grande?

As tarefas que já estavam em andamento continuam falhando até você recarregar o saldo. Trate esse erro como definitivo, sem retentativa automática, e dispare um alerta — retomar o lote depois da recarga é mais simples do que tentar recuperar cada tarefa individualmente.

Esse cliente funciona bem para workers rodando em sa-east-1 (São Paulo)?

Sim, e ajuda: como a resolução acontece no servidor da CaptchaAI, a latência entre o seu worker e a origem do CAPTCHA importa menos do que a latência entre o worker e a própria API da CaptchaAI.

Vale mais usar aiohttp do que httpx nesse caso?

Para cargas de alta concorrência, sim — aiohttp é a biblioteca assíncrona mais madura do ecossistema Python, com o melhor desempenho documentado nesse cenário. O guia de integração com httpx cobre a alternativa, que também funciona bem para volumes menores.

Guias relacionados

• Integração HTTPX + CaptchaAI
• Resolução de CAPTCHA em paralelo
• Integração Scrapy + CaptchaAI