Xây dựng vi dịch vụ giải CAPTCHA với FastAPI và CaptchaAI

Khi một agency outsourcing chạy song song nhiều dự án scraping hoặc QA cho các khách hàng khác nhau, copy cùng một đoạn code gọi CaptchaAI vào từng repo là cách chắc chắn tạo ra bug lệch phiên bản — sửa một lỗi ở dự án A không có nghĩa dự án B cũng được sửa. Gom toàn bộ logic giải CAPTCHA vào một vi dịch vụ (microservice) dùng chung qua REST giải quyết đúng vấn đề này.

Luồng xử lý của vi dịch vụ đi qua đúng 4 bước, lặp lại cho mọi loại CAPTCHA:

1. Nhận request từ client (sitekey, pageurl, loại CAPTCHA)
2. Submit task tới CaptchaAI qua in.php, nhận về task ID
3. Polling res.php định kỳ cho đến khi có kết quả
4. Trả token đã giải về cho client qua response JSON

FastAPI phù hợp cho việc này vì phần lớn thời gian xử lý là chờ CaptchaAI phản hồi chứ không phải tính toán CPU — async xử lý hàng trăm request đang chờ mà không cần mở thêm thread hệ điều hành cho từng request.

Chuẩn bị trước khi dựng vi dịch vụ

Cần ba thứ trước khi viết dòng code đầu tiên:

Cài các gói phụ thuộc:

pip install fastapi uvicorn httpx

Cấu trúc thư mục

Ba file, tách rõ trách nhiệm: routing HTTP nằm ở main.py, toàn bộ logic gọi CaptchaAI nằm ở solver.py.

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

Viết module gọi CaptchaAI (solver.py)

Module này chỉ làm một việc — nói chuyện với CaptchaAI. Ba phần chính:

• submit_task() — gửi params tới in.php, trả về task ID
• poll_result() — gọi res.php lặp lại tới khi có token hoặc hết max_attempts
• Các hàm solve_*() — gói tham số đúng cho từng loại CAPTCHA rồi gọi hai hàm trên

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

Khai báo endpoint trong main.py

Mỗi loại CAPTCHA có một Request model riêng bằng Pydantic, để FastAPI tự validate payload trước khi chạm tới solver:

1. /solve/recaptcha-v2 và /solve/recaptcha-v3 — nhận sitekey, pageurl, tuỳ chọn enterprise
2. /solve/turnstile — nhận sitekey, pageurl
3. /solve/image — nhận image_base64

Lỗi từ solver được bọc thành HTTPException mã 502, để client phân biệt lỗi gọi API sai với lỗi CaptchaAI trả về.

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

Chạy thử local

Khởi động vi dịch vụ bằng uvicorn:

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

Gọi thử bằng cURL

Vi dịch vụ đã chạy ở localhost:8000, gọi trực tiếp bằng cURL để xác nhận mọi thứ hoạt động đúng trước khi tích hợp vào code client.

Giải 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"}'

Giải Cloudflare Turnstile

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

Phản hồi mẫu:

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

Lưu ý khi đưa vào production

Endpoint /health đã có sẵn trong main.py — trỏ health check của Docker hoặc load balancer vào đó thay vì tự thêm route mới.

• Đọc API key từ biến môi trường, fail-fast ngay lúc khởi động nếu thiếu.
• Tách validate request khỏi phần gọi solver, để payload sai không chạm tới CaptchaAI.
• Trả lỗi có cấu trúc rõ ràng, phân biệt lỗi validate, lỗi CaptchaAI và lỗi upstream.

Các lỗi thường gặp

Câu hỏi thường gặp

Vi dịch vụ này tốn bao nhiêu thread trên gói CaptchaAI?

Số thread quyết định số CAPTCHA giải đồng thời, không phải số lần giải mỗi tháng — mỗi gói cho giải không giới hạn trong số thread đó:

• BASIC ($15/tháng, 5 thread) — đủ cho một service nội bộ nhỏ
• ADVANCE ($90/tháng, 50 thread) — khi nhiều dự án cùng gọi song song

CaptchaAI có giải được hCaptcha qua vi dịch vụ này không?

Chưa. CaptchaAI hỗ trợ reCAPTCHA v2/v3, Cloudflare Turnstile, GeeTest v3, CAPTCHA ảnh/OCR và BLS — hCaptcha và FunCaptcha (Arkose Labs) chưa được hỗ trợ, đừng thêm endpoint cho hai loại này vào solver.py.

Có thể thêm xác thực (authentication) cho từng endpoint không?

Có. Dùng dependency injection của FastAPI để kiểm tra header API key riêng của vi dịch vụ, hoặc gắn OAuth2, tránh để endpoint /solve/* mở công khai cho bất kỳ ai gọi tới.

Nên polling hay dùng webhook để nhận token?

Ví dụ trong bài dùng polling res.php vì đơn giản và đủ nhanh cho service nội bộ. Nếu khối lượng lớn và muốn giảm số lần gọi lặp lại, thay poll_result() bằng callback URL để CaptchaAI tự gửi kết quả về khi giải xong.

Có cần Dockerize vi dịch vụ này không?

Nên có, nhất là khi chạy nhiều vi dịch vụ tương tự cho nhiều dự án. Dùng Dockerfile với FROM python:3.11-slim, cài requirements.txt, expose cổng 8000 — giống cách chạy uvicorn ở trên.

Bắt đầu dựng vi dịch vụ của bạn

Lấy API key tại captchaai.com, copy solver.py ở trên và có endpoint giải CAPTCHA đầu tiên chạy trong vài phút — không cần viết lại logic polling cho từng dự án mới.

Hướng dẫn liên quan

• Giải reCAPTCHA v2 qua API từng bước
• Giải Cloudflare Turnstile bằng API
• Dùng aiohttp để gọi CaptchaAI bất đồng bộ thay vì FastAPI
• Giải CAPTCHA song song ở quy mô lớn