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

验证码识别任务大部分时间都花在等第三方 API 返回结果上，真正占用 CPU 的时间很短——这正是 AWS Lambda 的强项。把 CaptchaAI 求解逻辑放进 Lambda 函数，你不用为空闲时段付费，也不用维护常驻进程：请求来了才启动，处理完立刻释放，还能直接接入 API Gateway、SQS 或 Step Functions。

编写 Lambda 处理函数

下面的处理函数接收一次求解请求，从环境变量读取 API Key，调用 in.php 提交任务，再轮询 res.php 直到拿到 token 或超时，逻辑和普通脚本一致，只是入口换成了 event/context：

# lambda_function.py
import json
import os
import time
import urllib.request
import urllib.parse


def lambda_handler(event, context):
    """AWS Lambda handler for CaptchaAI solving."""
    api_key = os.environ["CAPTCHAAI_KEY"]

    # Parse input
    body = json.loads(event.get("body", "{}")) if isinstance(event.get("body"), str) else event

    method = body.get("method", "userrecaptcha")
    params = body.get("params", {})

    try:
        token = solve_captcha(api_key, method, params)
        return {
            "statusCode": 200,
            "body": json.dumps({"token": token}),
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)}),
        }


def solve_captcha(api_key, method, params, timeout=90):
    """Solve CAPTCHA using CaptchaAI API."""
    # Submit task
    submit_data = urllib.parse.urlencode({
        "key": api_key,
        "method": method,
        "json": 1,
        **params,
    }).encode()

    req = urllib.request.Request(
        "https://ocr.captchaai.com/in.php",
        data=submit_data,
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read())

    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")

    task_id = result["request"]

    # Poll for result
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        poll_url = (
            f"https://ocr.captchaai.com/res.php"
            f"?key={api_key}&action=get&id={task_id}&json=1"
        )
        with urllib.request.urlopen(poll_url, timeout=15) as resp:
            data = json.loads(resp.read())

        if data["request"] != "CAPCHA_NOT_READY":
            if data.get("status") == 1:
                return data["request"]
            raise RuntimeError(f"Solve error: {data['request']}")

    raise TimeoutError("Solve timeout")

用 Secrets Manager 管理 API Key

API Key 写死在环境变量里能跑起来，但生产环境更推荐用 Secrets Manager 集中管理，配合 IAM 策略控制读取权限：

import json
import boto3


def get_api_key():
    """Retrieve CaptchaAI key from AWS Secrets Manager."""
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="captchaai/api-key")
    secret = json.loads(response["SecretString"])
    return secret["api_key"]

先把密钥存进 Secrets Manager：

aws secretsmanager create-secret \
  --name captchaai/api-key \
  --secret-string '{"api_key":"YOUR_API_KEY"}'

用 SAM 模板做基础设施即代码

接下来用 AWS SAM 把部署声明成代码——Lambda 函数、API Gateway 路由、IAM 权限一次定义清楚，回滚也更可控：

# template.yaml
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Timeout: 120
    MemorySize: 256
    Runtime: python3.11

Resources:
  CaptchaSolverFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda_function.lambda_handler
      Environment:
        Variables:
          CAPTCHAAI_KEY: !Sub "{{resolve:secretsmanager:captchaai/api-key:SecretString:api_key}}"
      Events:
        SolveApi:
          Type: Api
          Properties:
            Path: /solve
            Method: post
      Policies:

        - AWSSecretsManagerGetSecretValuePolicy:
            SecretArn: !Sub "arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:captchaai/api-key-*"

Outputs:
  SolveApiUrl:
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/solve"

构建与部署

两条命令就能上线，再用 curl 验证接口是否正常返回 token：

# Build and deploy
sam build
sam deploy --guided

# Test
curl -X POST https://YOUR_API_ID.execute-api.us-east-1.amazonaws.com/Prod/solve \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

结合 SQS 实现批量验证码处理

验证码任务批量产生时——比如一次抓取要处理几百个页面——用 SQS 触发 Lambda 比同步调用更稳：队列做缓冲，Lambda 按并发消费，单个任务失败不会拖垮整批：

import json
import os
import time
import urllib.request
import urllib.parse


def sqs_handler(event, context):
    """Process CAPTCHA tasks from SQS queue."""
    api_key = os.environ["CAPTCHAAI_KEY"]
    results = []

    for record in event["Records"]:
        task = json.loads(record["body"])
        try:
            token = solve_captcha(
                api_key,
                task["method"],
                task["params"],
            )
            results.append({
                "task_id": task.get("id"),
                "status": "success",
                "token": token[:50],
            })
        except Exception as e:
            results.append({
                "task_id": task.get("id"),
                "status": "error",
                "error": str(e),
            })

    return {"results": results}

Lambda 参数怎么选：超时、内存、并发

这几个参数基本不用反复调整：

在 AWS 中国区部署要注意什么

Lambda 若部署在 AWS 中国区（cn-north-1、cn-northwest-1），出公网访问 ocr.captchaai.com 这类境外地址的延迟通常不如 ap-southeast-1（新加坡）稳定，批量高并发时更容易出现偶发超时。用户主要在国内的团队，选 ap-southeast-1 往往更省心，上线前先跑一轮真实延迟测试，再对齐轮询超时时间。

常见故障排查

常见问题

Lambda 处理验证码划算吗？

划算。一次 256 MB、60 秒的调用大约 $0.0001，在 CaptchaAI 的 API 费用之上几乎可以忽略。CaptchaAI 按并发线程数收费，不是按次数——比如 BASIC 套餐 $15/月含 5 个线程，线程内解题次数不限——真正影响成本的是套餐线程数，不是 Lambda 本身。

Lambda 超时应该设多久？

大多数验证码在 10-60 秒内出结果，把超时设为 120 秒足够覆盖轮询等待。任务里包含 reCAPTCHA Enterprise 这类更复杂的类型时，建议留到 180 秒。

能不能不用每次都部署，先在本地测试处理函数？

可以。sam local invoke 在本地用 Docker 模拟 Lambda 运行时，传入测试用的 event.json 就能跑通提交、轮询、返回 token 的完整逻辑，确认无误再 sam deploy。

如何把轮询产生的 Lambda 计费时长降到最低？

把首次轮询间隔从 5 秒起步改成逐步递增（3 秒、5 秒、8 秒……），多数简单验证码在前几次轮询就能拿到结果，明显减少空转；耗时长的类型可以拆到 SQS 异步流程里，用 Step Functions 编排轮询节奏。

SQS 批量处理里失败的任务需要重试吗？

需要。给队列配置死信队列（DLQ），设置合理的 maxReceiveCount（比如 3 次），超过重试次数的任务会自动转入 DLQ，方便你判断是验证码识别失败，还是网络或 API Key 配置问题。

相关指南

• Google Cloud Functions 集成 CaptchaAI
• Azure Functions 集成 CaptchaAI

从服务器运维中解脱出来——立即获取你的 CaptchaAI API Key。