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

Wenn mehrere Dienste oder Teammitglieder eine CAPTCHA-Lösung benötigen, vermeidet die Zentralisierung in einem Microservice die Duplizierung der Logik über Projekte hinweg. Durch die asynchrone Unterstützung von FastAPI eignet es sich hervorragend – die CAPTCHA-Lösung erfordert das Warten auf externe API-Antworten, die asynchron effizient verarbeitet werden, ohne Threads zu blockieren.

Dieser Leitfaden erstellt einen FastAPI-Mikroservice, der CAPTCHA-Lösungsanfragen über REST akzeptiert und gelöste Token über CaptchaAI zurückgibt.

Wann lohnt sich ein Microservice wirklich?

Was Sie brauchen

Abhängigkeiten installieren:

pip install fastapi uvicorn httpx

Projektstruktur

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

CaptchaAI Solver-Modul

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

FastAPI-Anwendung

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

Führen Sie den Dienst aus

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

Anwendungsbeispiele

Lösen Sie reCAPTCHA v2

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

Lösen Sie Cloudflare Turnstile

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

Fehlerbehebung

FAQ

Warum FastAPI für einen CAPTCHA-Lösungsdienst verwenden?

FastAPI verarbeitet asynchrone I/O nativ, was ideal für die CAPTCHA-Lösung ist, bei der die meiste Zeit damit verbracht wird, auf die Antwort von CaptchaAI zu warten. Mehrere Anfragen können gleichzeitig ohne Threading verarbeitet werden.

Kann ich den Endpunkten eine Authentifizierung hinzufügen?

Ja. Fügen Sie die Abhängigkeitsinjektion von FastAPI mit API-Schlüsselheader-Validierung oder OAuth2 hinzu, um den Zugriff einzuschränken.

Wie viele gleichzeitige Anfragen können damit bearbeitet werden?

Begrenzt durch das gleichzeitige Aufgabenlimit Ihres CaptchaAI-Plans. FastAPI selbst kann Tausende gleichzeitiger Verbindungen verarbeiten.

Soll ich das andocken?

Ja. Fügen Sie ein Dockerfile mit FROM python:3.11-slim hinzu, installieren Sie Abhängigkeiten und machen Sie Port 8000 verfügbar.

Kann ich eine Ratenbegrenzung hinzufügen?

Ja. Verwenden Sie slowapi oder einen Reverse-Proxy (Nginx, Traefik), um die Anfragen pro Client zu begrenzen.

Erstellen Sie Ihren Microservice zur CAPTCHA-Lösung

Holen Sie sich Ihren API-Schlüssel unter captchaai.com.

Verwandte Leitfäden

• So lösen Sie reCAPTCHA v2 mithilfe der API
• So lösen Sie Cloudflare Turnstile mithilfe der API
• Async CaptchaAI mit aiohttp
• Parallele vs. sequenzielle CAPTCHA-Performance