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

Cuando varios servicios o miembros del equipo necesitan resolver CAPTCHA, centralizarlo en un microservicio evita la duplicación de la lógica entre proyectos. El soporte asíncrono de FastAPI lo convierte en una excelente opción: la resolución de CAPTCHA implica esperar respuestas de API externas, que async maneja de manera eficiente sin bloquear subprocesos.

Esta guía crea un microservicio FastAPI que acepta solicitudes de resolución CAPTCHA a través de REST y devuelve tokens resueltos a través de CaptchaAI.

lo que necesitas

Instalar 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 solucionador CaptchaAI

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

Aplicación FastAPI

# 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"}

ejecutar el servicio

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

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..."
}

Solución de problemas

Preguntas frecuentes

¿Por qué utilizar FastAPI para un servicio de resolución de CAPTCHA?

FastAPI maneja el asíncrono I/O de forma nativa, lo cual es ideal para resolver CAPTCHA donde se pasa la mayor parte del tiempo esperando la respuesta de CaptchaAI. Se pueden procesar varias solicitudes simultáneamente sin subprocesos.

¿Puedo agregar autenticación a los puntos finales?

Sí. Agregue la inyección de dependencia de FastAPI con validación de encabezado de clave API u OAuth2 para restringir el acceso.

¿Cuántas solicitudes simultáneas puede manejar esto?

Limitado por el límite de tareas simultáneas de su plan CaptchaAI. La propia FastAPI puede manejar miles de conexiones simultáneas.

¿Debería dockerizar esto?

Sí. Agregue un Dockerfile con FROM python:3.11-slim, instale dependencias y exponga el puerto 8000.

¿Puedo agregar limitación de velocidad?

Sí. Utilice slowapi o un proxy inverso (nginx, Traefik) para limitar las solicitudes por cliente.

Cree su microservicio de resolución de CAPTCHA

Obtenga su clave API encaptchaai.com. Centralice la resolución de CAPTCHA con un microservicio FastAPI.

Guías relacionadas

• Cómo resolver reCAPTCHA v2 usando API
• Cómo resolver Cloudflare Turnstile usando API
• Asíncrono CaptchaAI con aiohttp
• Resolución CAPTCHA paralela