Erstellen Sie einen Microservice zur CAPTCHA-Lösung mit FastAPI und CaptchaAI

Ein eigener Solve-Service rechnet sich ab dem Moment, in dem der zweite Dienst denselben CaptchaAI-Schlüssel braucht. Statt Submit- und Polling-Logik in Scraper, Testsuite und Backend-Job zu kopieren, stellen Sie einen kleinen FastAPI-Dienst bereit: ein Endpunkt pro CAPTCHA-Typ, Sitekey und Page-URL rein, fertiges Token raus. Schlüssel, Timeouts und Fehlerbehandlung liegen danach an genau einer Stelle.

FastAPI ist dafür die passende Basis, weil beim Lösen fast die gesamte Zeit auf die Antwort von CaptchaAI entfällt. Mit async/await und httpx blockiert eine wartende Anfrage keinen Thread – ein einzelner uvicorn-Prozess hält problemlos viele offene Solve-Anfragen gleichzeitig.

Der Leitfaden baut den Dienst Schritt für Schritt auf – Solver-Modul, Anwendung, Beispielaufrufe – und behandelt danach Absicherung, Thread-Planung und Protokollierung.

Was der Dienst am Ende leistet

• POST /solve/recaptcha-v2 – reCAPTCHA v2, optional als Enterprise-Variante
• POST /solve/recaptcha-v3 – reCAPTCHA v3 inklusive action-Parameter
• POST /solve/turnstile – Cloudflare Turnstile
• POST /solve/image – Bild-CAPTCHAs als Base64-String
• GET /health – Health-Check für Loadbalancer und Container-Orchestrierung

Alle Solve-Endpunkte liefern dasselbe Antwortschema: das gelöste Token und, sofern mitgeliefert, den passenden User-Agent. Aufrufende Dienste kennen damit nur noch eine interne URL.

Zentraler Solve-Service oder Integration je Projekt?

Faustregel: Ab dem zweiten aufrufenden Dienst lohnt sich der Microservice. Der interne Hop kostet Millisekunden und fällt neben einer Lösungszeit von mehreren Sekunden kaum ins Gewicht.

Voraussetzungen

Abhängigkeiten installieren:

pip install fastapi uvicorn httpx

Projektstruktur

captcha-service/
├── main.py          # FastAPI app with endpoints
├── solver.py        # CaptchaAI solving logic
└── requirements.txt

Die Trennung ist Absicht: solver.py kennt nur CaptchaAI, main.py nur HTTP. So bleibt das Solver-Modul einzeln testbar und in Worker oder Cron-Jobs wiederverwendbar.

Solver-Modul: Aufgabe übermitteln, Ergebnis abfragen

Der Ablauf ist bei jedem Typ identisch: Aufgabe an in.php übermitteln, Task-ID entgegennehmen, das Ergebnis an res.php abfragen. Die Antwort CAPCHA_NOT_READY ist kein Fehler, sondern das Signal zum Weiterwarten.

# solver.py
import httpx
import asyncio

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


async def submit_task(params: dict) -> str:
    """Submit a CAPTCHA task and return the task ID."""
    params["key"] = API_KEY
    params["json"] = 1

    async with httpx.AsyncClient() as client:
        response = await client.post(f"{BASE_URL}/in.php", data=params)
        data = response.json()

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


async def poll_result(task_id: str, initial_wait: int = 15, max_attempts: int = 30) -> dict:
    """Poll for the CAPTCHA result."""
    await asyncio.sleep(initial_wait)

    async with httpx.AsyncClient() as client:
        for _ in range(max_attempts):
            response = await client.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1
            })
            data = response.json()

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

            await asyncio.sleep(5)

    raise TimeoutError("Solve timed out")


async def solve_recaptcha_v2(sitekey: str, pageurl: str, enterprise: bool = False) -> dict:
    params = {"method": "userrecaptcha", "googlekey": sitekey, "pageurl": pageurl}
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_recaptcha_v3(sitekey: str, pageurl: str, action: str, enterprise: bool = False) -> dict:
    params = {
        "method": "userrecaptcha", "version": "v3",
        "googlekey": sitekey, "pageurl": pageurl, "action": action
    }
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_turnstile(sitekey: str, pageurl: str) -> dict:
    task_id = await submit_task({"method": "turnstile", "sitekey": sitekey, "pageurl": pageurl})
    return await poll_result(task_id, initial_wait=10)


async def solve_image(image_base64: str) -> dict:
    task_id = await submit_task({"method": "base64", "body": image_base64})
    return await poll_result(task_id, initial_wait=5, max_attempts=15)

Die Werte für initial_wait folgen den typischen Lösungszeiten: Bild-CAPTCHAs sind in unter 0,5 Sekunden gelöst, Turnstile in unter 10 Sekunden, reCAPTCHA v2 in unter 60 Sekunden. Zu frühes Polling erzeugt nur überflüssige Anfragen an res.php.

FastAPI-Anwendung mit einem Endpunkt pro CAPTCHA-Typ

Jeder Endpunkt bekommt ein eigenes Pydantic-Modell: keine manuelle Validierung, automatische Dokumentation unter /docs, sofort sichtbare Tippfehler im Sitekey-Feld. Fehler aus dem Solver landen bewusst auf 502 – fehlgeschlagen ist der nachgelagerte Aufruf, nicht Ihr Dienst.

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import solver

app = FastAPI(title="CaptchaAI Solver Service")


class RecaptchaV2Request(BaseModel):
    sitekey: str
    pageurl: str
    enterprise: bool = False


class RecaptchaV3Request(BaseModel):
    sitekey: str
    pageurl: str
    action: str
    enterprise: bool = False


class TurnstileRequest(BaseModel):
    sitekey: str
    pageurl: str


class ImageRequest(BaseModel):
    image_base64: str


class SolveResponse(BaseModel):
    token: str
    user_agent: Optional[str] = ""


@app.post("/solve/recaptcha-v2", response_model=SolveResponse)
async def solve_recaptcha_v2(req: RecaptchaV2Request):
    try:
        result = await solver.solve_recaptcha_v2(req.sitekey, req.pageurl, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/recaptcha-v3", response_model=SolveResponse)
async def solve_recaptcha_v3(req: RecaptchaV3Request):
    try:
        result = await solver.solve_recaptcha_v3(req.sitekey, req.pageurl, req.action, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/turnstile", response_model=SolveResponse)
async def solve_turnstile(req: TurnstileRequest):
    try:
        result = await solver.solve_turnstile(req.sitekey, req.pageurl)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/image", response_model=SolveResponse)
async def solve_image(req: ImageRequest):
    try:
        result = await solver.solve_image(req.image_base64)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "ok"}

Dienst starten

uvicorn main:app --host 0.0.0.0 --port 8000

Lokal ergänzen Sie --reload, im Betrieb stattdessen mehrere Worker hinter einem Reverse-Proxy. Über http://localhost:8000/docs probieren Sie die Endpunkte ohne eigenen Client durch.

Beispielaufrufe

reCAPTCHA v2 lösen

curl -X POST http://localhost:8000/solve/recaptcha-v2 \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkS...", "pageurl": "https://example.com/login"}'

Cloudflare Turnstile lösen

curl -X POST http://localhost:8000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAA...", "pageurl": "https://example.com/form"}'

Antwort:

{
  "token": "03AGdBq24PBCqLmOx2V4...",
  "user_agent": "Mozilla/5.0..."
}

Das Token trägt der aufrufende Dienst in das vorgesehene Formularfeld ein und sendet das Formular ab. Fordern Sie es erst kurz vor der Übermittlung an – Tokens sind nur wenige Minuten gültig.

Betrieb: Absicherung, Limits, Deployment

• Nicht öffentlich exponieren. Der Dienst gehört ins interne Netz oder hinter ein VPN; er trägt Ihren API-Schlüssel.
• Authentifizierung ergänzen. Eine FastAPI-Dependency, die einen Header mit internem Schlüssel prüft, genügt meist; OAuth2 nur, wenn Sie es ohnehin betreiben.
• Rate-Limiting setzen. slowapi oder Anfragedrosselung im Reverse-Proxy (Nginx, Traefik) verhindert, dass ein fehlerhafter Client alle Threads belegt.
• Als Container ausliefern. Ein Dockerfile auf Basis von python:3.11-slim mit Port 8000 reicht; gebaut in GitLab CI, betrieben auf einer kleinen Instanz bei Hetzner, IONOS oder netcup.
• Sinnvoll protokollieren. Task-ID, CAPTCHA-Typ, Dauer und Statuscode genügen für die Fehlersuche.

Threads planen: wie viele parallele Solves der Dienst verträgt

Das Limit setzt in der Praxis nicht FastAPI, sondern Ihr Plan. CaptchaAI rechnet Thread-basiert ab: Ein Thread ist eine gleichzeitig laufende Lösung; die Zahl der Lösungen pro Thread ist im Abrechnungsmonat nicht gedeckelt. BASIC (15 $/Monat, 5 Threads) trägt einen nächtlichen Testlauf, ADVANCE (90 $/Monat, 50 Threads) mehrere parallele Scraping-Worker. Die Preise sind in US-Dollar angegeben.

Begrenzen Sie die gleichzeitigen Aufrufe deshalb im Dienst – etwa über ein asyncio.Semaphore – auf die Thread-Zahl Ihres Plans, statt Überlast weiterzureichen. Rechenbeispiel: CaptchaAI löst Turnstile in unter 10 Sekunden; fünf Threads ergeben damit rechnerisch rund 30 Lösungen pro Minute, real liegt der Wert wegen Latenz und Polling-Intervall darunter.

Protokollierung und DSGVO

Der Dienst sieht Page-URLs, Sitekeys und teilweise User-Agents. Page-URLs enthalten häufig Session-IDs oder Kundennummern, IP-Adressen gelten in der EU als personenbezogene Daten – ein Solve-Service, der jede Anfrage vollständig mitschreibt, wird schnell zum ungeplanten Datenspeicher.

Praktisch heißt das: Query-Strings vor dem Logging abschneiden, eine Aufbewahrungsfrist festlegen und den Aufruf des externen Dienstleisters im Verarbeitungsverzeichnis erfassen. Prüfen Sie die Datenflüsse mit den Verantwortlichen in Ihrem Haus.

Fehlerbilder und ihre Ursachen

Häufige Fragen

Wie lange bleibt ein gelöstes Token gültig?

Nur wenige Minuten. Rufen Sie den Solve-Endpunkt daher erst auf, wenn das Formular ausgefüllt ist, und legen Sie keine Token-Vorräte an – abgelaufene Token weist die Zielseite zurück.

Was tun, wenn der Dienst häufiger in ein Timeout läuft?

Zuerst max_attempts und initial_wait an den CAPTCHA-Typ anpassen. Bleibt es dabei, wiederholen Sie den Aufruf einmal mit exponentiellem Backoff, statt die Wartezeit pauschal zu verlängern – ein hängender Task blockiert sonst einen Thread.

Kann der Dienst auch Bild-CAPTCHAs aus einem Screenshot lösen?

Ja, über POST /solve/image. Sie schneiden das CAPTCHA-Bild aus, kodieren es als Base64 und senden es im Feld image_base64; Rasterbild- und BLS-Aufgaben laufen ebenfalls auf diesem Weg.

Löst der Dienst auch hCaptcha?

Nein. hCaptcha und FunCaptcha unterstützt CaptchaAI nicht, GeeTest v4 ist als bald verfügbar angekündigt. Abgedeckt sind reCAPTCHA v2/v3 inklusive Enterprise, Cloudflare Turnstile und Challenge, GeeTest v3, Bild-, Raster- und BLS-CAPTCHAs; CaptchaFox, Friendly Captcha und Lemin laufen in der Beta.

Wie teste ich den Dienst ohne echte CAPTCHA-Seite?

Über GET /health für den Deployment-Check und einen Testdatensatz für die Solver-Funktionen. Für Integrationstests eignet sich eine eigene Staging-Seite mit einem Test-Sitekey besser als eine fremde Produktivseite.

Nächster Schritt

Holen Sie sich Ihren API-Schlüssel unter captchaai.com und bringen Sie den Dienst mit den beiden Dateien aus diesem Leitfaden zum Laufen – das erste gelöste Token liefert er wenige Minuten später.

Verwandte Leitfäden

• reCAPTCHA v2 per API lösen
• Cloudflare Turnstile per API lösen
• CaptchaAI asynchron mit aiohttp nutzen
• Parallel oder sequenziell lösen: Performance im Vergleich