Intégration HTTPX + CaptchaAI

httpx appelle l'API CaptchaAI en synchrone comme en asynchrone depuis un seul client — idéal pour résoudre plusieurs CAPTCHA en parallèle. Contrairement à requests, qui traite un appel à la fois, httpx apporte l'async natif et HTTP/2, deux atouts directs pour le débit de résolution. Ce guide couvre les deux modes, le dimensionnement en threads, la gestion des erreurs et un exemple de scraping de bout en bout.

Prérequis

pip install httpx

Pourquoi httpx plutôt que requests

requests reste parfait pour un script séquentiel, mais il n'a ni async ni HTTP/2. Or la résolution d'un CAPTCHA passe l'essentiel de son temps à attendre : vous soumettez la tâche, puis vous interrogez le résultat toutes les quelques secondes. httpx vous laisse lancer des dizaines de résolutions de front avec asyncio, tout en gardant une API quasi identique à requests — la migration est indolore.

Client synchrone : le premier appel bloquant

Pour un script simple ou un job planifié, le client synchrone suffit : il soumet le CAPTCHA à in.php, interroge res.php toutes les 5 secondes jusqu'au token, et expose le solde du compte.

import httpx
import time
import os


class CaptchaAISync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.Client(timeout=30)

    def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = self.client.get(f"{self.base_url}/in.php", params=params)
        text = resp.text

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

        task_id = text.split("|")[1]

        # Poll
        deadline = time.time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while time.time() < deadline:
            time.sleep(5)
            result = self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

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

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

    def get_balance(self):
        resp = self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    def close(self):
        self.client.close()


# Usage
solver = CaptchaAISync(os.environ["CAPTCHAAI_API_KEY"])

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

La clé vient de la variable d'environnement CAPTCHAAI_API_KEY : ne la codez jamais en dur dans le script. Ici method=userrecaptcha cible reCAPTCHA v2, mais le même client résout Cloudflare Turnstile, GeeTest v3 ou l'OCR d'image simplement en changeant les paramètres soumis. Le timeout=300 borne la boucle d'interrogation : au-delà, la méthode lève une TimeoutError que votre code appelant peut rattraper pour réessayer.

Client asynchrone : plusieurs CAPTCHA en parallèle

Dès que vous traitez plusieurs pages, la version asynchrone change la donne. asyncio.gather lance toutes les résolutions de front et récupère chaque token dès qu'il est prêt, au lieu de les enchaîner une par une.

import httpx
import asyncio
import os


class CaptchaAIAsync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.AsyncClient(timeout=30)

    async def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = await self.client.get(
            f"{self.base_url}/in.php", params=params
        )
        text = resp.text

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

        task_id = text.split("|")[1]

        # Poll
        deadline = asyncio.get_event_loop().time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)
            result = await self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

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

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

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

    async def close(self):
        await self.client.aclose()


# Usage
async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])

    # Solve multiple concurrently
    tasks = [
        solver.solve({
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": f"https://example.com/page{i}",
        })
        for i in range(5)
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Page {i}: FAILED - {r}")
        else:
            print(f"Page {i}: solved ({len(r)} chars)")

    await solver.close()

asyncio.run(main())

return_exceptions=True est le détail qui compte en production : un CAPTCHA en échec ne fait pas tomber tout le lot, chaque résultat est inspecté individuellement. Vous journalisez les échecs et réessayez seulement les pages concernées.

Concurrence et threads : dimensionner votre plan

La concurrence de votre code doit rester alignée sur votre allocation de threads CaptchaAI. La facturation se fait par thread simultané, avec des résolutions illimitées par thread : un thread correspond à un CAPTCHA en cours, et il se libère dès que la résolution se termine. Les 5 résolutions lancées ci-dessus tiennent donc exactement dans le plan BASIC ($15/mois, 5 threads). Passez la boucle à 15 tâches simultanées et il vous faut STANDARD ($30/mois, 15 threads) ; pour des pipelines plus denses, ADVANCE ($90/mois, 50 threads) laisse de la marge. Dépasser vos threads ne casse rien : les résolutions en trop attendent simplement qu'un thread se libère.

Activer HTTP/2 pour réduire la latence

httpx prend en charge HTTP/2, ce qui réduit la surcharge de connexion :

pip install httpx[http2]

client = httpx.AsyncClient(http2=True, timeout=30)

HTTP/2 multiplexe les requêtes sur une seule connexion : la soumission et les interrogations répétées de plusieurs CAPTCHA partagent le même tunnel. Le gain se ressent surtout quand vos workers tournent sur une région européenne proche — par exemple eu-west-3 (Paris) ou un hébergeur comme OVHcloud ou Scaleway — où chaque aller-retour économisé raccourcit le temps de bout en bout.

Scraping avec résolution de CAPTCHA : exemple concret

Voici le schéma complet : récupérer la page, détecter le sitekey reCAPTCHA, envoyer la résolution à CaptchaAI, puis renvoyer le formulaire avec le token dans le champ g-recaptcha-response.

import httpx
import re
import os

async def scrape_with_captcha(url, solver):
    async with httpx.AsyncClient() as client:
        # Fetch page
        resp = await client.get(url)
        html = resp.text

        # Check for reCAPTCHA
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        if not match:
            return html

        site_key = match.group(1)
        token = await solver.solve({
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

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


async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
    content = await scrape_with_captcha("https://example.com", solver)
    print(f"Got {len(content)} chars")
    await solver.close()

asyncio.run(main())

Si votre scraping touche des données personnelles, gardez le réflexe RGPD : minimisez les champs collectés avant tout stockage. Côté token, ne constituez pas de réserve : il n'est valable que quelques minutes, donc soumettez le formulaire aussitôt après la résolution.

Gérer les erreurs et les délais

Trois cas reviennent en boucle. Tant que la résolution n'est pas prête, l'API renvoie CAPCHA_NOT_READY : c'est normal, la boucle continue d'interroger. Si la soumission échoue d'emblée, la réponse ne commence pas par OK| — le plus souvent une clé API absente, un solde à zéro ou un googlekey erroné. Enfin, si aucun token n'arrive avant le timeout, la méthode lève une TimeoutError.

En production, encadrez chaque solve() d'une nouvelle tentative avec un backoff court, et vérifiez le solde au démarrage via get_balance() pour ne pas lancer un lot voué à l'échec. Deux ou trois tentatives suffisent : au-delà, le problème vient de la page cible, pas du service.

Dépannage

httpx, requests ou aiohttp : lequel choisir ?

FAQ

Combien de CAPTCHA puis-je résoudre en parallèle avec httpx async ?

Autant que votre plan autorise de threads. asyncio.gather peut lancer des centaines de coroutines, mais le débit réel est plafonné par vos threads : 5 en simultané sur BASIC, 15 sur STANDARD, 50 sur ADVANCE.

Que faire quand l'API renvoie CAPCHA_NOT_READY ?

Rien de spécial : c'est la réponse normale tant que la résolution est en cours. La boucle d'interrogation continue jusqu'au token ou jusqu'au timeout. N'interrogez pas plus vite que toutes les 5 secondes environ, sinon vous multipliez les requêtes sans accélérer la résolution.

Quel plan CaptchaAI choisir pour du solving asynchrone ?

Comptez le nombre de résolutions réellement simultanées, pas le nombre total. Un scraper qui garde 10 pages ouvertes à la fois tient dans STANDARD ($30/mois, 15 threads).

Puis-je réutiliser un seul client httpx pour tout mon script ?

Oui, c'est recommandé. Instanciez le client une fois et gardez-le ouvert : le pool de connexions est réutilisé et évite un handshake TLS à chaque requête. Fermez-le à la fin avec close() ou aclose().

httpx fonctionne-t-il avec Scrapy ?

Pas directement : Scrapy repose sur la boucle d'événements de Twisted. Utilisez httpx dans des scripts autonomes ou avec un framework asyncio comme FastAPI.

Guides connexes

• Résolution asynchrone avec aiohttp
• Appeler CaptchaAI avec Axios, sans navigateur
• Scraping de CAPTCHA en Python, pas à pas