Cree un microservicio de resolución de CAPTCHA con FastAPI y CaptchaAI

En cuanto tienes dos o tres proyectos llamando a la misma API de resolución, la respuesta corta es sí: saca esa lógica de cada repositorio y ponla detrás de un único servicio HTTP interno. Tus scrapers, tus suites de QA y tus workers dejan de saber qué es in.php; hacen POST /solve/recaptcha-v2 y reciben un token.

FastAPI encaja aquí por un motivo concreto: resolver un CAPTCHA consiste, casi todo el rato, en esperar una respuesta remota. Con async y httpx, un solo proceso mantiene decenas de solves en vuelo sin dedicar un hilo del sistema operativo a cada uno. Esto es lo que vas a montar:

• un módulo solver que habla con in.php y res.php;
• una app FastAPI con una ruta por tipo de CAPTCHA;
• pruebas con curl y el dimensionado de threads que lo sostiene.

Por qué un servicio y no una librería compartida

Una librería interna parece más simple, pero arrastra tres costes que el servicio elimina:

• La clave API vive en un solo sitio, no copiada en cinco archivos .env.
• Cambias el solver una vez, sin redesplegar cada consumidor.
• El consumo se mide de verdad: latencia, errores y gasto por equipo en un solo punto.

A cambio introduces un salto de red y algo más que mantener. Con un solo script gana la librería; a partir del segundo consumidor, gana el servicio.

Lo que necesitas

Instala las dependencias:

pip install fastapi uvicorn httpx

Estructura del proyecto

captcha-service/
├── main.py          # FastAPI app with endpoints
├── solver.py        # CaptchaAI solving logic
└── requirements.txt

Módulo solver: envío a in.php y sondeo de res.php

Toda la conversación con CaptchaAI cabe en dos funciones. submit_task envía la tarea al endpoint in.php y devuelve un identificador; poll_result espera y consulta res.php hasta que el token está listo. El resto son envoltorios finos por tipo, con su tiempo de espera inicial: 20 segundos para reCAPTCHA, 10 para Turnstile y 5 para imagen. Sondear antes solo suma llamadas que devolverán CAPCHA_NOT_READY.

# solver.py
import os
import httpx
import asyncio

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
BASE_URL = "https://ocr.captchaai.com"


async def submit_task(params: dict) -> str:
    """Submit a CAPTCHA task and return the task ID."""
    params["key"] = API_KEY
    params["json"] = 1

    async with httpx.AsyncClient() as client:
        response = await client.post(f"{BASE_URL}/in.php", data=params)
        data = response.json()

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


async def poll_result(task_id: str, initial_wait: int = 15, max_attempts: int = 30) -> dict:
    """Poll for the CAPTCHA result."""
    await asyncio.sleep(initial_wait)

    async with httpx.AsyncClient() as client:
        for _ in range(max_attempts):
            response = await client.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1
            })
            data = response.json()

            if data.get("status") == 1:
                return {
                    "token": data["request"],
                    "user_agent": data.get("user_agent", "")
                }
            if data.get("request") != "CAPCHA_NOT_READY":
                raise ValueError(f"Solve error: {data['request']}")

            await asyncio.sleep(5)

    raise TimeoutError("Solve timed out")


async def solve_recaptcha_v2(sitekey: str, pageurl: str, enterprise: bool = False) -> dict:
    params = {"method": "userrecaptcha", "googlekey": sitekey, "pageurl": pageurl}
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_recaptcha_v3(sitekey: str, pageurl: str, action: str, enterprise: bool = False) -> dict:
    params = {
        "method": "userrecaptcha", "version": "v3",
        "googlekey": sitekey, "pageurl": pageurl, "action": action
    }
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_turnstile(sitekey: str, pageurl: str) -> dict:
    task_id = await submit_task({"method": "turnstile", "sitekey": sitekey, "pageurl": pageurl})
    return await poll_result(task_id, initial_wait=10)


async def solve_image(image_base64: str) -> dict:
    task_id = await submit_task({"method": "base64", "body": image_base64})
    return await poll_result(task_id, initial_wait=5, max_attempts=15)

La clave se lee de CAPTCHAAI_API_KEY. Nunca la escribas en el código: es el único secreto que este servicio custodia por todos los demás.

La aplicación FastAPI: un endpoint por tipo

Cada tipo recibe su modelo de Pydantic y su ruta: el contrato queda explícito —quien llama sabe que reCAPTCHA v3 exige action— y /docs se genera sola. Los fallos del solver salen como 502: el error viene de aguas arriba, no de una solicitud mal formada. El /health existe para tu orquestador.

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import solver

app = FastAPI(title="CaptchaAI Solver Service")


class RecaptchaV2Request(BaseModel):
    sitekey: str
    pageurl: str
    enterprise: bool = False


class RecaptchaV3Request(BaseModel):
    sitekey: str
    pageurl: str
    action: str
    enterprise: bool = False


class TurnstileRequest(BaseModel):
    sitekey: str
    pageurl: str


class ImageRequest(BaseModel):
    image_base64: str


class SolveResponse(BaseModel):
    token: str
    user_agent: Optional[str] = ""


@app.post("/solve/recaptcha-v2", response_model=SolveResponse)
async def solve_recaptcha_v2(req: RecaptchaV2Request):
    try:
        result = await solver.solve_recaptcha_v2(req.sitekey, req.pageurl, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/recaptcha-v3", response_model=SolveResponse)
async def solve_recaptcha_v3(req: RecaptchaV3Request):
    try:
        result = await solver.solve_recaptcha_v3(req.sitekey, req.pageurl, req.action, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/turnstile", response_model=SolveResponse)
async def solve_turnstile(req: TurnstileRequest):
    try:
        result = await solver.solve_turnstile(req.sitekey, req.pageurl)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/image", response_model=SolveResponse)
async def solve_image(req: ImageRequest):
    try:
        result = await solver.solve_image(req.image_base64)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "ok"}

Levantar el servicio

uvicorn main:app --host 0.0.0.0 --port 8000

En producción, publica el proceso solo en tu red interna y pon delante nginx o Traefik. Un servicio que convierte solicitudes en saldo consumido no debería quedar expuesto a internet sin autenticación.

Ejemplos de uso

Resolver reCAPTCHA v2

curl -X POST http://localhost:8000/solve/recaptcha-v2 \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkS...", "pageurl": "https://staging.example.com/qa-login"}'

Resolver Cloudflare Turnstile

curl -X POST http://localhost:8000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAA...", "pageurl": "https://example.com/form"}'

Respuesta:

{
  "token": "03AGdBq24PBCqLmOx2V4...",
  "user_agent": "Mozilla/5.0..."
}

El campo user_agent importa en reCAPTCHA: el token se resolvió con ese navegador declarado, así que reenvíalo junto al token al enviar el formulario.

Qué rutas tiene sentido exponer

No expongas un endpoint por cada tipo que exista, sino por los que CaptchaAI resuelve:

Puedes añadir rutas para GeeTest v3, Cloudflare Challenge y BLS con el mismo patrón. CaptchaFox (beta), Friendly Captcha (beta) y Lemin (beta) van etiquetados como beta en tu documentación interna. hCaptcha y FunCaptcha (Arkose Labs) no son compatibles y GeeTest v4 figura como próximamente: no crees rutas que prometan lo que la API todavía no devuelve.

Dimensionar los threads del plan

El límite de concurrencia no lo pone el servicio: lo pone tu plan. CaptchaAI factura por threads —cada thread es un solve en curso y cada plan incluye solves ilimitados por thread—, así que el coste mensual es fijo y predecible en USD, algo que agradecen los equipos que facturan en monedas volátiles.

Caso típico en el mercado hispanohablante: un equipo vigila precios en marketplaces regionales y comprueba cada mañana los trámites de portales públicos con CAPTCHA (cita previa en España, portales fiscales en México o Argentina). Ambos flujos beben del mismo pool de threads y ves en una gráfica cuál se come el plan. Si el sondeo se encola, sube de plan antes de tocar max_attempts, y respeta los términos de servicio del sitio y la normativa de protección de datos aplicable.

Errores frecuentes y cómo resolverlos

Preguntas frecuentes

¿Qué hago si res.php responde CAPCHA_NOT_READY una y otra vez?

Es normal mientras el solve está en curso: el bucle sigue consultando cada 5 segundos. Solo al agotarse max_attempts conviene tratarlo como fallo y reintentar la tarea desde cero, nunca en paralelo.

¿Puedo guardar un token y reutilizarlo?

No. Un token de reCAPTCHA o Turnstile es de un solo uso y caduca en pocos minutos: devuélvelo al formulario que lo pidió y descártalo. Lo que sí conviene reutilizar es la conexión httpx.

¿Cuántas solicitudes simultáneas aguanta el servicio?

FastAPI sostiene miles de conexiones abiertas; el techo real son los threads de tu plan. Cuando se saturan, las tareas esperan turno y la latencia sube.

¿Puedo exponer una ruta para hCaptcha o FunCaptcha?

No. CaptchaAI no es compatible con hCaptcha ni con FunCaptcha (Arkose Labs). Sí cubre reCAPTCHA v2/v3, Cloudflare Turnstile y Challenge, GeeTest v3, imagen/OCR, grid y BLS.

¿Necesito httpx o me vale requests?

Necesitas un cliente asíncrono. requests bloquea el event loop en cada espera y convierte tu servicio async en uno secuencial; httpx o aiohttp son las opciones correctas.

Pon tu microservicio en marcha

Obtén tu clave API en captchaai.com, despliega el servicio en tu red interna y apunta el primer scraper a /solve/recaptcha-v2. Cada proyecto nuevo se integra después en una línea.

Guías relacionadas

• Resolver reCAPTCHA v2 con la API
• Resolver Cloudflare Turnstile con la API
• Integración asíncrona con aiohttp
• Resolución de CAPTCHA en paralelo