Integración HTTPX + CaptchaAI

Sí: httpx sirve perfectamente como cliente de la API de CaptchaAI, y en modo asíncrono es la opción más cómoda cuando necesitas resolver varios CAPTCHA a la vez sin levantar un navegador. Todo el flujo son dos llamadas HTTP —enviar la tarea a in.php, consultar el resultado en res.php— así que no hace falta ningún SDK: con dos clases de unas cuarenta líneas cubres la mayoría de los escenarios de scraping y automatización en Python.

Esta guía parte de una decisión práctica que suele aparecer a mitad de proyecto: ya tienes un scraper que funciona con requests, empieza a aparecer un reCAPTCHA v2 en el formulario y quieres resolverlo sin reescribir el código ni pagar la latencia de un navegador headless. httpx es el camino más corto, porque su API imita a la de requests y además te abre la puerta a async y a HTTP/2 cuando el volumen crezca.

httpx frente a requests y aiohttp

Antes del código, la decisión de librería. Estas son las diferencias que importan cuando el cuello de botella es esperar la resolución de un CAPTCHA:

En resumen: si vienes de requests y quieres migrar sin dolor, httpx síncrono es un reemplazo casi literal. Si tu carga es de decenas de CAPTCHA simultáneos, el cliente asíncrono con HTTP/2 aprovecha mucho mejor cada thread contratado.

Requisitos

pip install httpx

Guarda tu clave en la variable de entorno CAPTCHAAI_API_KEY; nunca la escribas dentro del script ni la subas al repositorio.

Cliente síncrono para scripts y cron

Esta clase encapsula el ciclo completo: envía la tarea, espera cinco segundos, consulta el resultado y repite hasta que llega el token o vence el tiempo de espera. También incluye get_balance() para vigilar el saldo desde el mismo script.

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()

Tres detalles que conviene no tocar: el timeout=30 del cliente es por solicitud HTTP y es distinto del timeout=300 del ciclo de sondeo; CAPCHA_NOT_READY se escribe así, sin la "t", porque es la cadena literal que devuelve la API; y el token que recibes se envía después en el campo g-recaptcha-response del formulario.

Cliente asíncrono para resolver en paralelo

La versión async es idéntica en estructura, pero permite lanzar varias resoluciones a la vez con asyncio.gather. Como cada tarea pasa la mayor parte del tiempo esperando, la concurrencia se traduce casi uno a uno en throughput.

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())

Fíjate en return_exceptions=True: sin él, un fallo cancela el lote entero; con él, puedes reintentar solo las páginas que fallaron.

Cuántas tareas simultáneas puedes lanzar

El límite no lo pone httpx, sino tu plan. CaptchaAI factura por threads concurrentes, con resoluciones ilimitadas dentro del mes, así que un thread equivale a un CAPTCHA en vuelo. Con BASIC ($15/mes, 5 threads) el range(5) del ejemplo va justo al límite; STANDARD ($30/mes, 15 threads) da margen para reintentos, y ADVANCE ($90/mes, 50 threads) es el escalón habitual cuando un scraper corre en varios procesos. Ajusta la concurrencia de asyncio.gather a los threads que tengas contratados: pedir más no acelera nada, solo llena la cola.

Para equipos y freelancers que facturan en monedas locales volátiles, este modelo tiene una ventaja práctica: el costo mensual en USD es fijo y predecible, en lugar de variar con cada resolución.

HTTP/2: menos sobrecarga de conexión

pip install httpx[http2]

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

HTTP/2 multiplexa todas las solicitudes sobre una única conexión TCP. Cuando estás enviando tareas y sondeando resultados en paralelo, eso elimina buena parte del costo de abrir conexiones nuevas, algo que se nota especialmente si tu servidor está lejos de los endpoints de la API.

Caso práctico: monitorizar un formulario protegido

El patrón más común en la región es el mismo: un script que consulta a diario un portal público con CAPTCHA —trámites de administración electrónica, un panel de proveedor, un marketplace regional donde monitorizas tus propios precios— y que un día empieza a devolver la página del desafío en lugar de los datos. La función siguiente detecta el sitekey en el HTML, pide el token y reenvía el formulario:

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())

Si no hay CAPTCHA, la función devuelve el HTML sin gastar saldo: la comprobación previa del sitekey es lo que evita que el costo se dispare en ejecuciones donde el desafío no aparece. Y una nota obligada: automatiza solo flujos sobre los que tengas permiso, y respeta los términos de servicio y la normativa de protección de datos aplicable (GDPR y LOPDGDD en España, LFPDPPP en México y equivalentes en el resto de la región).

Errores frecuentes al integrar httpx

• Crear un AsyncClient nuevo por cada tarea. Pierdes el pool de conexiones y, con HTTP/2, la ventaja de multiplexar. Instancia el cliente una vez y ciérralo con aclose().
• Sondear demasiado rápido. Bajar el intervalo de cinco segundos no adelanta la resolución; solo suma solicitudes a res.php.
• Olvidar el pageurl real. Debe ser la URL donde vive el formulario, no la del recurso que quieres descargar después.
• Confundir tiempos de espera. Un timeout de cliente demasiado corto aborta el sondeo antes de que la tarea termine.

Preguntas frecuentes

¿Cuántos threads necesito para resolver CAPTCHA en paralelo con httpx?

Tantos como resoluciones quieras tener en vuelo al mismo tiempo. Cinco tareas simultáneas necesitan cinco threads; si tu plan tiene menos, las sobrantes esperan. Empieza por BASIC ($15/mes, 5 threads) y sube cuando midas la cola real.

¿Qué hago si el sondeo devuelve CAPCHA_NOT_READY una y otra vez?

Es el estado normal mientras la tarea está en proceso: sigue consultando res.php cada cinco segundos. Si se agota el timeout de 300 segundos, trata la tarea como fallida, registra el id y reintenta con una tarea nueva en lugar de seguir sondeando la anterior.

¿Conviene reutilizar el mismo cliente httpx o crear uno por solicitud?

Reutilízalo. Un único Client o AsyncClient mantiene el pool de conexiones abierto y, con http2=True, multiplexa envíos y sondeos sobre la misma conexión.

¿Puedo resolver hCaptcha con este mismo cliente?

No. CaptchaAI no admite hCaptcha ni FunCaptcha, y GeeTest v4 figura como próximamente. Los tipos compatibles hoy son reCAPTCHA v2 y v3, Cloudflare Turnstile y Cloudflare Challenge, GeeTest v3, imagen/OCR y grid-image; CaptchaFox (beta), Friendly Captcha (beta) y Lemin (beta) están disponibles en fase beta.

¿Puedo usar httpx con Scrapy?

No directamente: Scrapy usa el bucle de eventos de Twisted. Usa httpx en scripts independientes o dentro de frameworks basados en asyncio, como FastAPI.

Guías relacionadas

• Cliente asíncrono con aiohttp y CaptchaAI
• Resolver CAPTCHA con Axios sin navegador
• Scraping en Python con CAPTCHA de por medio