Bangun Microservice Solve CAPTCHA dengan FastAPI dan CaptchaAI

Ketika beberapa layanan atau anggota tim memerlukan solve CAPTCHA, memusatkannya ke dalam microservice menghindari duplikasi logika di seluruh proyek. Dukungan async FastAPI membuatnya sangat cocok — solve CAPTCHA melibatkan menunggu respons API eksternal, yang ditangani async secara efisien tanpa blocking thread.

Panduan ini membangun microservice FastAPI yang menerima request solve CAPTCHA via REST dan mengembalikan token yang di-solve melalui CaptchaAI.

Yang Anda Butuhkan

Instal dependensi:

pip install fastapi uvicorn httpx

Struktur Proyek

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

Modul Solver 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)

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

Jalankan Service

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

Contoh Penggunaan

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

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

Respon:

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

Catatan Hardening Deployment

• Baca API key dari environment variable dan fail cepat saat startup jika secret tidak ada.
• Pisahkan validasi request dari eksekusi solver agar payload yang tidak valid tidak pernah mencapai jalur outbound call.
• Tampilkan respons error terstruktur yang membedakan kegagalan validasi, kegagalan solver, dan penolakan upstream.

Pemecahan Masalah

Pertanyaan Umum

Mengapa menggunakan FastAPI untuk service solve CAPTCHA?

FastAPI menangani async I/O secara native, ideal untuk solve CAPTCHA di mana sebagian besar waktu dihabiskan menunggu respons CaptchaAI. Beberapa request dapat diproses secara concurrent tanpa threading.

Bisakah saya menambahkan autentikasi ke endpoint?

Ya. Tambahkan dependency injection FastAPI dengan validasi header API key atau OAuth2 untuk membatasi akses.

Berapa banyak concurrent request yang dapat ditangani?

Dibatasi oleh batas concurrent task paket CaptchaAI Anda. FastAPI sendiri dapat menangani ribuan koneksi secara bersamaan.

Haruskah saya menjalankannya dalam Docker?

Ya. Tambahkan Dockerfile dengan FROM python:3.11-slim, instal dependensi, dan expose port 8000.

Bisakah saya menambahkan rate limiting?

Ya. Gunakan slowapi atau reverse proxy (nginx, Traefik) untuk membatasi request per client.

Mulai Sekarang

Dapatkan API key Anda di captchaai.com. Sentralisasi solve CAPTCHA dengan microservice FastAPI.

Panduan Terkait

• Cara Solve reCAPTCHA v2 Menggunakan API
• Cara Solve Cloudflare Turnstile Menggunakan API
• Async CaptchaAI dengan aiohttp