Créez un microservice de résolution de CAPTCHA avec FastAPI et CaptchaAI

Lorsque plusieurs services ou membres de l'équipe ont besoin d'une résolution CAPTCHA, la centraliser dans un microservice évite la duplication de la logique entre les projets. La prise en charge asynchrone de FastAPI en fait un choix idéal : la résolution de CAPTCHA implique d'attendre les réponses d'API externes, que l'asynchrone gère efficacement sans bloquer les threads.

Ce guide crée un microservice FastAPI qui accepte les requêtes de résolution CAPTCHA via REST et renvoie les jetons résolus via CaptchaAI.

Quand un microservice FastAPI vaut-il l'effort ?

Ce dont vous avez besoin

Installer les dépendances :

pip install fastapi uvicorn httpx

Structure du projet

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

Module solveur CaptchaAI

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

Application FastAPI

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

Exécuter le service

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

Exemples d'utilisation

Résoudre 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"}'

Résoudre Cloudflare Turnstile

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

Réponse :

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

Dépannage

FAQ

Pourquoi utiliser FastAPI pour un service de résolution de CAPTCHA ?

FastAPI gère nativement I/O asynchrone, ce qui est idéal pour la résolution de CAPTCHA où la plupart du temps est passé à attendre la réponse de CaptchaAI. Plusieurs requêtes peuvent être traitées simultanément sans thread.

Puis-je ajouter une authentification aux points de terminaison ?

Oui. Ajoutez l'injection de dépendances de FastAPI avec la validation d'en-tête de clé API ou OAuth2 pour restreindre l'accès.

Combien de requêtes simultanées cela peut-il gérer ?

Limité par la limite de tâches simultanées de votre plan CaptchaAI. FastAPI lui-même peut gérer des milliers de connexions simultanées.

Dois-je docker cela ?

Oui. Ajoutez un Dockerfile avec FROM python:3.11-slim, installez les dépendances et exposez le port 8000.

Puis-je ajouter une limitation de débit ?

Oui. Utilisez slowapi ou un proxy inverse (nginx, Traefik) pour limiter les requêtes par client.

Créez votre microservice de résolution de CAPTCHA

Obtenez votre clé API surcaptchaai.com. Centralisez la résolution de CAPTCHA avec un microservice FastAPI.

Guides associés

• Comment résoudre reCAPTCHA v2 à l'aide de l'API
• Comment résoudre Cloudflare Turnstile à l'aide de l'API
• Asynchrone CaptchaAI avec aiohttp
• Résolution parallèle de CAPTCHA