Erstellen von Client-CAPTCHA-Pipelines mit CaptchaAI

Eine Agentur, die Scraping und Automatisierung für ein Dutzend Kunden betreibt, sollte CAPTCHAs nicht in jedem Projekt neu lösen. Die tragfähige Antwort ist eine einzige, wiederverwendbare Pipeline: Jeder Client übergibt seine Lösungsanfrage an dieselbe Warteschlange, ein Pool von Workern reicht sie an CaptchaAI weiter, und die fertigen Token landen zentral im Ergebnisspeicher. Dieser Leitfaden zeigt die Architektur und den vollständigen Code – in Python und Node.js.

Warum eine gemeinsame Pipeline statt Einzellösungen

Einmaliger Lösungscode pro Projekt fühlt sich anfangs schneller an, rächt sich aber im Betrieb: Jeder Kunde bekommt eine leicht andere Retry-Logik, das Fehlerverhalten driftet auseinander, und niemand weiß mehr, welches Skript welchen API-Schlüssel verbrennt. Eine zentrale Pipeline dreht das um.

• Ein Codepfad, viele Kunden – Sie pflegen Warteschlange, Polling und Fehlerbehandlung an genau einer Stelle.
• Isolierte Parallelität pro Client – ein Kunde mit einem großen Batch blockiert nicht die Lösungen der anderen.
• Nachvollziehbare Abrechnung – über die client_id lässt sich jede Lösung dem richtigen Projekt zuordnen.

Für DACH-Agenturen kommt ein praktischer Punkt hinzu: Wer Worker auf günstigen VPS bei Hetzner, IONOS oder netcup betreibt, will die CAPTCHA-Last bündeln, statt pro Kundenprojekt eine eigene Instanz hochzufahren. Genau das leistet die folgende Architektur.

Architektur der Pipeline

┌──────────────┐    ┌───────────────┐    ┌──────────────┐
│  Client A    │──▶ │               │    │              │
│  Client B    │──▶ │  Task Queue   │──▶ │  CaptchaAI   │
│  Client C    │──▶ │               │    │  API         │
└──────────────┘    └───────────────┘    └──────────────┘
                           │                    │
                           ▼                    ▼
                    ┌───────────────┐    ┌──────────────┐
                    │  Result Store │◀── │  Polling      │
                    │  (Redis/DB)   │    │  Workers      │
                    └───────────────┘    └──────────────┘

Vier Bausteine tragen die Pipeline:

1. Aufgabenannahme – nimmt die Lösungsanfragen der Client-Scraper entgegen.
2. Warteschlange – puffert Aufgaben und erzwingt Parallelitätslimits pro Client.
3. Worker – übermitteln an CaptchaAI und fragen das Ergebnis ab.
4. Ergebnisspeicher – hält die gelösten Token zum Abruf durch den Consumer bereit.

Diese Trennung sorgt dafür, dass Sie jeden Baustein einzeln skalieren können: mehr Worker bei hoher Last, ein persistenter Ergebnisspeicher (Redis oder Datenbank) für den Neustart-Fall.

Python-Pipeline aufbauen

Die zentrale Löser-Klasse

Die Klasse CaptchaPipeline kapselt Warteschlange, Übermittlung an in.php und Polling gegen res.php. Aufgaben landen über enqueue in der Queue; process_queue füllt aktive Slots bis zum Limit max_concurrent und fragt laufende Aufgaben ab, bis ein Token vorliegt.

import requests
import time
from dataclasses import dataclass
from typing import Optional
from collections import deque
from threading import Lock

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"

@dataclass
class SolveRequest:
    client_id: str
    method: str
    params: dict
    callback: Optional[callable] = None

@dataclass
class SolveResult:
    client_id: str
    task_id: str
    token: Optional[str] = None
    error: Optional[str] = None


class CaptchaPipeline:
    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.max_concurrent = max_concurrent
        self.queue = deque()
        self.active = {}
        self.lock = Lock()

    def enqueue(self, request: SolveRequest):
        with self.lock:
            self.queue.append(request)

    def submit_task(self, request: SolveRequest) -> Optional[str]:
        data = {
            "key": self.api_key,
            "method": request.method,
            "json": 1,
            **request.params
        }

        try:
            resp = requests.post(SUBMIT_URL, data=data, timeout=15)
            result = resp.json()

            if result.get("status") == 1:
                return result["request"]
            else:
                print(f"[{request.client_id}] Submit error: {result.get('error_text', result.get('request'))}")
                return None
        except requests.RequestException as e:
            print(f"[{request.client_id}] Network error: {e}")
            return None

    def poll_result(self, task_id: str, max_wait: int = 120) -> Optional[str]:
        elapsed = 0
        interval = 5
        while elapsed < max_wait:
            time.sleep(interval)
            elapsed += interval

            try:
                resp = requests.get(RESULT_URL, params={
                    "key": self.api_key,
                    "action": "get",
                    "id": task_id,
                    "json": 1
                }, timeout=10)
                result = resp.json()

                if result.get("status") == 1:
                    return result["request"]
                elif result.get("request") == "CAPCHA_NOT_READY":
                    continue
                else:
                    print(f"Poll error for {task_id}: {result.get('error_text', result.get('request'))}")
                    return None
            except requests.RequestException:
                continue

        return None

    def process_queue(self):
        while self.queue or self.active:
            # Fill active slots
            with self.lock:
                while self.queue and len(self.active) < self.max_concurrent:
                    request = self.queue.popleft()
                    task_id = self.submit_task(request)
                    if task_id:
                        self.active[task_id] = request

            # Poll active tasks
            completed = []
            for task_id, request in list(self.active.items()):
                token = self.poll_result(task_id, max_wait=10)
                if token:
                    result = SolveResult(
                        client_id=request.client_id,
                        task_id=task_id,
                        token=token
                    )
                    if request.callback:
                        request.callback(result)
                    completed.append(task_id)

            with self.lock:
                for task_id in completed:
                    del self.active[task_id]

Nutzung für mehrere Clients

Jeder Kunde bekommt eine eigene client_id und seine passende method – hier reCAPTCHA v2 (userrecaptcha) für Client A und Cloudflare Turnstile (turnstile) für Client B. Der callback reicht das gelöste Token an den jeweiligen Consumer zurück.

pipeline = CaptchaPipeline(api_key="YOUR_API_KEY", max_concurrent=15)

# Client A — reCAPTCHA v2
pipeline.enqueue(SolveRequest(
    client_id="client_a",
    method="userrecaptcha",
    params={
        "googlekey": "6Le-SITEKEY-A",
        "pageurl": "https://client-a-target.com/form"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

# Client B — Turnstile
pipeline.enqueue(SolveRequest(
    client_id="client_b",
    method="turnstile",
    params={
        "sitekey": "0x4AAAA-SITEKEY-B",
        "pageurl": "https://client-b-target.com/login"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

pipeline.process_queue()

Dieselbe Pipeline in Node.js

Wenn Ihr Stack auf Node.js läuft, bildet diese Variante dasselbe Muster ab – nur promise-basiert. enqueue gibt ein Promise zurück, _processNext hält die Parallelität unter maxConcurrent, und _poll fragt das Ergebnis in festen Intervallen ab.

const axios = require("axios");

const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

class CaptchaPipeline {
  constructor(apiKey, maxConcurrent = 10) {
    this.apiKey = apiKey;
    this.maxConcurrent = maxConcurrent;
    this.queue = [];
    this.activeCount = 0;
  }

  enqueue(clientId, method, params) {
    return new Promise((resolve, reject) => {
      this.queue.push({ clientId, method, params, resolve, reject });
      this._processNext();
    });
  }

  async _processNext() {
    if (this.activeCount >= this.maxConcurrent || this.queue.length === 0) return;

    this.activeCount++;
    const task = this.queue.shift();

    try {
      const token = await this._solve(task);
      task.resolve({ clientId: task.clientId, token });
    } catch (err) {
      task.reject(err);
    } finally {
      this.activeCount--;
      this._processNext();
    }
  }

  async _solve(task) {
    const submitResp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: this.apiKey,
        method: task.method,
        json: 1,
        ...task.params,
      },
      timeout: 15000,
    });

    if (submitResp.data.status !== 1) {
      throw new Error(submitResp.data.error_text || submitResp.data.request);
    }

    const taskId = submitResp.data.request;
    return this._poll(taskId);
  }

  async _poll(taskId, maxWait = 120000) {
    const interval = 5000;
    let elapsed = 0;

    while (elapsed < maxWait) {
      await new Promise((r) => setTimeout(r, interval));
      elapsed += interval;

      try {
        const resp = await axios.get(RESULT_URL, {
          params: {
            key: this.apiKey,
            action: "get",
            id: taskId,
            json: 1,
          },
          timeout: 10000,
        });

        if (resp.data.status === 1) return resp.data.request;
        if (resp.data.request !== "CAPCHA_NOT_READY") {
          throw new Error(resp.data.error_text || resp.data.request);
        }
      } catch (err) {
        if (err.response) throw err;
      }
    }

    throw new Error(`Timeout waiting for task ${taskId}`);
  }
}

// Usage
(async () => {
  const pipeline = new CaptchaPipeline("YOUR_API_KEY", 15);

  const results = await Promise.allSettled([
    pipeline.enqueue("client_a", "userrecaptcha", {
      googlekey: "6Le-SITEKEY-A",
      pageurl: "https://client-a-target.com/form",
    }),
    pipeline.enqueue("client_b", "turnstile", {
      sitekey: "0x4AAAA-SITEKEY-B",
      pageurl: "https://client-b-target.com/login",
    }),
  ]);

  results.forEach((r) => {
    if (r.status === "fulfilled") {
      console.log(`[${r.value.clientId}] Token: ${r.value.token.slice(0, 40)}...`);
    } else {
      console.error(`Failed: ${r.reason.message}`);
    }
  });
})();

Konfiguration pro Client

Nicht jeder Kunde braucht dieselben Einstellungen. Halten Sie Proxy, Solver-Präferenz und Parallelitätslimit pro Client in einer zentralen Konfiguration – so bleibt die Pipeline-Logik generisch, während die Feinheiten datengetrieben bleiben.

CLIENT_CONFIG = {
    "client_a": {
        "proxy": "host:port:user:pass",
        "proxytype": "HTTP",
        "max_concurrent": 5,
        "default_method": "userrecaptcha"
    },
    "client_b": {
        "proxy": None,
        "proxytype": None,
        "max_concurrent": 10,
        "default_method": "turnstile"
    }
}

def build_params(client_id, params):
    config = CLIENT_CONFIG.get(client_id, {})
    if config.get("proxy"):
        params["proxy"] = config["proxy"]
        params["proxytype"] = config["proxytype"]
    return params

Parallelität an Ihr Thread-Kontingent koppeln

Die entscheidende Stellschraube ist max_concurrent. CaptchaAI rechnet nach Threads ab, nicht pro Lösung: Ein Thread ist eine gleichzeitig laufende CAPTCHA-Abfrage, und jeder Tarif enthält unbegrenzte Lösungen pro Thread im Abrechnungsmonat. Die Summe der Parallelität über alle Clients sollte deshalb Ihr Thread-Kontingent nicht überschreiten.

Wer also drei Kunden mit je fünf gleichzeitigen Aufgaben bedient, liegt bei 15 Threads – das deckt STANDARD (30 $/Monat, 15 Threads) ab. Skaliert die Agentur auf mehr Kunden, greift ADVANCE (90 $/Monat, 50 Threads). Die aktuellen Werte stehen auf der Preisseite von CaptchaAI; Preise verstehen sich in US-Dollar.

Fehler robust behandeln

Eine Multi-Client-Pipeline lebt oder stirbt mit ihrer Fehlerbehandlung. Ordnen Sie jedem API-Fehlercode eine klare Reaktion zu, statt eine Aufgabe stillschweigend fallen zu lassen.

Häufige Probleme im Betrieb

DSGVO im Blick behalten

Sobald Sie Proxys pro Kunde einsetzen, fließen fremde IP-Adressen durch Ihre Infrastruktur – und IP-Adressen gelten in der EU als personenbezogene Daten. Prüfen Sie je Client die Rechtsgrundlage und die Datenflüsse, bevor Sie Traffic über einen gemeinsamen Proxy-Pool routen. Das ist eine Frage Ihrer eigenen Sorgfaltspflicht als Auftragsverarbeiter, nicht eine Compliance-Zusage des Lösungsdienstes.

Häufige Fragen

Wie hängt max_concurrent mit meinem CaptchaAI-Tarif zusammen?

Direkt: Die Summe von max_concurrent über alle Clients sollte die Thread-Zahl Ihres Tarifs nicht übersteigen. BASIC bietet 5 Threads, STANDARD 15, ADVANCE 50. Übermitteln Sie mehr parallel, warten die überzähligen Aufgaben in der Warteschlange.

Kann eine Pipeline reCAPTCHA v2 und Turnstile gleichzeitig verarbeiten?

Ja. Die method steckt in jeder SolveRequest, nicht in der Pipeline. Client A kann userrecaptcha fahren, Client B turnstile – dieselbe Warteschlange und dieselben Worker bedienen beide, weil nur der method-Parameter und die zugehörigen Felder (googlekey bzw. sitekey) wechseln.

Wie gehe ich mit Proxys und der DSGVO um?

Behandeln Sie Kunden-IPs als personenbezogene Daten und dokumentieren Sie je Client die Rechtsgrundlage. Trennen Sie Proxy-Konfigurationen sauber pro client_id, damit kein Traffic versehentlich über den falschen Pool läuft.

Was passiert mit laufenden Aufgaben bei einem Neustart?

Ohne Persistenz gehen sie verloren. Legen Sie Warteschlange und Ergebnisspeicher in Redis oder einer Datenbank ab; nach dem Neustart laden Sie offene Aufgaben neu und setzen die Verarbeitung fort.

Starten Sie Ihre Client-Pipeline mit CaptchaAI

Legen Sie mit dem Aufbau wiederverwendbarer Client-Pipelines los – die API-Basis dafür finden Sie unter captchaai.com.

Verwandte Leitfäden

• CAPTCHAs parallel lösen
• Wiederholungslogik sauber implementieren
• Verteilte Verarbeitung mit Redis-Warteschlange
• Skript zur Health-Check-Überwachung