Микросервис для решения CAPTCHA на FastAPI и CaptchaAI

Если решение CAPTCHA нужно нескольким скриптам, парсерам или командам сразу, самый быстрый путь — не тащить обработку in.php/res.php в каждый проект, а вынести её в один REST-сервис. Остальной код тогда обращается к внутреннему /solve/... эндпоинту и не знает вообще ничего про CaptchaAI, ротацию задач или поллинг.

FastAPI подходит для этой роли лучше, чем синхронный Flask: решение CAPTCHA — это в первую очередь ожидание. Сервис отправляет задачу и ждёт от 5 до 30+ секунд, пока CaptchaAI её решит. Async event loop FastAPI обрабатывает это ожидание без блокировки потока — пока один запрос ждёт res.php, воркер уже принимает следующий.

Зачем выносить решение CAPTCHA в отдельный сервис

Типичный повод — не один скрипт, а несколько: продакшен-парсер, тестовый стенд и локальная разработка используют один API-ключ, но каждый пишет свою логику поллинга и обработки таймаутов. Микросервис снимает это дублирование: ключ, ретраи и парсинг ответа res.php живут в одном месте, а команда меняет только URL, на который стучится клиентский код.

Это же удобно для тарификации. CaptchaAI считает потоки, а не отдельные решения — например, план STANDARD ($30/мес, 15 потоков) покрывает нагрузку небольшой команды или агентства без пересчёта по каждому решённому токену. Для фрилансеров и агентств, которые выставляют счета в долларах, а расходы несут в местной валюте, фиксированная помесячная стоимость такого рода предсказуемее, чем оплата за штуку.

Что понадобится

Именно httpx, а не requests — у него есть асинхронный клиент, без которого весь смысл FastAPI в этой задаче теряется: синхронный HTTP-клиент внутри async-функции всё равно заблокирует поток на время ожидания res.php.

Установите зависимости:

pip install fastapi uvicorn httpx

Структура проекта

Логику решения CAPTCHA стоит держать отдельно от роутинга FastAPI — так проще тестировать solver.py без поднятия HTTP-сервера и переиспользовать его в CLI-скриптах при необходимости.

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

Модуль решения CaptchaAI

submit_task отправляет запрос в in.php и возвращает ID задачи, poll_result опрашивает res.php, пока не придёт токен или не истечёт max_attempts. Обратите внимание на initial_wait: он подобран под тип CAPTCHA — 20 секунд для reCAPTCHA, 10 для Turnstile, 5 для изображений, — чтобы не тратить запросы на заведомо ранний опрос.

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

Приложение FastAPI

Роутер ниже — тонкая обвязка над solver.py: Pydantic-модели проверяют входные данные ещё до вызова CaptchaAI, а любая ошибка решателя (ValueError или TimeoutError) превращается в понятный 502 с текстом причины в detail, вместо голого traceback.

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

Запуск сервиса

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

/health стоит завести в проверку живости за балансировщиком или в Docker Compose/Kubernetes — без него оркестратор не отличит зависший процесс от рабочего.

Как вызывать эндпоинты

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

Cloudflare Turnstile

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

Ответ:

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

user_agent возвращается не для всех типов CAPTCHA — для reCAPTCHA его стоит передать в тот же браузерный контекст, где решается форма, иначе токен может не приняться сайтом-источником.

Диагностика проблем

Безопасность в продакшене

• Читайте API-ключ из переменных окружения и завершайте запуск с ошибкой, если секрет не задан — не оставляйте ключ в коде или в дефолтном значении.
• Разделяйте валидацию запроса и вызов решателя: некорректные данные не должны доходить до исходящего запроса к CaptchaAI.
• Возвращайте структурированные ошибки, которые различают ошибку валидации, ошибку решателя и отказ на стороне CaptchaAI — так проще диагностировать сбой по логам, не вскрывая тело каждого запроса.
• Если сервис используется для парсинга или сбора данных, логируйте только то, что действительно нужно для отладки: не пишите в лог sitekey, pageurl реальных клиентов и токены целиком — собирайте только данные, которые вы вправе обрабатывать по 152-ФЗ или GDPR, если через сервис проходят чужие данные.

Часто задаваемые вопросы

Можно ли добавить GeeTest v3 или BLS CAPTCHA, а не только типы из примера?

Да — добавьте в solver.py ещё одну функцию solve_* по образцу solve_turnstile, укажите нужный method для in.php и подходящий initial_wait, затем заведите под неё отдельный роутер в main.py. Также можно подключить CaptchaFox, Friendly Captcha и Lemin — но CaptchaAI поддерживает их только в бета-режиме, это стоит отразить в документации сервиса. hCaptcha, FunCaptcha и GeeTest v4 API CaptchaAI пока не решает, добавлять их в сервис не нужно.

Что делать при ошибке ERROR_ZERO_BALANCE или ERROR_WRONG_USER_KEY?

ERROR_ZERO_BALANCE значит, что на аккаунте закончились активные потоки или средства — проверьте баланс и план в панели управления CaptchaAI. ERROR_WRONG_USER_KEY — ключ введён неверно или не тот, что активен; сверьте значение API_KEY с ключом из аккаунта.

Можно ли обслуживать несколько проектов одним инстансом сервиса?

Да, если добавить проверку клиента на уровне зависимостей FastAPI (заголовок с внутренним ключом или OAuth2) и логировать, какой клиент какие потоки расходует — иначе один шумный проект выест лимит потоков у остальных.

Нужно ли докеризовать сервис?

Да, это упрощает деплой. Добавьте Dockerfile с FROM python:3.11-slim, установите зависимости из requirements.txt и откройте порт 8000.

Можно ли добавить ограничение частоты запросов?

Да. Используйте slowapi внутри приложения или ограничение на уровне обратного прокси (nginx, Traefik) — это защитит сервис от одного клиента, который отправляет задачи быстрее, чем позволяет тарифный план.

Разверните свой микросервис для решения CAPTCHA

Получите API-ключ на captchaai.com, соберите solver.py и main.py по примерам выше и подключите к нему первый проект — остальные могут переключиться на общий эндпоинт позже, без изменения своей интеграционной логики.

Похожие материалы

• Как решить reCAPTCHA v2 через API
• Как решить Cloudflare Turnstile через API
• Асинхронный CaptchaAI с aiohttp
• Параллельное решение CAPTCHA