Manejo de CAPTCHA en aplicaciones Flask con CaptchaAI

¿Necesitas resolver un CAPTCHA desde una aplicación Flask? La forma directa es encapsular la API de CaptchaAI en una clase de servicio y exponer un endpoint que envíe la tarea y sondee el resultado hasta recibir el token. Con ese único patrón cubres reCAPTCHA v2, Cloudflare Turnstile y CAPTCHA de imagen, lo mismo en un microservicio de una sola ruta que en un backend con varios workers.

Flask encaja bien aquí precisamente porque no impone estructura: puedes montar un servicio de automatización en un archivo o crecer hacia Blueprints sin reescribir nada. En esta guía construimos la integración por capas —la clase de servicio, los endpoints síncronos, la resolución en segundo plano con hilos y la organización con Flask Blueprints— y cerramos con los errores más comunes en producción.

Configuración del proyecto

Solo necesitas dos dependencias: Flask para el servidor y requests para hablar con la API de CaptchaAI.

pip install flask requests

Estructura de la aplicación

Separa la lógica de resolución en su propio módulo de servicios para que las rutas queden limpias y la clase sea reutilizable:

myapp/
├── app.py
├── config.py
├── services/
│   └── captcha_solver.py
└── templates/
    └── form.html

La clase de servicio de CaptchaAI

Toda la comunicación con la API vive en una sola clase. Envía la tarea al endpoint in.php, consulta res.php cada 5 segundos y devuelve el token en cuanto el CAPTCHA queda resuelto. Aislar esta lógica te deja llamarla desde cualquier ruta, hilo o Blueprint sin duplicar el flujo de envío y sondeo.

# services/captcha_solver.py
import time
import requests


class CaptchaSolver:
    """CaptchaAI solver service for Flask applications."""

    API_BASE = "https://ocr.captchaai.com"

    def __init__(self, api_key):
        self.api_key = api_key

    def solve_recaptcha_v2(self, sitekey, page_url):
        """Solve reCAPTCHA v2."""
        return self._submit_and_poll({
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
        })

    def solve_turnstile(self, sitekey, page_url):
        """Solve Cloudflare Turnstile."""
        return self._submit_and_poll({
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": page_url,
        })

    def solve_image(self, image_base64):
        """Solve image CAPTCHA."""
        return self._submit_and_poll({
            "method": "base64",
            "body": image_base64,
        })

    def get_balance(self):
        """Check API balance."""
        resp = requests.get(f"{self.API_BASE}/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        }, timeout=30)
        return float(resp.json().get("request", 0))

    def _submit_and_poll(self, params, timeout=120):
        """Submit and poll for result."""
        submit_data = {"key": self.api_key, "json": 1, **params}

        resp = requests.post(f"{self.API_BASE}/in.php", data=submit_data, timeout=30)
        resp.raise_for_status()
        data = resp.json()

        if data.get("status") != 1:
            raise CaptchaSolveError(f"Submit failed: {data.get('request')}")

        task_id = data["request"]

        start = time.time()
        while time.time() - start < timeout:
            time.sleep(5)
            result = requests.get(f"{self.API_BASE}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            }, timeout=30).json()

            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise CaptchaSolveError("CAPTCHA unsolvable")

        raise CaptchaSolveError("Solve timed out")


class CaptchaSolveError(Exception):
    pass

Fíjate en los nombres de método por tipo: userrecaptcha para reCAPTCHA v2 y turnstile para Cloudflare Turnstile. Son los mismos que documenta la API, así que la clase sirve tal cual para cada tipo compatible.

Endpoints Flask para resolver CAPTCHA

Con la clase lista, exponer la resolución es cuestión de una ruta por tipo. Cada endpoint valida la entrada, delega en el servicio y devuelve el token en JSON. Piensa, por ejemplo, en una agencia de Ciudad de México que hace QA del alta de usuarios de un cliente: puede levantar /solve/turnstile como microservicio interno y llamarlo desde su suite de pruebas.

# app.py
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"

solver = CaptchaSolver(app.config["CAPTCHAAI_API_KEY"])


@app.route("/solve/recaptcha", methods=["POST"])
def solve_recaptcha():
    """Solve reCAPTCHA v2 via API."""
    data = request.get_json()
    sitekey = data.get("sitekey")
    page_url = data.get("url")

    if not sitekey or not page_url:
        return jsonify({"error": "sitekey and url required"}), 400

    try:
        token = solver.solve_recaptcha_v2(sitekey, page_url)
        return jsonify({"token": token})
    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/solve/turnstile", methods=["POST"])
def solve_turnstile():
    """Solve Cloudflare Turnstile via API."""
    data = request.get_json()
    sitekey = data.get("sitekey")
    page_url = data.get("url")

    if not sitekey or not page_url:
        return jsonify({"error": "sitekey and url required"}), 400

    try:
        token = solver.solve_turnstile(sitekey, page_url)
        return jsonify({"token": token})
    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/balance", methods=["GET"])
def check_balance():
    """Check CaptchaAI balance."""
    balance = solver.get_balance()
    return jsonify({"balance": balance})


if __name__ == "__main__":
    app.run(debug=True, port=5000)

Cómo probar los endpoints

Con el servidor en marcha, comprueba cada ruta con curl. El endpoint de saldo es útil para verificar que tu clave API responde antes de enviar tareas reales:

# Solve reCAPTCHA
curl -X POST http://localhost:5000/solve/recaptcha \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkSAAAA...", "url": "https://staging.example.com/qa-login"}'

# Solve Turnstile
curl -X POST http://localhost:5000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAAAAAC3DHQ...", "url": "https://example.com/signup"}'

# Check balance
curl http://localhost:5000/balance

Proteger un formulario Flask con Turnstile

El caso inverso es igual de habitual: quieres proteger tus propios formularios Flask con Cloudflare Turnstile y verificar el token en el servidor antes de procesar el envío. Aquí no resuelves un CAPTCHA, lo validas contra Cloudflare con tu clave secreta.

# app.py
from flask import Flask, request, render_template, redirect, url_for, flash
import requests as http_requests

app = Flask(__name__)
app.secret_key = "your-secret-key"
app.config["TURNSTILE_SITE_KEY"] = "0x4AAAAAAAC3DHQhMMQ_Rxrg"
app.config["TURNSTILE_SECRET_KEY"] = "0x4AAAAAAAC3DHQhYYY_secret"


def verify_turnstile(token, remote_ip=None):
    """Verify Turnstile token with Cloudflare."""
    data = {
        "secret": app.config["TURNSTILE_SECRET_KEY"],
        "response": token,
    }
    if remote_ip:
        data["remoteip"] = remote_ip

    resp = http_requests.post(
        "https://challenges.cloudflare.com/turnstile/v0/siteverify",
        data=data,
        timeout=10,
    )
    return resp.json().get("success", False)


@app.route("/contact", methods=["GET", "POST"])
def contact():
    if request.method == "POST":
        turnstile_token = request.form.get("cf-turnstile-response")

        if not turnstile_token:
            flash("CAPTCHA required")
            return redirect(url_for("contact"))

        if not verify_turnstile(turnstile_token, request.remote_addr):
            flash("CAPTCHA verification failed")
            return redirect(url_for("contact"))

        # Process the form
        name = request.form.get("name")
        email = request.form.get("email")
        # ... save or email the data
        flash("Message sent successfully")
        return redirect(url_for("contact"))

    return render_template("form.html",
                           turnstile_sitekey=app.config["TURNSTILE_SITE_KEY"])

El campo que llega en el POST se llama cf-turnstile-response: ese es el nombre exacto del token de Turnstile, y el widget lo inyecta en la plantilla a partir del sitekey.

<!-- templates/form.html -->
<!DOCTYPE html>
<html>
<body>
    <form method="post">
        <input name="name" placeholder="Name" required>
        <input name="email" type="email" placeholder="Email" required>
        <textarea name="message" placeholder="Message" required></textarea>
        <div class="cf-turnstile" data-sitekey="{{ turnstile_sitekey }}"></div>
        <button type="submit">Send</button>
    </form>
    <script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>
</body>
</html>

Resolución en segundo plano con hilos

Flask atiende cada petición de forma síncrona, así que una resolución que tarda entre 15 y 120 segundos dejaría el worker bloqueado todo ese tiempo. La solución es lanzar el trabajo en un hilo aparte y devolver de inmediato un identificador de tarea que el cliente pueda consultar después.

import uuid
import threading
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

app = Flask(__name__)
solver = CaptchaSolver("YOUR_API_KEY")

# In-memory task storage (use Redis in production)
tasks = {}


def solve_in_background(task_id, captcha_type, sitekey, page_url):
    """Background CAPTCHA solver."""
    try:
        if captcha_type == "recaptcha_v2":
            token = solver.solve_recaptcha_v2(sitekey, page_url)
        elif captcha_type == "turnstile":
            token = solver.solve_turnstile(sitekey, page_url)
        else:
            raise ValueError(f"Unknown type: {captcha_type}")

        tasks[task_id] = {"status": "solved", "token": token}

    except CaptchaSolveError as e:
        tasks[task_id] = {"status": "failed", "error": str(e)}


@app.route("/solve/async", methods=["POST"])
def solve_async():
    """Submit CAPTCHA for background solving."""
    data = request.get_json()
    captcha_type = data.get("type", "recaptcha_v2")
    sitekey = data.get("sitekey")
    page_url = data.get("url")

    if not sitekey or not page_url:
        return jsonify({"error": "sitekey and url required"}), 400

    task_id = str(uuid.uuid4())
    tasks[task_id] = {"status": "pending"}

    thread = threading.Thread(
        target=solve_in_background,
        args=(task_id, captcha_type, sitekey, page_url),
    )
    thread.start()

    return jsonify({"task_id": task_id}), 202


@app.route("/solve/status/<task_id>")
def solve_status(task_id):
    """Check solving status."""
    task = tasks.get(task_id)
    if not task:
        return jsonify({"error": "Task not found"}), 404
    return jsonify(task)

El diccionario tasks en memoria funciona para probar, pero en producción sustitúyelo por Redis: si el proceso se reinicia, pierdes el estado, y con varios workers cada uno tendría su propia copia.

Cómo probar el flujo asíncrono

El cliente envía la tarea, recibe un task_id y consulta el estado hasta que pasa de pending a solved:

# Submit async solve
curl -X POST http://localhost:5000/solve/async \
  -H "Content-Type: application/json" \
  -d '{"type": "turnstile", "sitekey": "0x4AAA...", "url": "https://example.com"}'
# Returns: {"task_id": "abc-123-..."}

# Check status
curl http://localhost:5000/solve/status/abc-123-...
# Returns: {"status": "pending"}  or  {"status": "solved", "token": "..."}

Organizar las rutas con un Flask Blueprint

Cuando el proyecto crece, meter todo en app.py se vuelve inmanejable. Un Flask Blueprint agrupa las rutas de CAPTCHA bajo un prefijo (/api/captcha) y las mantiene aisladas del resto de la aplicación:

# blueprints/captcha.py
from flask import Blueprint, request, jsonify, current_app
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

captcha_bp = Blueprint("captcha", __name__, url_prefix="/api/captcha")


def get_solver():
    return CaptchaSolver(current_app.config["CAPTCHAAI_API_KEY"])


@captcha_bp.route("/solve", methods=["POST"])
def solve():
    data = request.get_json()
    captcha_type = data.get("type")
    sitekey = data.get("sitekey")
    url = data.get("url")

    solver = get_solver()

    try:
        if captcha_type == "recaptcha_v2":
            token = solver.solve_recaptcha_v2(sitekey, url)
        elif captcha_type == "turnstile":
            token = solver.solve_turnstile(sitekey, url)
        elif captcha_type == "image":
            image_b64 = data.get("image")
            token = solver.solve_image(image_b64)
        else:
            return jsonify({"error": f"Unknown type: {captcha_type}"}), 400

        return jsonify({"token": token})

    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@captcha_bp.route("/balance")
def balance():
    solver = get_solver()
    return jsonify({"balance": solver.get_balance()})

Registrar el Blueprint en la aplicación principal es una sola línea, y a partir de ahí un mismo endpoint despacha reCAPTCHA v2, Turnstile e imagen según el campo type:

# app.py
from flask import Flask
from blueprints.captcha import captcha_bp

app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"
app.register_blueprint(captcha_bp)

Solución de problemas

Preguntas frecuentes

¿Cómo evito que Flask se bloquee mientras se resuelve el CAPTCHA?

Lanza la resolución en un hilo y responde de inmediato con un task_id. El cliente consulta el estado en /solve/status/<task_id> hasta que pase de pending a solved. Así el worker queda libre para atender otras peticiones.

¿Qué plan de CaptchaAI necesito para un servicio Flask con varios workers?

Depende de cuántas resoluciones concurrentes esperes, no del número de workers. CaptchaAI factura por thread (una resolución en curso), con resoluciones ilimitadas por thread. El plan BASIC ($15/mes, 5 threads) cubre un servicio pequeño; si tu volumen crece, sube a STANDARD ($30/mes, 15 threads) o ADVANCE ($90/mes, 50 threads).

¿Puedo resolver reCAPTCHA, Turnstile e imagen con la misma clase de servicio?

Sí. La clase CaptchaSolver expone un método por tipo y el patrón de envío y sondeo es idéntico: solo cambian los parámetros que envías. Con un if sobre el campo type en el Blueprint despachas los tres desde un único endpoint.

¿Necesito Redis para las tareas en segundo plano?

Para probar, no: el diccionario en memoria basta. En producción sí, porque con varios procesos cada worker tendría su propia copia del estado y un reinicio la borraría. Redis centraliza las tareas y les puede aplicar un TTL.

¿Cómo protejo el endpoint de resolución frente al abuso?

Añade flask-limiter para poner un límite de solicitudes por IP o por clave. Es un endpoint que consume saldo, así que conviene protegerlo igual que cualquier ruta que cueste dinero.

Resumen

Flask se integra con CaptchaAI mediante una clase de servicio que gestiona el flujo de envío y sondeo. Usa endpoints síncronos para casos simples, hilos en segundo plano para resolver sin bloquear al worker y Flask Blueprints cuando la aplicación crece. El mismo servicio cubre reCAPTCHA v2, Cloudflare Turnstile y CAPTCHA de imagen, así que no hay que reescribir la integración por cada tipo.

Artículos relacionados

• Turnstile devuelve 403 tras validar el token: cómo solucionarlo
• Los modos del widget de Cloudflare Turnstile, explicados