Integrations

Bangun Microservice Solve CAPTCHA dengan FastAPI dan CaptchaAI

Python Otomatisasi Semua Tipe CAPTCHA
Oleh Tim CaptchaAI Indonesia
Diterbitkan pada May 31, 2026
Bangun Microservice Solve CAPTCHA dengan FastAPI dan CaptchaAI
Gambar unggulan tidak tersedia

Begitu proyek automation tim Anda bertambah dari satu ke tiga, empat, lima repo, pola yang sama selalu muncul: tiap service menulis ulang logika submit-poll-retry ke CaptchaAI sendiri-sendiri, dan API key ikut tersebar ke setiap codebase. Jawabannya bukan menulis ulang lagi di proyek keenam, melainkan menaruh semua logika solve CAPTCHA di satu microservice FastAPI yang dipanggil lewat REST oleh service lain mana pun.

Intinya: satu microservice, satu API key, satu plan thread CaptchaAI — service lain tinggal panggil endpoint REST-nya, tidak perlu implementasi submit-poll sendiri-sendiri lagi.

Kapan Pola Microservice Ini Masuk Akal

FastAPI cocok untuk peran ini karena dukungan async-nya native — sebagian besar waktu solve CAPTCHA dihabiskan menunggu respons CaptchaAI, bukan memakai CPU, jadi event loop tetap bisa melayani request lain sambil menunggu. Panduan ini membangun microservice tersebut dari nol: endpoint REST yang submit ke CaptchaAI, polling hasil, lalu mengembalikan token yang sudah selesai untuk reCAPTCHA v2/v3, Cloudflare Turnstile, dan CAPTCHA gambar.

Tanda-tanda pola ini cocok dipakai tim Anda:

  1. Lebih dari satu service atau proyek klien butuh solve CAPTCHA — misalnya agensi otomatisasi kecil yang mengerjakan beberapa proyek scraping dan form-testing sekaligus.
  2. API key CaptchaAI mulai tersebar di banyak repo berbeda, dan tidak ada satu tempat untuk mengubah logika retry.
  3. Anda ingin satu plan thread dipakai bersama oleh semua proyek, bukan beli plan terpisah per proyek.

Untuk tim kecil dengan beberapa proyek berjalan paralel, plan ADVANCE ($90/bulan, 50 thread) biasanya cukup — billing CaptchaAI berbasis thread konkuren dengan solve tak terbatas per thread, jadi menambah proyek baru di belakang microservice yang sama tidak otomatis menambah biaya per CAPTCHA. Kalau Anda deploy microservice ini di region yang sama dengan service pemanggilnya, misalnya AWS ap-southeast-1 (Singapura) atau GCP asia-southeast2 (Jakarta) untuk tim yang server produksinya di Asia Tenggara, round-trip internal tetap rendah; waktu total tetap didominasi waktu solve CaptchaAI itu sendiri.

Prasyarat: API Key, Python, dan FastAPI + httpx

  • API Key CaptchaAI — ambil di captchaai.com
  • Python 3.9+
  • FastAPI + httpx — untuk request HTTP async ke CaptchaAI tanpa blocking
pip install fastapi uvicorn httpx

Struktur Proyek Microservice

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

Microservice ini dipecah jadi dua modul dengan tanggung jawab terpisah: solver.py menyimpan semua komunikasi dengan CaptchaAI — submit, polling, dan error handling — sementara main.py hanya mendefinisikan endpoint REST dan validasi request lewat Pydantic. Pemisahan ini memudahkan Anda menambah endpoint baru nanti, misalnya untuk GeeTest v3, tanpa menyentuh logika HTTP yang sudah teruji.

Modul solver.py: Logika Solve 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)

Perhatikan initial_wait berbeda di tiap fungsi solve — ini sengaja dibedakan per tipe CAPTCHA supaya microservice tidak langsung nge-poll res.php begitu task disubmit dan boros request percuma. Kalau volume trafik Anda cukup besar untuk mengukur waktu solve rata-rata sendiri, sesuaikan angka initial_wait dan max_attempts dengan data itu alih-alih memakai nilai default di atas.

main.py: Endpoint FastAPI untuk Tiap Jenis CAPTCHA

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

Tiap route memvalidasi body request lewat model Pydantic sebelum menyentuh solver.py, jadi payload yang salah bentuk gagal di lapisan validasi, bukan di tengah panggilan ke CaptchaAI. Error dari solver.py (ValueError untuk kegagalan submit/solve, TimeoutError untuk polling yang habis waktu) diteruskan sebagai HTTP 502 dengan pesan spesifik di field detail, supaya service pemanggil tahu persis apa yang gagal. Endpoint /health sendiri berguna untuk readiness check load balancer atau orchestrator seperti Kubernetes.

Menjalankan Service dengan Uvicorn

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

--host 0.0.0.0 wajib kalau microservice ini dijalankan di dalam container atau diakses dari service lain di jaringan yang sama — 127.0.0.1 hanya menerima koneksi dari mesin itu sendiri. Untuk trafik produksi yang lebih tinggi, tambahkan --workers N supaya Uvicorn menjalankan beberapa proses worker, atau taruh reverse proxy (nginx, Traefik) di depannya untuk load balancing dan TLS termination.

Contoh Request dan Respons

Solve reCAPTCHA v2 lewat endpoint /solve/recaptcha-v2:

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

Solve Cloudflare Turnstile lewat endpoint /solve/turnstile:

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

Respons:

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

Field token inilah yang Anda suntikkan ke g-recaptcha-response (reCAPTCHA) atau cf-turnstile-response (Turnstile) di form target, sementara user_agent opsional dipakai kalau browser automation Anda perlu menyamakan header dengan sesi yang dipakai CaptchaAI saat solve.

Checklist Hardening Sebelum Deploy ke Production

  1. Baca API key dari environment variable dan fail cepat saat startup jika secret tidak ada — jangan hardcode YOUR_API_KEY seperti di contoh solver.py.
  2. Pisahkan validasi request dari eksekusi solver agar payload yang tidak valid tidak pernah mencapai jalur outbound call ke CaptchaAI.
  3. Tampilkan respons error terstruktur yang membedakan kegagalan validasi, kegagalan solver, dan penolakan upstream.
  4. Jangan log API key atau payload penuh; cukup log task ID dan status supaya kredensial tidak ikut bocor ke log aggregator.
  5. Pantau jumlah request yang sedang diproses dan bandingkan dengan thread yang tersedia di plan CaptchaAI Anda — upgrade plan dulu sebelum request mulai antre.

Pemecahan Masalah Umum

Kalau microservice sudah jalan tapi hasilnya tidak sesuai ekspektasi, cek dulu empat penyebab paling umum:

  • Response 502 — CaptchaAI mengembalikan error; periksa field detail untuk pesan spesifik.
  • Timeout solve — CAPTCHA makan waktu lebih lama dari perkiraan; naikkan max_attempts atau cek status CaptchaAI.
  • Connection refused — service belum berjalan; pastikan uvicorn aktif di port yang benar.
  • Response lambat — ada blocking I/O; pastikan httpx.AsyncClient dipakai, bukan requests.

Pertanyaan Umum Seputar Setup Ini

Apakah endpoint ini juga bisa dipakai untuk hCaptcha?

Belum. hCaptcha tidak didukung CaptchaAI saat ini, jadi menambah endpoint /solve/hcaptcha di main.py di atas tidak akan berfungsi — kode solver.py hanya menangani tipe yang benar-benar didukung: reCAPTCHA v2/v3, Turnstile, dan CAPTCHA gambar. Kalau proyek Anda perlu hCaptcha, itu perlu layanan terpisah.

Apakah microservice ini bisa dipasang di server Jakarta atau Singapura untuk latensi lebih rendah?

Bisa. Kode di atas tidak terikat region tertentu, jadi Anda bebas deploy di AWS ap-southeast-1, ap-southeast-3, atau GCP asia-southeast2 seperti service lain di stack Anda. Menempatkan microservice ini dekat dengan service pemanggilnya memangkas latensi jaringan internal, tapi waktu total tetap didominasi waktu solve CaptchaAI, bukan jarak ke region.

Berapa thread CaptchaAI idealnya dipakai untuk trafik microservice tim kecil?

Tergantung berapa banyak solve yang berjalan bersamaan, bukan berapa banyak proyek yang memanggil microservice ini — billing CaptchaAI berbasis thread konkuren dengan solve tak terbatas per thread. Untuk tim kecil yang baru mulai, BASIC ($15/bulan, 5 thread) atau STANDARD ($30/bulan, 15 thread) biasanya cukup; naikkan ke ADVANCE ($90/bulan, 50 thread) begitu beberapa proyek klien berjalan paralel lewat microservice yang sama.

Butuh autentikasi dan Docker sebelum masuk production?

  • Autentikasi endpoint — tambahkan dependency FastAPI yang memvalidasi header X-API-Key (atau skema OAuth2) sebelum request masuk ke solver.py, dipasang lewat parameter Depends() di tiap route, bukan diulang manual di tiap fungsi.
  • Docker — disarankan, terutama kalau microservice ini dipanggil banyak service lain. Image python:3.11-slim dengan dependensi di atas sudah cukup untuk Dockerfile sederhana yang expose port 8000, dan memudahkan Anda menjalankan beberapa instance di belakang load balancer kalau trafik naik.

Siap Bangun Microservice CAPTCHA Anda Sendiri?

Ambil API key CaptchaAI Anda di captchaai.com, lalu jalankan solver.py dan main.py di atas sebagai titik solve CAPTCHA tunggal untuk semua service Anda — satu API key, satu plan thread, tanpa logika submit-poll yang berulang di tiap proyek.

Panduan Terkait

Komentar dinonaktifkan untuk artikel ini.