Sí, se pueden resolver decenas de CAPTCHA a la vez desde un único proceso de Python, y la pieza que lo permite no es el solver: es el bucle de eventos. Con requests, cada CAPTCHA congela tu scraper mientras esperas el token, y con veinte páginas en cola esa espera acaba siendo la parte más cara del trabajo. Con aiohttp las esperas se solapan: envías las tareas a la API de CaptchaAI, devuelves el control al bucle y recoges los tokens conforme llegan. Aquí montamos ese cliente desde cero, con control de concurrencia y una regla clara para saber cuántos threads necesitas.
Lo que necesitas antes de escribir código
| Requisito | Detalles |
|---|---|
| Python | 3.8+ |
| aiohttp | 3.8+ |
| API key de CaptchaAI | Regístrate y copia tu clave |
pip install aiohttp
Una sola dependencia. Si tu stack ya usa httpx, el patrón es idéntico y tienes la versión equivalente con httpx: envío, sondeo y semáforo se trasladan sin cambios.
Cliente asíncrono de CaptchaAI en una clase
La API funciona en dos pasos: envías la tarea a in.php y consultas el resultado en res.php hasta que esté listo. Esta clase encapsula el ciclo y expone un único solve().
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())
El detalle clave es el await asyncio.sleep(5) del sondeo: no bloquea nada. Mientras esa corrutina duerme, el bucle atiende al resto, así que cien sondeos en curso cuestan casi lo mismo que uno. La clave API se lee de una variable de entorno, nunca del archivo.
Primera prueba: un reCAPTCHA v2
Antes de montar lotes, valida credenciales y saldo con una sola resolución. Si get_balance() devuelve un número, la autenticación está resuelta y solo queda ajustar parámetros.
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())
El googlekey es el sitekey del HTML de la página objetivo y pageurl debe ser la URL exacta donde se muestra el desafío: un sitekey correcto con una pageurl equivocada es la causa número uno de tokens rechazados.
Varios CAPTCHA en paralelo con asyncio.gather
Aquí el enfoque asíncrono se paga solo: asyncio.gather lanza todas las resoluciones a la vez y espera al conjunto, no una por una.
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..."))
return_exceptions=True es obligatorio en producción: sin él, un único fallo cancela el gather entero y pierdes los tokens ya resueltos.
Scraping cuando el CAPTCHA aparece de forma intermitente
Pocas páginas muestran el desafío siempre: lo habitual es que salte cuando baja la reputación de la IP o sube la frecuencia de solicitudes. Detéctalo en el HTML y resuélvelo solo cuando toca.
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()
El token vuelve al sitio en g-recaptcha-response, el mismo campo que usaría el widget en el navegador. Y una nota obligada en cualquier proyecto de web scraping: respeta los términos de servicio del sitio y la normativa de protección de datos aplicable (GDPR y LOPDGDD en España, LFPDPPP en México).
Semáforo: ajustar la concurrencia a tus threads
Lanzar mil corrutinas a la vez no multiplica la velocidad, solo llena la cola. CaptchaAI factura por thread — un thread es un CAPTCHA en vuelo — y ese número es el techo real de tu paralelismo, así que alinea el asyncio.Semaphore con él.
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
La regla es sencilla: max_concurrent igual a los threads de tu plan. Con BASIC ($15/mes, 5 threads) pones 5; con STANDARD ($30/mes, 15 threads), 15; con ADVANCE ($90/mes, 50 threads), 50. Todos los planes incluyen resoluciones ilimitadas por thread: no pagas por CAPTCHA resuelto, sino por cuántos tienes en vuelo a la vez.
Turnstile con el mismo cliente
Cambiar de tipo no exige otro cliente: solo otro method y sitekey en lugar de googlekey.
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
Turnstile se resuelve en menos de 10 s con una alta tasa de éxito, muy por debajo del techo de reCAPTCHA v2, así que los lotes mixtos terminan antes de lo que sugiere el peor caso. Ahí el token viaja en cf-turnstile-response.
Caso práctico: monitorización de precios en marketplaces regionales
Imagina una agencia de datos en Ciudad de México que revisa cada mañana el catálogo de tres marketplaces de la región y una tienda en Amazon.es: unas 900 fichas de producto, de las que alrededor del 15% dispara un desafío. En secuencial con requests, esas ~135 resoluciones suman más de una hora de espera pura. Con el patrón de esta guía y ADVANCE ($90/mes, 50 threads), el lote pasa a estar dominado por la descarga del HTML, no por el CAPTCHA.
En agencias y equipos freelance de la región, lo que suele decidir la compra no es la velocidad sino la previsibilidad: un costo mensual fijo en USD frente a una factura variable por resolución. Si entra un cliente nuevo, subes el semáforo y, si hace falta, el plan.
Tipos de CAPTCHA que cubre este cliente
El mismo AsyncCaptchaAI resuelve reCAPTCHA v2 y v3 (incluidas Invisible y Enterprise), Cloudflare Turnstile y Cloudflare Challenge, GeeTest v3, CAPTCHA de imagen y OCR, grid-image y BLS CAPTCHA, además de CaptchaFox (beta), Friendly Captcha (beta) y Lemin (beta). En cambio, hCaptcha y FunCaptcha (Arkose Labs) no son compatibles, y GeeTest v4 figura como próximamente: si tu objetivo usa alguno de esos, este cliente no te sirve para ese flujo concreto.
Errores frecuentes y cómo resolverlos
| Error | Causa | Solución |
|---|---|---|
ClientConnectorError |
Problema de red | Comprobar conectividad |
Submit failed: ERROR_ZERO_BALANCE |
Sin fondos | Recargar cuenta |
TimeoutError |
resolución lenta | Aumentar el parámetro de tiempo de espera |
RuntimeError: Event loop is closed |
Usando asyncio.run en Jupyter |
Usa nest_asyncio |
Añade dos síntomas propios del código asíncrono: Session is closed significa que usas la sesión fuera de su bloque async with, y una caída brusca del rendimiento al subir el semáforo por encima de tus threads indica que encolas, no aceleras.
Preguntas frecuentes
¿Cuántos CAPTCHA por hora puedo resolver con cada plan?
Depende del tipo. Toma el peor caso publicado y divide: con 50 threads y un reCAPTCHA v2 de hasta 60 s, el suelo son unas 50 resoluciones por minuto; con Turnstile, en menos de 10 s, el mismo plan rinde bastante más.
¿Qué pasa si una tarea falla dentro de asyncio.gather?
Con return_exceptions=True, la excepción llega como un elemento más de la lista y el resto de resoluciones continúa. Filtra esos elementos y reintenta solo las URL afectadas, con retroceso exponencial.
¿Funciona este código dentro de FastAPI, Celery o un cronjob?
Sí, con un matiz por entorno. En FastAPI llamas a las corrutinas desde el endpoint asíncrono; en un worker de Celery clásico envuelves la llamada en asyncio.run; en Jupyter necesitas nest_asyncio porque ya hay un bucle en marcha.
¿Necesito un navegador headless para esto?
No. Todo el flujo es HTTP puro contra la API, sin abrir ningún navegador headless, y eso es lo que hace viable una concurrencia alta con memoria mínima.
¿Conviene reutilizar la misma ClientSession entre resoluciones?
Sí. Cada ClientSession mantiene su pool de conexiones; crear una por CAPTCHA repite el handshake TLS y añade latencia. Crea una por proceso o por lote.
