Integração HTTPX + CaptchaAI

Se o seu scraper em Python já roda sobre o HTTPX, não é preciso trocar de biblioteca para lidar com reCAPTCHA e outros desafios: basta chamar a API da CaptchaAI dentro do mesmo cliente — síncrono ou assíncrono — sem sair do ecossistema HTTP/2.

Neste guia você encontra:

• O cliente síncrono completo, com envio, polling e consulta de saldo
• A versão assíncrona para resolver várias tarefas em paralelo com asyncio
• Como ativar HTTP/2 e por que isso reduz o tempo de resposta no polling
• Um exemplo de scraping que só aciona o solver quando o CAPTCHA realmente aparece

Pré-requisitos para resolver CAPTCHA com HTTPX

Antes de copiar os exemplos, confirme que o ambiente tem o essencial:

Dica: antes de rodar os exemplos, teste a chave com uma chamada rápida a getbalance — isso confirma que a autenticação está correta sem gastar créditos com um CAPTCHA de verdade.

pip install httpx

Cliente síncrono: resolver CAPTCHA com HTTPX passo a passo

A classe abaixo cobre o ciclo completo: envia a tarefa para o endpoint in.php, faz o polling em res.php até o token sair e ainda expõe o saldo da conta. Enquanto o desafio não é processado, a API responde CAPCHA_NOT_READY — o loop apenas aguarda e consulta de novo.

import httpx
import time
import os


class CaptchaAISync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.Client(timeout=30)

    def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = self.client.get(f"{self.base_url}/in.php", params=params)
        text = resp.text

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

        task_id = text.split("|")[1]

        # Poll
        deadline = time.time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while time.time() < deadline:
            time.sleep(5)
            result = self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

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

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

    def get_balance(self):
        resp = self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    def close(self):
        self.client.close()


# Usage
solver = CaptchaAISync(os.environ["CAPTCHAAI_API_KEY"])

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

Cliente assíncrono: resolvendo vários CAPTCHAs em paralelo

A versão assíncrona reaproveita a mesma lógica de envio e consulta, mas troca httpx.Client por httpx.AsyncClient — o que permite disparar várias tarefas ao mesmo tempo com asyncio.gather em vez de resolver um CAPTCHA de cada vez.

import httpx
import asyncio
import os


class CaptchaAIAsync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.AsyncClient(timeout=30)

    async def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = await self.client.get(
            f"{self.base_url}/in.php", params=params
        )
        text = resp.text

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

        task_id = text.split("|")[1]

        # Poll
        deadline = asyncio.get_event_loop().time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)
            result = await self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

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

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

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

    async def close(self):
        await self.client.aclose()


# Usage
async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])

    # Solve multiple concurrently
    tasks = [
        solver.solve({
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": f"https://example.com/page{i}",
        })
        for i in range(5)
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Page {i}: FAILED - {r}")
        else:
            print(f"Page {i}: solved ({len(r)} chars)")

    await solver.close()

asyncio.run(main())

HTTP/2 no HTTPX: menos overhead ao consultar o resultado

O HTTPX suporta HTTP/2 nativamente, o que reduz a sobrecarga de conexão:

pip install httpx[http2]

client = httpx.AsyncClient(http2=True, timeout=30)

Com HTTP/2, várias requisições são multiplexadas em uma única conexão — ganho real quando o worker envia e consulta vários CAPTCHAs ao mesmo tempo. Se os workers rodam longe do tráfego que estão atendendo, a diferença aparece no tempo de polling: para tráfego brasileiro, por exemplo, um worker na região AWS sa-east-1 (São Paulo) reduz o RTT de cada chamada e deixa o loop de consulta mais responsivo, principalmente em volume alto.

Do scraping ao CAPTCHA resolvido: fluxo completo com detecção automática

O exemplo a seguir junta as duas pontas. Em três passos:

1. Busca a página com o AsyncClient
2. Procura a sitekey do reCAPTCHA no HTML — só aciona o solver se ela existir
3. Reenvia o formulário com o token no campo g-recaptcha-response

Isso importa porque, em ambiente de QA ou staging, a mesma rota às vezes exibe o CAPTCHA e às vezes não.

import httpx
import re
import os

async def scrape_with_captcha(url, solver):
    async with httpx.AsyncClient() as client:
        # Fetch page
        resp = await client.get(url)
        html = resp.text

        # Check for reCAPTCHA
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        if not match:
            return html

        site_key = match.group(1)
        token = await solver.solve({
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

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


async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
    content = await scrape_with_captcha("https://example.com", solver)
    print(f"Got {len(content)} chars")
    await solver.close()

asyncio.run(main())

Esse padrão evita gastar créditos resolvendo desafios que nem apareceram.

Se os dados coletados incluírem informação pessoal, trate a base com a mesma atenção que a LGPD exige de qualquer outro processamento — rode esse fluxo em ambientes autorizados (QA, staging, monitoramento com permissão).

HTTPX vs requests vs aiohttp: qual cliente usar para resolver CAPTCHA

Na prática, a escolha depende de onde está o gargalo: overhead de conexão, paralelismo ou compatibilidade com código já existente.

Perguntas frequentes

HTTPX no lugar do requests: vale a pena trocar?

Sim, principalmente se o projeto já lida com várias tarefas ao mesmo tempo: a API do httpx é praticamente idêntica à do requests, então a migração do código de envio e polling é rápida, e você ganha suporte assíncrono e HTTP/2 sem reescrever a lógica de resolução.

O cliente assíncrono do HTTPX resolve CAPTCHA mais rápido que o aiohttp?

Depende da carga de trabalho: o aiohttp tem overhead menor em cargas 100% assíncronas, mas o httpx leva vantagem em conexões HTTP/2 e é mais prático quando o mesmo projeto mistura chamadas síncronas e assíncronas — cenário comum em pipelines de scraping.

Quantas tarefas simultâneas eu posso enviar com o AsyncClient?

Isso depende do plano contratado na CaptchaAI, não do httpx. Cada plano libera um número fixo de threads simultâneas, por exemplo:

• BASIC — US$ 15/mês, 5 threads
• ADVANCE — US$ 90/mês, 50 threads

O asyncio.gather pode disparar mais tarefas do que isso, mas as excedentes ficam na fila até uma thread liberar.

Faz diferença rodar o worker perto da região do meu tráfego?

Sim. Como o polling em res.php envolve várias requisições HTTP curtas, rodar o worker perto da origem do tráfego (por exemplo, um servidor em São Paulo, região AWS sa-east-1) reduz o RTT de cada chamada e deixa o loop de consulta mais responsivo — o ganho fica mais visível em volume alto.

Preciso me preocupar com a LGPD ao usar esse fluxo em scraping?

Se os dados coletados incluírem informação pessoal, sim — a LGPD trata a coleta automatizada como qualquer outro processamento de dados. Use esse fluxo em ambientes autorizados (QA, staging, monitoramento de preços com permissão) e trate a base coletada com o mesmo cuidado que daria a qualquer outro dado pessoal.

Guias relacionados

• Resolvendo CAPTCHA em paralelo com aiohttp
• Axios + CaptchaAI sem abrir navegador
• Guia de web scraping em Python com resolução de CAPTCHA