aiohttp + CaptchaAI: Asynchrone CAPTCHA-Lösung

Der Engpass beim automatisierten CAPTCHA-Lösen ist selten die Rechenzeit, sondern das Warten. Zwischen der Übermittlung an in.php und dem fertigen Token vergehen je nach Typ wenige Sekunden bis knapp eine Minute – Zeit, in der ein synchrones Skript blockiert. Mit aiohttp warten zwanzig Aufgaben nebeneinander statt nacheinander, und die Gesamtlaufzeit richtet sich nach der langsamsten Lösung, nicht nach ihrer Summe.

Dieser Leitfaden zeigt einen vollständigen asynchronen CaptchaAI-Client: übermitteln, Status abfragen, Token weiterverwenden. Dazu kommt, was im Dauerbetrieb zählt – Parallelität sinnvoll begrenzen, Ausnahmen einzelner Aufgaben abfangen und denselben Client für reCAPTCHA v2 wie für Cloudflare Turnstile nutzen.

Warum asynchron: Das Polling bestimmt die Laufzeit

Der Ablauf besteht immer aus zwei Anfragen und einer Wartezeit dazwischen. Sie übermitteln Sitekey und Page-URL an in.php und erhalten eine Task-ID; danach fragen Sie res.php im Sekundentakt ab, bis das Token vorliegt. reCAPTCHA v2 wird typischerweise in unter 60 Sekunden gelöst, Cloudflare Turnstile in unter 10 Sekunden – der Prozess selbst ist in dieser Zeit vollständig untätig.

Genau dieses Warten lässt sich überlappen. asyncio gibt die Kontrolle während await an den Event-Loop zurück, sodass ein einzelner Python-Prozess dutzende Lösungen gleichzeitig verfolgt, ohne Threads oder zusätzliche Worker-Prozesse. aiohttp liefert dazu den passenden HTTP-Client mit Verbindungspooling. Wer denselben Ablauf lieber synchron und asynchron aus einer Codebasis fahren möchte, findet die Alternative im Leitfaden zur HTTPX-Integration.

Voraussetzungen

pip install aiohttp

Der Schlüssel gehört nicht in den Quellcode, sondern in die Umgebungsvariable CAPTCHAAI_API_KEY – alle folgenden Beispiele lesen ihn von dort.

Der asynchrone CaptchaAI-Client

Die Klasse kapselt den kompletten Ablauf: submit() übermittelt die Aufgabe und liefert die Task-ID, poll() fragt das Ergebnis ab, solve() verbindet beides, get_balance() liest das Guthaben.

import aiohttp
import asyncio


class AsyncCaptchaAI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    async def submit(self, session, params):
        """Submit a CAPTCHA task and return the task ID."""
        params["key"] = self.api_key
        async with session.get(
            f"{self.base_url}/in.php", params=params
        ) as resp:
            text = await resp.text()

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        return text.split("|")[1]

    async def poll(self, session, task_id, timeout=300):
        """Poll for the result with a timeout."""
        params = {
            "key": self.api_key,
            "action": "get",
            "id": task_id,
        }
        deadline = asyncio.get_event_loop().time() + timeout

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)

            async with session.get(
                f"{self.base_url}/res.php", params=params
            ) as resp:
                text = await resp.text()

            if text == "CAPCHA_NOT_READY":
                continue
            if text.startswith("OK|"):
                return text.split("|", 1)[1]
            raise Exception(f"Solve failed: {text}")

        raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

    async def solve(self, session, params, timeout=300):
        """Submit and poll in one call."""
        task_id = await self.submit(session, params)
        return await self.poll(session, task_id, timeout)

    async def get_balance(self, session):
        """Check account balance."""
        params = {"key": self.api_key, "action": "getbalance"}
        async with session.get(
            f"{self.base_url}/res.php", params=params
        ) as resp:
            return float(await resp.text())

Zwei Details sind wichtiger, als sie aussehen. CAPCHA_NOT_READY ist kein Fehler, sondern der erwartete Zustand in den ersten Sekunden – nur eine Antwort, die weder so lautet noch mit OK| beginnt, rechtfertigt einen Abbruch. Und die Session wird bewusst von außen übergeben: Ein einziges ClientSession-Objekt bedient alle Aufgaben und hält den Verbindungspool warm.

Ein einzelnes CAPTCHA lösen

Der Einstieg besteht aus einem Guthaben-Check als kurzem Health-Check und einer reCAPTCHA-v2-Aufgabe mit method, googlekey und pageurl.

import asyncio
import os

async def main():
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Check balance
        balance = await solver.get_balance(session)
        print(f"Balance: ${balance:.2f}")

        # Solve reCAPTCHA v2
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": "https://example.com",
        })
        print(f"Token: {token[:50]}...")

asyncio.run(main())

Das zurückgegebene Token tragen Sie anschließend als g-recaptcha-response in das Formular ein und senden es ab. Lösen und Absenden gehören in denselben Task: reCAPTCHA-Tokens laufen nach rund zwei Minuten ab.

Mehrere CAPTCHAs gleichzeitig lösen

asyncio.gather() startet alle Aufgaben zusammen und liefert die Ergebnisse in der Reihenfolge der Eingabeliste zurück.

async def solve_batch(urls, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        tasks = [
            solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })
            for url in urls
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"FAILED {url}: {result}")
            else:
                print(f"SOLVED {url}: {len(result)} chars")

        return results


urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    "https://example.com/page4",
    "https://example.com/page5",
]
asyncio.run(solve_batch(urls, "6Le-wvkS..."))

return_exceptions=True ist hier die halbe Fehlerbehandlung: Ohne diesen Parameter reißt eine einzige gescheiterte Aufgabe den gesamten gather()-Aufruf ab. So bleibt der Rest des Stapels intakt, und Sie protokollieren pro URL, was tatsächlich passiert ist.

Nur lösen, wenn die Seite ein CAPTCHA zeigt

In der Praxis liefert nicht jede Seite eine CAPTCHA-Abfrage aus. Der folgende Ablauf holt zuerst das HTML, prüft auf den reCAPTCHA-Marker und ruft den Solver nur im Bedarfsfall auf – das spart Threads und Laufzeit.

async def scrape_with_captcha(url, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Fetch the page
        async with session.get(url) as resp:
            html = await resp.text()

        # Check if page has a CAPTCHA
        if "g-recaptcha" not in html:
            return html  # No CAPTCHA, return content

        # Solve the CAPTCHA
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit with solved token
        async with session.post(url, data={
            "g-recaptcha-response": token,
        }) as resp:
            return await resp.text()

Für den produktiven Einsatz kommen eine Wiederholungslogik mit exponentiellem Backoff für Netzwerkfehler und eine saubere Auswertung der res.php-Fehlercodes dazu: Ein Konto-Problem behandeln Sie anders als eine Zeitüberschreitung.

Parallelität begrenzen: Semaphore und Thread-Kontingent

Wie viele Lösungen wirklich gleichzeitig laufen, entscheidet nicht asyncio, sondern Ihr Tarif. CaptchaAI rechnet Thread-basiert ab – ein Thread ist ein CAPTCHA in Bearbeitung, die Zahl der Lösungen je Thread ist im Abrechnungsmonat nicht gedeckelt. BASIC (15 $/Monat) stellt 5 Threads bereit, STANDARD (30 $/Monat) 15, ADVANCE (90 $/Monat) 50 und PREMIUM (170 $/Monat) 100. Setzen Sie max_concurrent auf Ihre Thread-Zahl, statt eine offene Liste in gather() zu werfen.

async def solve_with_limit(urls, site_key, max_concurrent=10):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
    semaphore = asyncio.Semaphore(max_concurrent)

    async def solve_one(session, url):
        async with semaphore:
            return await solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })

    async with aiohttp.ClientSession() as session:
        tasks = [solve_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    solved = sum(1 for r in results if not isinstance(r, Exception))
    print(f"Solved {solved}/{len(urls)} CAPTCHAs")
    return results

Die Semaphore hält die Warteschlange im Prozess, statt sie an die API weiterzureichen. Für eine Liste mit 500 URLs und 15 Threads heißt das: 15 Lösungen laufen, der Rest wartet geordnet.

Turnstile lösen: Nur die Payload ändert sich

Der Client bleibt unverändert, Sie tauschen lediglich die Parameter aus. Statt googlekey erwartet Turnstile den sitekey.

async def solve_turnstile(url, sitekey):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        token = await solver.solve(session, {
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": url,
        })
        return token

Nach demselben Muster laufen weitere Typen über dieselbe solve()-Methode:

Nicht unterstützt sind hCaptcha und FunCaptcha; GeeTest v4 ist bislang lediglich als bald verfügbar angekündigt. CaptchaFox (Beta), Friendly Captcha (Beta) und Lemin (Beta) sind über eigene method-Werte erreichbar.

Praxisbeispiel: nächtlicher Aggregationslauf

Eine Hamburger Agentur zieht jede Nacht öffentlich zugängliche Angebotsseiten von rund 15 Partnerportalen zusammen. Etwa jedes dritte Portal antwortet mit einer Turnstile-Abfrage. Der Job läuft als Container auf einem kleinen netcup-Server, aiohttp verarbeitet die Portale parallel, eine Semaphore mit 15 Slots deckt sich exakt mit dem Thread-Kontingent. Da Turnstile typischerweise in unter 10 Sekunden gelöst wird, endet der komplette Lauf in wenigen Minuten – ohne Headless-Browser, also mit deutlich geringerem Speicherbedarf auf der Instanz.

Zwei Punkte gehören im deutschsprachigen Umfeld in die Planung: IP-Adressen gelten nach DSGVO als personenbezogene Daten, sodass Protokollierung und Aufbewahrung Ihrer Abrufdaten dokumentiert sein sollten – und Rechnungsbeträge fallen in US-Dollar an, der Euro-Gegenwert schwankt mit dem Wechselkurs.

Fehler richtig einordnen

FAQ

Wie viele Lösungen darf ich gleichzeitig starten?

So viele, wie Ihr Tarif an Threads bereitstellt: 5 bei BASIC, 15 bei STANDARD, 50 bei ADVANCE. Die Zahl der Lösungen je Thread ist im Monat nicht begrenzt – begrenzt ist nur, wie viele davon zeitgleich laufen. Eine asyncio.Semaphore in Höhe dieser Zahl hält gather() verlässlich in diesem Rahmen.

Was passiert, wenn eine einzelne Aufgabe im gather() fehlschlägt?

Mit return_exceptions=True landet die Ausnahme als Ergebnis in der Liste, alle übrigen Aufgaben laufen weiter. Ohne den Parameter bricht der gesamte Aufruf beim ersten Fehler ab. Prüfen Sie jedes Ergebnis mit isinstance(result, Exception) und wiederholen Sie gezielt nur die betroffenen URLs.

Blockiert await asyncio.sleep(5) beim Abfragen die anderen Aufgaben?

Nein. asyncio.sleep() gibt die Kontrolle an den Event-Loop zurück, sodass in dieser Zeit andere Aufgaben ihre HTTP-Anfragen abarbeiten. Blockierend wäre nur time.sleep() – dieser Aufruf hat in asynchronem Code nichts zu suchen.

Brauche ich zusätzlich einen Headless-Browser?

Nein. Der Client spricht ausschließlich HTTP: Sitekey und Page-URL übermitteln, Token abholen, Formular absenden. Ein Browser wird erst nötig, wenn die Zielanwendung ihre Formulardaten selbst per JavaScript zusammenbaut.

Warum meldet der Code in Jupyter Event loop is closed?

Weil Notebooks bereits einen Event-Loop betreiben und asyncio.run() einen zweiten anlegen will. Rufen Sie in Notebooks die Coroutine direkt mit await auf oder installieren Sie nest_asyncio.

Verwandte Leitfäden

• CAPTCHAs mit HTTPX lösen
• Parallel oder sequenziell: Durchsatz im Vergleich
• Scrapy-Spider mit CAPTCHA-Handling