用 S3 Object Lambda 把 Claude 塞进读取路径

以前做代码审查总得等 CI 跑完，开 PR 了才能看到评论。等反馈回来，上下文早断了，改起来特别别扭。前两天把 S3 Object Lambda 搭起来，直接在 GetObject 里挂个 Lambda，文件一读就把 Claude 的建议夹在里面返回——存储层直接变成了实时审查服务，省去了整套 CI 流水线。

核心思路：拦截读请求，注入审查结果

S3 Object Lambda 本质上是给 Access Point 挂个 Lambda 函数。客户端走这个 AP 做 GetObject，请求先被 Lambda 拦截，Lambda 去读原文件、丢给 Claude、拿到建议、拼成新对象流返回。调用方完全感知不到中间发生了什么，就像平常读文件一样。

关键点有三个：

• Access Point 不继承桶策略，必须单独给 Lambda 调用权限（s3-object-lambda:InvokeFunction）和读对象权限（s3:GetObject）
• Lambda 里要处理 x-amz-object-lambda 头，按协议写回 WriteGetObjectResponse
• 响应体里把审查意见塞进自定义元数据或直接拼进文件内容，看客户端怎么消费

一、建 Object Lambda Access Point

// create-access-point.ts
import { S3Client, CreateAccessPointCommand } from "@aws-sdk/client-s3";

const s3 = new S3Client({ region: "us-east-1" });

async function createAP() {
  const cmd = new CreateAccessPointCommand({
    AccountId: "123456789012",
    Name: "code-review-ap",
    ObjectLambdaConfiguration: {
      SupportingAccessPoint: "arn:aws:s3:us-east-1:123456789012:accesspoint/source-bucket-ap",
      CloudWatchMetricsEnabled: true,
      // 这里要填有权限的 Lambda ARN，IAM 角色稍后配
    },
  });
  const res = await s3.send(cmd);
  console.log("AP ARN:", res.AccessPointArn);
}

createAP().catch(console.error);

注意 SupportingAccessPoint 要先建好，指向源桶。这个 AP 只负责转发读请求，不做任何转换。

二、Lambda 处理函数：读原文 → 扔给 Claude → 写回

# lambda_function.py
import json, os, boto3
from botocore.config import Config

s3 = boto3.client("s3", config=Config(signature_version="s3v4"))
claude = boto3.client("bedrock-runtime", region_name="us-east-1")

PROMPT = """你是资深代码审查员。请对下面的代码给出逐行改进建议，
只输出 JSON：{"reviews": [{"line": 12, "severity": "warning", "message": "..."}]}"""

def handler(event, context):
    # 1. 解析 Object Lambda 传来的上下文
    ctx = event["getObjectContext"]
    route = ctx["outputRoute"]
    token = ctx["outputToken"]
    src_bucket = event["configuration"]["supportingAccessPointArn"].split(":")[-1]
    src_key = event["userRequest"]["url"].split("/")[-1]

    # 2. 读原文件
    obj = s3.get_object(Bucket=src_bucket, Key=src_key)
    code = obj["Body"].read().decode("utf-8")

    # 3. 调用 Claude（这里用 Bedrock，换 OpenAI 也行）
    payload = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 2000,
        "messages": [{"role": "user", "content": f"{PROMPT}\n\n

python\n{code}\n

"}],
    }
    resp = claude.invoke_model(modelId="anthropic.claude-3-sonnet-20240229-v1:0", body=json.dumps(payload))
    review_json = json.loads(resp["body"].read())["content"][0]["text"]

    # 4. 把审查结果塞进自定义元数据返回
    s3.write_get_object_response(
        Route=route,
        RequestToken=token,
        Body=code.encode("utf-8"),  # 原文照传
        Metadata={"x-amz-meta-claude-review": review_json},
    )
    return {"status_code": 200}

关键细节：

• 必须用 write_get_object_response 把流写回 Object Lambda 提供的路由，不能直接 return
• Metadata 里带上审查 JSON，客户端拿 HeadObject 或 GetObject 的元数据就能解析
• Lambda 超时设 30 秒以上，Claude 思考慢的话容易截断

三、IAM 最小权限策略

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowInvoke",
      "Effect": "Allow",
      "Action": "s3-object-lambda:InvokeFunction",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:code-review-lambda"
    },
    {
      "Sid": "ReadSource",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::source-bucket/*"
    },
    {
      "Sid": "BedrockAccess",
      "Effect": "Allow",
      "Action": "bedrock:InvokeModel",
      "Resource": "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0"
    }
  ]
}

把这策略挂给 Lambda 执行角色，再在 AP 的 ObjectLambdaConfiguration 里引用这个角色 ARN。

四、客户端怎么用

# 走 Object Lambda AP 读文件
aws s3api get-object \
  --bucket arn:aws:s3:us-east-1:123456789012:accesspoint/code-review-ap \
  --key src/main.py \
  /dev/stdout

# 拿审查元数据
aws s3api head-object \
  --bucket arn:aws:s3:us-east-1:123456789012:accesspoint/code-review-ap \
  --key src/main.py \
  --query 'Metadata.x-amz-meta-claude-review'

IDE 里装个插件读这个元数据，光标悬停就能弹出建议，体验接近 Copilot 但完全自建、数据不出桶。

踩过的坑

1. 并发限制：Object Lambda 单 AP 默认 1000 并发，团队大了得提配额或分多个 AP
2. 大文件截断：超过 5 MB 的文件建议只审查变更区间，否则 Token