使用 FastAPI 和 CaptchaAI 构建验证码解决微服务 (实用指南)

当多个服务或团队成员需要验证码解决时，将其集中到微服务中可以避免跨项目重复逻辑。 FastAPI 的异步支持使其非常适合 - 验证码解决涉及等待外部 API 响应，异步可以有效处理该响应，而不会阻塞线程。

本指南构建了一个 FastAPI 微服务，该服务通过 REST 接受 CAPTCHA 解决请求，并通过 CaptchaAI 返回已解决的令牌。

你需要什么

要求	细节
CaptchaAI API 密钥	验证码网站
Python 3.9+
FastAPI + httpx	对于异步 HTTP 处理

安装依赖项：

pip install fastapi uvicorn httpx

项目结构

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

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)

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

运行服务

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

使用示例

解决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..."
}

部署强化说明

从环境中读取 API 密钥，如果密钥丢失，则会在启动过程中快速失败。
将请求验证与求解器执行分开，以便无效的有效负载永远不会到达出站调用路径。
返回区分验证失败、求解器失败和上游拒绝的结构化错误响应。

故障排除

问题	原因	处理方式
502响应	CaptchaAI 返回错误	检查 `detail` 字段是否有具体错误
求解超时	验证码花费的时间太长	增加`max_attempts`或检查CaptchaAI状态
连接被拒绝	服务未运行	验证 `uvicorn` 是否在预期端口上运行
反应慢	阻止 I/O	确保使用 `httpx.AsyncClient`，而不是 `requests`

常问问题

为什么使用 FastAPI 进行验证码解决服务？

FastAPI 原生处理异步 I/O，这非常适合验证码解决方案，因为大多数时间都花在等待 CaptchaAI 的响应上。可以同时处理多个请求，无需线程。

我可以向端点添加身份验证吗？

是的。添加带有 API 密钥标头验证或 OAuth2 的 FastAPI 依赖注入来限制访问。

它可以处理多少个并发请求？

受 CaptchaAI 计划的并发任务限制的限制。 FastAPI 本身可以处理数千个并发连接。

我应该将其docker化吗？

是的。添加 Dockerfile 和 FROM python:3.11-slim，安装依赖项并公开端口 8000。

我可以添加速率限制吗？

是的。使用 slowapi 或反向代理（nginx、Traefik）来限制每个客户端的请求。

构建您的验证码解决微服务

获取您的 API 密钥：验证码网站。使用 FastAPI 微服务集中验证码解决。

使用 FastAPI 和 CaptchaAI 构建验证码解决微服务

你需要什么

项目结构

CaptchaAI 求解器模块

FastAPI应用

运行服务

使用示例

解决reCAPTCHA v2

解决Cloudflare Turnstile

部署强化说明

故障排除

常问问题

为什么使用 FastAPI 进行验证码解决服务？

我可以向端点添加身份验证吗？

它可以处理多少个并发请求？

我应该将其docker化吗？

我可以添加速率限制吗？

构建您的验证码解决微服务

相关指南

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控

你需要什么

项目结构

CaptchaAI 求解器模块

FastAPI应用

运行服务

使用示例

解决reCAPTCHA v2

解决Cloudflare Turnstile

部署强化说明

故障排除

常问问题

为什么使用 FastAPI 进行验证码解决服务？

我可以向端点添加身份验证吗？

它可以处理多少个并发请求？

我应该将其docker化吗？

我可以添加速率限制吗？

构建您的验证码解决微服务

相关指南

相关文章

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控