Azure Functions + CaptchaAI: resolver CAPTCHA en la nube

¿Necesitas resolver CAPTCHA dentro de un flujo serverless sin mantener un servidor encendido las 24 horas? Con Azure Functions basta una función con trigger HTTP que envíe el desafío a la API de CaptchaAI, espere el token y lo devuelva. Lo interesante es que el resto del ecosistema de Azure te da gratis las piezas de producción:

• Key Vault guarda tu API key fuera del código.
• Queue Storage reparte el trabajo en lotes y absorbe los picos.
• Application Insights te da la telemetría de cada ejecución.

En esta guía montamos esas piezas paso a paso, con código Python listo para desplegar, y cerramos con un apunte sobre cómo dimensionar tus threads de CaptchaAI para que la escala automática de Azure no se quede esperando.

Función con trigger HTTP

El corazón del servicio es una única función HTTP: recibe un method y sus params, llama a la API de CaptchaAI y responde con el token. La función solve envía la tarea a in.php, sondea res.php cada 5 segundos y corta con un tiempo de espera de 90 segundos si el resultado no llega.

# function_app.py
import json
import time
import os
import logging
import urllib.request
import urllib.parse
import azure.functions as func

app = func.FunctionApp()


@app.route(route="solve", methods=["POST"])
def solve_captcha(req: func.HttpRequest) -> func.HttpResponse:
    """HTTP trigger for CAPTCHA solving."""
    try:
        body = req.get_json()
    except ValueError:
        return func.HttpResponse(
            json.dumps({"error": "JSON body required"}),
            status_code=400,
            mimetype="application/json",
        )

    method = body.get("method", "userrecaptcha")
    params = body.get("params", {})
    api_key = os.environ["CAPTCHAAI_KEY"]

    try:
        token = solve(api_key, method, params)
        return func.HttpResponse(
            json.dumps({"token": token}),
            mimetype="application/json",
        )
    except Exception as e:
        logging.error(f"Solve failed: {e}")
        return func.HttpResponse(
            json.dumps({"error": str(e)}),
            status_code=500,
            mimetype="application/json",
        )


def solve(api_key, method, params, timeout=90):
    """Solve CAPTCHA via CaptchaAI API."""
    submit_data = urllib.parse.urlencode({
        "key": api_key,
        "method": method,
        "json": 1,
        **params,
    }).encode()

    req = urllib.request.Request(
        "https://ocr.captchaai.com/in.php",
        data=submit_data,
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read())

    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")

    task_id = result["request"]

    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        poll_url = (
            f"https://ocr.captchaai.com/res.php"
            f"?key={api_key}&action=get&id={task_id}&json=1"
        )
        with urllib.request.urlopen(poll_url, timeout=15) as resp:
            data = json.loads(resp.read())

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

    raise TimeoutError("Solve timeout")

Un endpoint, varios tipos de CAPTCHA

El cuerpo POST usa userrecaptcha para reCAPTCHA v2, pero la misma función sirve sin tocar el código para otros tipos que resuelve CaptchaAI:

• reCAPTCHA v3, cambiando el method y añadiendo min_score.
• Cloudflare Turnstile.
• GeeTest v3.
• Imágenes y OCR de texto.

Como el token vuelve en el campo request de la respuesta JSON, el resto de tu automatización no tiene que saber qué tipo de CAPTCHA se resolvió.

Guardar la API key en Key Vault

Nunca dejes la clave en local.settings.json ni en variables de entorno en texto plano dentro del portal. Guárdala en Azure Key Vault y deja que la identidad administrada de la función la lea en tiempo de ejecución. Estos comandos crean el vault, guardan el secreto y conceden acceso a la función:

# Create Key Vault
az keyvault create \
  --name captchaai-vault \
  --resource-group myResourceGroup

# Store secret
az keyvault secret set \
  --vault-name captchaai-vault \
  --name CaptchaAIKey \
  --value "YOUR_API_KEY"

# Grant function access
az webapp identity assign \
  --name my-captcha-function \
  --resource-group myResourceGroup

az keyvault set-policy \
  --name captchaai-vault \
  --object-id <principal-id> \
  --secret-permissions get

Después, en la configuración de la aplicación referencias el secreto en lugar de pegar su valor. Así puedes rotar la API key sin volver a desplegar:

CAPTCHAAI_KEY=@Microsoft.KeyVault(SecretUri=https://captchaai-vault.vault.azure.net/secrets/CaptchaAIKey/)

Procesamiento por lotes con Queue Storage

Para volúmenes altos no expongas la resolución por HTTP: encola las tareas y deja que un segundo trigger las consuma a su ritmo. Azure Queue Storage absorbe los picos y reintenta los mensajes fallidos sin que tu código gestione la concurrencia.

@app.queue_trigger(
    arg_name="msg",
    queue_name="captcha-tasks",
    connection="AzureWebJobsStorage",
)
def process_queue_task(msg: func.QueueMessage):
    """Process CAPTCHA task from queue."""
    task = json.loads(msg.get_body().decode())
    api_key = os.environ["CAPTCHAAI_KEY"]

    try:
        token = solve(api_key, task["method"], task["params"])
        logging.info(f"Task {task['id']} solved")

        # Store result in Table Storage or return queue
        _store_result(task["id"], "success", token)

    except Exception as e:
        logging.error(f"Task {task['id']} failed: {e}")
        _store_result(task["id"], "error", str(e))


def _store_result(task_id, status, value):
    """Store result (simplified — use Table Storage in production)."""
    logging.info(f"Result: {task_id} = {status}")

Dimensiona tus threads, no solo tus funciones

Aquí conviene un apunte de escala. Azure puede lanzar decenas de invocaciones en paralelo, pero CaptchaAI factura por thread concurrente, no por resolución, y cada plan incluye resoluciones ilimitadas dentro de sus threads. Un thread es un CAPTCHA en curso: en cuanto termina, queda libre para el siguiente. Por eso el límite real de tu fan-out no es Azure, sino cuántos threads tengas contratados. Ajusta ambos:

• ADVANCE ($90/mes, 50 threads) acompaña bien a decenas de invocaciones simultáneas.
• PREMIUM ($170/mes, 100 threads) cuando tu cola crece y quieres margen.

Estructura del proyecto

Una app de funciones en Python vive en cuatro archivos:

captcha-function/
├── function_app.py
├── requirements.txt
├── host.json
└── local.settings.json

• function_app.py reúne los dos triggers, HTTP y cola.
• requirements.txt solo necesita el paquete azure-functions.
• host.json fija el tiempo máximo de ejecución y el nivel de logs.
• local.settings.json guarda la configuración local, nunca la de producción.

requirements.txt:

azure-functions

host.json:

{
  "version": "2.0",
  "functionTimeout": "00:02:00",
  "logging": {
    "logLevel": {
      "default": "Information"
    }
  }
}

Sube el functionTimeout a 00:02:00 para que la función no muera mientras sondea un CAPTCHA lento. En local, local.settings.json guarda una clave de prueba y apunta al emulador de almacenamiento:

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "CAPTCHAAI_KEY": "YOUR_API_KEY_FOR_LOCAL_DEV"
  }
}

Desplegar en Azure con la CLI

El despliegue son tres pasos: crear la app de funciones en el plan de Consumo, publicarla con las Core Tools y comprobar el endpoint con un POST de prueba.

# Create function app
az functionapp create \
  --resource-group myResourceGroup \
  --consumption-plan-location westus2 \
  --runtime python \
  --runtime-version 3.11 \
  --functions-version 4 \
  --name my-captcha-solver \
  --storage-account mystorageaccount

# Deploy
func azure functionapp publish my-captcha-solver

# Test
curl -X POST https://my-captcha-solver.azurewebsites.net/api/solve \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

Enviar tareas a la cola

Para alimentar el trigger por cola, publica los mensajes desde cualquier cliente. Este ejemplo encola diez tareas de golpe:

from azure.storage.queue import QueueClient
import json

queue = QueueClient.from_connection_string(
    conn_str="YOUR_STORAGE_CONNECTION_STRING",
    queue_name="captcha-tasks",
)

# Submit batch
for i in range(10):
    task = {
        "id": f"task-{i}",
        "method": "userrecaptcha",
        "params": {
            "googlekey": "SITE_KEY",
            "pageurl": f"https://example.com/page{i}",
        },
    }
    queue.send_message(json.dumps(task))
    print(f"Queued task-{i}")

Este patrón encaja bien con equipos que hacen QA de portales públicos protegidos por CAPTCHA —trámites de cita previa, SAT o AFIP, por ejemplo— donde el trabajo llega a ráfagas y un costo mensual predecible en USD es más fácil de presupuestar que el pago por resolución.

Solución de problemas frecuentes

Preguntas frecuentes

¿Cuándo me conviene el plan de Consumo y cuándo el Premium?

El de Consumo es ideal para volúmenes bajos (menos de 100 resoluciones al día) y solo pagas por ejecución. Pásate a Premium cuando el tráfico sea constante: mantiene instancias activas, elimina los arranques en frío y habilita la integración con VNET.

¿Cómo mantengo mi API key fuera del código?

Guárdala en Key Vault y referénciala con la sintaxis @Microsoft.KeyVault(...) en la configuración de la aplicación. La función la lee mediante su identidad administrada, así que puedes rotar la clave sin volver a desplegar ni exponerla en el repositorio.

¿Cuántos CAPTCHA puedo resolver en paralelo desde Azure?

Tantos como threads tenga tu plan de CaptchaAI. Azure escala las invocaciones, pero cada CAPTCHA en curso ocupa un thread; con 50 threads (ADVANCE) sostienes unas 50 resoluciones simultáneas. Dimensiona el plan según tu concurrencia real.

¿Puedo usar Durable Functions para orquestar lotes?

Sí. Durable Functions implementa patrones fan-out/fan-in: lanzas varios CAPTCHA en paralelo y luego recopilas todos los resultados en un solo punto. Es una buena base para procesamiento por lotes ordenado.

Guías relacionadas

• AWS Lambda + CaptchaAI
• Funciones de Google Cloud + CaptchaAI

Despliega en Azure: obtén tu API key de CaptchaAI hoy.