CAPTCHA-Handhabung in Flask-Anwendungen mit CaptchaAI

CaptchaAI binden Sie in Flask am saubersten über eine einzige Serviceklasse ein: Sie übermittelt die CAPTCHA-Aufgabe an in.php, fragt das Ergebnis über res.php ab und gibt am Ende das fertige Token zurück. Ob Sie einen schlanken Solver-Endpunkt, ein Turnstile-geschütztes Kontaktformular oder eine Hintergrundverarbeitung für höhere Last bauen – die Integration bleibt dieselbe, nur das Aufrufmuster ändert sich. Dieser Leitfaden führt Sie Schritt für Schritt vom ersten Endpunkt bis zur Blueprint-Struktur für größere Anwendungen. Die Reihenfolge folgt der Praxis: erst das Projekt und die Serviceklasse, dann synchrone Endpunkte, danach Formularschutz, Threading und Modularisierung.

Projekt aufsetzen und strukturieren

pip install flask requests

Ein überschaubarer Aufbau trennt die Solver-Logik von den Routen. So bleibt die Serviceklasse testbar und lässt sich später ohne Umbau in einen Blueprint oder einen eigenen Worker verschieben:

• services/captcha_solver.py kapselt den Submit/Poll-Fluss und kennt die CAPTCHA-Typen.
• app.py bindet die Routen ein und hält den API-Schlüssel in der Konfiguration.
• templates/form.html liefert das Turnstile-Widget für den Formularschutz.

Projektstruktur

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

Die CaptchaAI-Serviceklasse

Kern der Integration ist eine Klasse, die den kompletten Submit/Poll-Fluss kapselt. Sie schickt die Aufgabe an in.php, fragt danach im Intervall res.php ab und liefert entweder das Token oder eine aussagekräftige Ausnahme. Für die einzelnen CAPTCHA-Typen gibt es je eine kleine Methode – reCAPTCHA v2, Cloudflare Turnstile und Bild-CAPTCHAs teilen sich dieselbe interne Logik.

# 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

Das timeout=120 deckt den vollen Bereich ab: Ein Lösungsvorgang dauert je nach Typ typischerweise 15–120 Sekunden, das time.sleep(5) hält das Polling bewusst ruhig, damit Sie res.php nicht unnötig belasten. Ersetzen Sie YOUR_API_KEY später durch Ihren echten Schlüssel aus dem Dashboard.

Erste Endpunkte zum CAPTCHA-Lösen

Für einfache Fälle genügt ein dünner Wrapper um die Serviceklasse: JSON rein, Token raus. Die folgenden Routen nehmen Sitekey und Ziel-URL entgegen, geben bei fehlenden Feldern sauber 400 zurück und melden Solver-Fehler als 500.

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

Aufruf über curl

# Solve reCAPTCHA
curl -X POST http://localhost:5000/solve/recaptcha \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkSAAAA...", "url": "https://example.com/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

Der /balance-Endpunkt ist praktisch fürs Monitoring: Prüfen Sie Ihr Guthaben regelmäßig, bevor ein Batch startet, statt erst beim ersten fehlgeschlagenen Solve davon zu erfahren.

Flask-Formulare mit Cloudflare Turnstile absichern

Bis hierher hat Flask CAPTCHAs gelöst. Der umgekehrte Fall ist genauso häufig: Sie betreiben selbst ein Formular – etwa das Kontaktformular eines DACH-SaaS-Anbieters – und wollen es mit Cloudflare Turnstile gegen Spam-Bots absichern. Dann prüft Flask das eingehende Token serverseitig über Cloudflares siteverify-Endpunkt.

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

<!-- 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>

Das Token landet als cf-turnstile-response im POST-Body – dieser Feldname ist bei Turnstile fest vorgegeben.

Datenschutz-Hinweis: Sobald Sie request.remote_addr an siteverify weiterreichen, verarbeiten Sie eine IP-Adresse, die nach DSGVO als personenbezogenes Datum gilt. Prüfen Sie Ihre Rechtsgrundlage und dokumentieren Sie den Datenfluss in Ihrer Datenschutzerklärung; der Parameter remoteip ist optional und kann entfallen, wenn Sie ihn nicht benötigen.

Nicht blockierendes Lösen im Hintergrund-Thread

Flask verarbeitet Anfragen standardmäßig synchron. Ein Solve, der 15–120 Sekunden dauert, blockiert damit einen ganzen Worker – bei mehreren gleichzeitigen Anfragen läuft Ihnen schnell die Kapazität aus. Das Muster für nicht blockierendes Lösen:

1. Aufgabe annehmen, eine Task-ID erzeugen und sofort mit 202 antworten.
2. Den eigentlichen Solve in einem Hintergrund-Thread ausführen.
3. Den Status über einen separaten Endpunkt abfragen, bis das Token bereitliegt.

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)

Aufruf über curl

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

Wie viele Solves parallel laufen dürfen, bestimmt bei CaptchaAI Ihr Tarif – abgerechnet wird pro Thread, nicht pro Lösung. Der Einstiegstarif BASIC (15 $/Monat, 5 Threads) erlaubt fünf gleichzeitige Solves bei unbegrenzten Lösungen pro Thread; wächst die Last, skalieren Sie über ADVANCE (90 $/Monat, 50 Threads) oder höher. Das In-Memory-tasks-Dictionary eignet sich nur für einen einzelnen Prozess: In Produktion mit mehreren Gunicorn-Workern gehört der Status in einen gemeinsamen Speicher wie Redis.

Routen als Flask Blueprint organisieren

Wächst die Anwendung, wird ein einzelnes app.py schnell unübersichtlich. Ein Blueprint bündelt alle CAPTCHA-Routen unter einem gemeinsamen Präfix und holt den Solver pro Anfrage aus der App-Konfiguration – so bleibt der Schlüssel an einer Stelle und die Routen bleiben wiederverwendbar.

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

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

Mit register_blueprint sind alle Routen unter /api/captcha/... erreichbar. Denselben Blueprint können Sie später in einem anderen Projekt einhängen, ohne die Solver-Logik erneut zu schreiben.

Fehlerbehebung

Häufige Fragen

Wie viele CAPTCHAs kann ich mit CaptchaAI parallel lösen?

So viele, wie Ihr Tarif Threads bereitstellt. Abgerechnet wird pro gleichzeitigem Thread, nicht pro Lösung: BASIC (15 $/Monat) deckt 5 parallele Solves ab, ADVANCE (90 $/Monat) bereits 50. Innerhalb eines Tarifs sind die Lösungen pro Thread unbegrenzt.

Warum sollte ich in Flask ein Timeout für den Solver setzen?

Weil ein synchroner Solve 15–120 Sekunden dauern kann und Ihren WSGI-Server sonst über sein Standard-Limit hinaus blockiert. Setzen Sie in Gunicorn --timeout 180 und lassen Sie die _submit_and_poll-Grenze etwas darunter – so bricht der Client kontrolliert ab, statt hart abzuschneiden.

Welche CAPTCHA-Typen deckt diese Serviceklasse ab?

Die Beispiele lösen reCAPTCHA v2, Cloudflare Turnstile und Bild-CAPTCHAs über den Submit/Poll-Fluss; reCAPTCHA v3 und weitere GA-Typen ergänzen Sie mit denselben zwei Methoden. hCaptcha und FunCaptcha werden von CaptchaAI nicht unterstützt – planen Sie diese Typen nicht ein.

Eignet sich Threading oder eine Task-Queue wie Celery besser?

Für gelegentliche Hintergrund-Solves reicht threading mit einem gemeinsamen Statusspeicher. Sobald Sie mehrere Prozesse, Wiederholungslogik oder persistente Aufgaben brauchen, ist eine Task-Queue wie Celery mit Redis oder RabbitMQ die robustere Wahl.

Fazit

CaptchaAI lässt sich in Flask über eine einzige Serviceklasse einbinden, die den Submit/Poll-Fluss kapselt. Wählen Sie das Aufrufmuster nach Ihrer Last: synchrone Endpunkte für einfache Fälle, Hintergrund-Threading für nicht blockierendes Lösen und einen Blueprint für größere, modular aufgebaute Anwendungen. Dieselbe Serviceklasse bedient dabei reCAPTCHA, Turnstile und Bild-CAPTCHAs – Sie erweitern sie um weitere Typen, ohne die Endpunkte umzuschreiben.

Verwandte Artikel

• GeeTest und reCAPTCHA im Vergleich
• Cloudflare Turnstile: Fehler erkennen und beheben
• Die Widget-Modi von Cloudflare Turnstile erklärt