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

Un microservice FastAPI expose la résolution de CAPTCHA derrière une seule API REST, que toutes vos applications appellent au lieu de recopier la même logique dans chaque projet. C'est le bon choix dès que plusieurs services — scraping, tests QA, back-office — partagent le même besoin de résolution et doivent rester cohérents.

FastAPI repose sur l'asynchrone, et la résolution de CAPTCHA passe l'essentiel de son temps à attendre la réponse d'une API externe. L'async traite donc des dizaines de requêtes en parallèle sans immobiliser un thread par appel. Ce guide construit ce microservice de bout en bout : il accepte les demandes de résolution en REST et renvoie les tokens résolus via CaptchaAI.

Microservice ou intégration directe : que choisir ?

Si vous débutez avec l'API, prenez d'abord en main l'authentification et la vérification du solde avant de découper la logique en service à part.

Prérequis

Installez les dépendances :

pip install fastapi uvicorn httpx

Arborescence du projet

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

Le module solveur CaptchaAI

Ce module isole toute la mécanique CaptchaAI : soumettre la tâche, interroger le résultat, puis exposer une fonction par type de CAPTCHA. Le reste de l'application n'a jamais à connaître les endpoints in.php / res.php.

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

Chaque fonction ajuste son initial_wait au temps de résolution attendu du type visé : plus court pour Turnstile et les images, plus long pour reCAPTCHA. Le polling interroge ensuite res.php toutes les 5 secondes jusqu'au token ou au timeout.

L'application FastAPI et ses endpoints

Chaque type de CAPTCHA reçoit son propre endpoint REST et son modèle Pydantic. Les erreurs remontées par le solveur (ValueError, TimeoutError) sont converties en réponse HTTP 502, ce qui donne aux applications clientes un contrat clair.

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

L'endpoint /health sert de sonde de disponibilité pour un orchestrateur (Docker, Kubernetes) ou un répartiteur de charge.

Lancer le service en local

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

Rendez-vous ensuite sur http://localhost:8000/docs : FastAPI génère automatiquement une documentation interactive de chaque endpoint.

Appeler le microservice : exemples

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

Injectez le token renvoyé dans le formulaire cible, exactement comme le montre le guide de résolution de reCAPTCHA v2 via l'API.

Déployer le microservice près de vos applications

En production, hébergez le service au plus près des applications qui l'appellent pour réduire la latence interne. Un déploiement sur OVHcloud, Scaleway ou une région AWS européenne comme eu-west-3 (Paris) reste un choix naturel pour une audience francophone. Conteneurisez main.py et solver.py, exposez le port 8000, puis placez le tout derrière un reverse proxy (nginx, Traefik) qui gère TLS et limitation de débit.

Côté conformité, restez sobre sur la journalisation : n'enregistrez ni les sitekey, ni les URL complètes, ni les tokens résolus. Journaliser des identifiants ou des données personnelles superflues alourdit vos obligations RGPD sans bénéfice opérationnel. Un identifiant de tâche et un statut suffisent pour le suivi.

Dépannage

FAQ

Quand préférer un microservice à une intégration directe ?

Dès que deux applications ou plus résolvent des CAPTCHA, ou que vous voulez un point unique pour la journalisation, les quotas et les retries. Pour un script isolé, l'intégration directe reste plus simple à maintenir.

Comment protéger le microservice en production ?

Ajoutez l'injection de dépendances de FastAPI pour valider un en-tête de clé API, ou branchez OAuth2. Complétez avec un reverse proxy (nginx, Traefik) et exposez le service uniquement sur votre réseau privé.

Le microservice prend-il en charge hCaptcha ou FunCaptcha ?

Non — hCaptcha et FunCaptcha ne sont pas pris en charge. Ajoutez des endpoints uniquement pour les types couverts par CaptchaAI : reCAPTCHA v2 et v3, Cloudflare Turnstile, GeeTest v3, et les CAPTCHA image/texte.

Quel plan CaptchaAI faut-il pour traiter plusieurs résolutions en parallèle ?

Le nombre de résolutions simultanées dépend des threads de votre plan, pas de FastAPI. BASIC ($15/mois, 5 threads) autorise 5 résolutions en vol ; STANDARD ($30/mois, 15 threads) et ADVANCE ($90/mois, 50 threads) montent en charge sans facturation à la résolution.

Faut-il conserver les tokens résolus pour les réutiliser ?

Non. Les tokens ont une durée de vie courte et sont liés à une session précise. Résolvez à la demande au moment de soumettre le formulaire, et laissez le token expirer plutôt que de le stocker.

Passez à l'implémentation

Obtenez votre clé API sur captchaai.com, puis centralisez la résolution de CAPTCHA derrière ce microservice FastAPI pour toutes vos applications.

Guides associés

• Résoudre reCAPTCHA v2 avec l'API
• Résoudre Cloudflare Turnstile avec l'API
• CaptchaAI en asynchrone avec aiohttp
• Résolution de CAPTCHA en parallèle