Integrasi HTTPX + CaptchaAI

httpx memungkinkan Anda mengirim puluhan request penyelesaian CAPTCHA secara paralel dari satu skrip Python — sesuatu yang canggung dilakukan dengan requests biasa. Panduan ini menghubungkan httpx dengan API CaptchaAI lewat dua jalur: klien sinkron untuk skrip sederhana, dan klien async untuk solve konkuren berskala besar, keduanya dengan dukungan HTTP/2 opsional.

Empat langkah intinya selalu sama, sama seperti integrasi CaptchaAI lainnya:

1. Kirim task ke in.php beserta parameter CAPTCHA.
2. Simpan task ID dari respons OK|.
3. Polling ke res.php sampai token siap.
4. Pakai token pada form atau request halaman target.

Yang berbeda hanya klien HTTP-nya. CaptchaAI menyelesaikan reCAPTCHA v2/v3, Cloudflare Turnstile dan Challenge, GeeTest v3, serta CAPTCHA image/OCR lewat endpoint dan pola polling yang sama — jadi satu kelas klien httpx cukup untuk semua tipe. Anda hanya mengganti nilai method dan parameter sitekey sesuai target.

Yang perlu disiapkan

pip install httpx

Klien httpx sinkron untuk solve CAPTCHA

Klien sinkron cocok untuk skrip satu-jalur: satu CAPTCHA dikirim, di-polling, lalu tokennya dipakai sebelum lanjut. Kelas berikut membungkus submit, polling, dan cek saldo dalam satu objek yang bisa dipakai ulang.

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

API key selalu dibaca dari environment variable, tidak ditulis langsung di kode. Nilai googlekey di atas adalah sitekey reCAPTCHA v2 dari halaman target Anda. Method get_balance memakai action=getbalance untuk membaca saldo thread tanpa mengirim solve — berguna untuk memasang peringatan sebelum kuota bulanan menipis. Karena httpx.Client menyimpan connection pool, buat satu instance saja, pakai ulang di seluruh skrip, lalu panggil close() ketika selesai agar koneksi ditutup rapi.

Klien httpx async untuk solve konkuren

Di sinilah httpx unggul. Dengan AsyncClient, Anda bisa menjalankan banyak penyelesaian sekaligus lewat asyncio.gather(), alih-alih menunggu satu per satu. Struktur kelasnya identik dengan versi sinkron, hanya berpindah ke async/await.

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

Perhatikan return_exceptions=True pada gather(): satu solve yang gagal — misalnya karena sitekey keliru — tidak menjatuhkan seluruh batch. Anda memeriksa tiap hasil dan hanya menangani yang berupa Exception, sehingga empat dari lima halaman tetap terselesaikan meski satu bermasalah.

Satu catatan penting untuk tim yang menjalankan volume besar: jumlah solve konkuren yang benar-benar berjalan dibatasi oleh alokasi thread paket CaptchaAI Anda, bukan oleh httpx. Paket BASIC ($15/bulan, 5 thread) menjalankan 5 penyelesaian sekaligus; ADVANCE ($90/bulan, 50 thread) menjalankan 50, dengan solve tak terbatas per thread. Jadi jika sebuah worker scraping di region ap-southeast-3 (Jakarta) menembak gather() untuk 100 halaman sekaligus, sesuaikan tingkat konkurensi dengan jumlah thread paket agar request tidak menumpuk di antrean.

Mengaktifkan HTTP/2 di httpx

httpx mendukung HTTP/2, yang mengurangi overhead koneksi:

pip install httpx[http2]

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

HTTP/2 mem-multiplex banyak request melalui satu koneksi. Manfaatnya paling terasa pada:

• submit dan polling beberapa CAPTCHA secara bersamaan,
• worker async yang menjaga satu koneksi tetap hidup untuk banyak solve.

Koneksi tunggal itu dipakai ulang, bukan dibuka-tutup per request, sehingga overhead handshake berkurang.

Contoh scraping dengan penanganan CAPTCHA

Pola paling umum: ambil halaman, deteksi sitekey reCAPTCHA, kirim ke CaptchaAI, lalu POST form dengan token pada field 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())

Kumpulkan hanya data yang memang berhak Anda proses; untuk kerja scraping di Indonesia, UU Pelindungan Data Pribadi (UU 27/2022) membuat kehati-hatian soal data pribadi menjadi relevan.

httpx vs requests vs aiohttp

Singkatnya, httpx memberi Anda API yang terasa seperti requests tetapi dengan async dan HTTP/2 sekaligus. Untuk integrasi CaptchaAI baru, ini titik awal yang praktis. Pilih klien sesuai kebutuhan:

• Klien sinkron — skrip sekali-jalan, cron sederhana, atau satu CAPTCHA per eksekusi; kodenya lebih pendek dan mudah di-debug.
• Klien async — banyak CAPTCHA dalam satu proses, misalnya menelusuri ratusan halaman produk secara paralel.

Karena keduanya memakai kelas yang hampir identik, berpindah antar keduanya nyaris tanpa biaya rekayasa ulang.

Pertanyaan umum

Berapa banyak CAPTCHA yang bisa saya solve sekaligus dengan httpx async?

httpx sendiri tidak membatasi; batas nyatanya adalah alokasi thread paket CaptchaAI Anda. 5 thread berarti 5 penyelesaian in-flight sekaligus. Atur jumlah task dalam gather() agar sejalan dengan jumlah thread paket.

Apakah HTTP/2 mempercepat penyelesaian CAPTCHA?

HTTP/2 tidak mempercepat solve di sisi server, tetapi mengurangi overhead jaringan saat Anda mengirim dan polling banyak CAPTCHA lewat satu koneksi. Manfaatnya muncul pada beban konkuren, bukan pada satu solve tunggal.

Bagaimana mengatur timeout agar polling tidak menggantung?

Ada dua batas waktu terpisah: timeout pada klien httpx (per request) dan argumen timeout pada method solve (total anggaran polling, default 300 detik). Sesuaikan yang kedua dengan tipe CAPTCHA — reCAPTCHA biasanya selesai jauh lebih cepat.

Bisakah saya memigrasi kode requests yang sudah ada ke httpx?

Bisa, dan biasanya mudah. httpx sengaja meniru API requests, jadi banyak pemanggilan get/post berjalan tanpa perubahan. Setelah itu Anda bisa memindahkan bagian yang butuh konkurensi ke AsyncClient secara bertahap.

Bisakah klien yang sama menyelesaikan Turnstile atau GeeTest, bukan cuma reCAPTCHA?

Bisa. Kelas klien yang sama bekerja untuk semua tipe yang didukung CaptchaAI — cukup ubah method menjadi turnstile atau geetest lalu kirim parameter yang sesuai. Pola submit, polling, dan pemakaian token tidak berubah sama sekali.

Panduan Terkait

• Integrasi aiohttp async dengan CaptchaAI
• Axios + CaptchaAI tanpa browser
• Panduan scraping CAPTCHA dengan Python