Amazon Bedrock AgentCore 开发实战(一):本地构建 AI Agent

在 AI Agent 开发过程中,快速迭代和即时反馈至关重要。Amazon Bedrock AgentCore 提供了完整的本地开发支持,让开发者能够在本地环境中构建、测试和调试 Agent,然后无缝部署到云端。

本文将带您从零开始,在本地搭建 AgentCore 开发环境,并通过详细的代码解析,帮助您深入理解每一个技术细节。

Amazon Bedrock AgentCore 简介

Amazon Bedrock AgentCore 是一套专为 AI Agent 设计的企业级基础设施服务。它解决了 Agent 从原型到生产过程中的核心挑战:

  • AgentCore Runtime:无服务器运行时环境,支持最长 8 小时的任务执行和 100MB 的请求负载
  • AgentCore Memory:提供短期和长期记忆管理,支持跨会话的上下文保持
  • AgentCore Gateway:统一的工具网关,支持 MCP 协议的工具发现和调用
  • AgentCore Identity:安全的身份认证和授权管理

本地开发的优势在于:开发者可以在熟悉的环境中快速验证想法,无需等待云端部署,同时保持与生产环境一致的代码结构。

阅读更多

零代码改动!用 Docker 将 Flask 应用部署到 AWS Lambda

你有一个现成的 Flask API,想部署到 AWS Lambda 享受 Serverless 的好处,但又不想改代码?AWS Lambda Web Adapter 可以帮你实现。

本文将手把手教你如何使用 Docker + Gunicorn + Lambda Web Adapter,将 Flask 应用部署到 Lambda,并通过 API Gateway 对外提供服务。

为什么选择这个方案?

传统方式部署 Flask 到 Lambda 需要使用 Mangum、aws-wsgi 等第三方库,需要修改代码添加 handler。而 Lambda Web Adapter 是 AWS 官方方案,有以下优势:

  • 零代码改动:Flask 代码完全不用改
  • 生产级配置:可以使用 Gunicorn 作为 WSGI 服务器
  • 本地开发友好:同一个 Docker 镜像本地和 Lambda 都能跑
  • 框架无关:Flask、Django、FastAPI 都支持

架构概览

1
2
3
4
5
客户端 → API Gateway → Lambda (Docker 容器)

Lambda Web Adapter

Gunicorn + Flask

Lambda Web Adapter 作为 Lambda Extension 运行,负责将 API Gateway 事件转换为标准 HTTP 请求,Flask 应用完全感知不到自己运行在 Lambda 上。

准备工作

确保你已安装:

  • Docker
  • AWS CLI(已配置凭证)
  • Python 3.11+

第一步:创建 Flask 应用

创建项目目录和文件:

1
mkdir flask-lambda && cd flask-lambda

app.py - 一个简单的 Flask API:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/')
def health():
return jsonify(status='healthy')

@app.route('/api/hello')
def hello():
name = request.args.get('name', 'World')
return jsonify(message=f'Hello, {name}!')

@app.route('/api/echo', methods=['POST'])
def echo():
data = request.get_json()
return jsonify(received=data)

requirements.txt

1
2
flask==3.0.0
gunicorn==21.2.0

第二步:配置 Gunicorn

gunicorn.conf.py - 针对 Lambda 优化的配置:

1
2
3
4
5
6
7
8
bind = '0.0.0.0:8080'
workers = 1 # Lambda 单实例,1 个 worker 足够
threads = 4 # 多线程处理并发
timeout = 30
keepalive = 2
accesslog = '-' # 输出到 stdout,方便 CloudWatch 收集
errorlog = '-'
loglevel = 'info'

第三步:编写 Dockerfile

Dockerfile

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
FROM python:3.11-slim

# 添加 Lambda Web Adapter
COPY --from=public.ecr.aws/awsguru/aws-lambda-adapter:0.9.1 /lambda-adapter /opt/extensions/lambda-adapter

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8080

CMD ["gunicorn", "-c", "gunicorn.conf.py", "app:app"]

核心就是这一行 COPY --from=...,它从 AWS 公共 ECR 仓库拉取 Lambda Web Adapter 二进制文件,放到 /opt/extensions/ 目录。Lambda 启动时会自动加载这个 Extension。

第四步:本地测试

构建前先登录 ECR Public:

1
2
aws ecr-public get-login-password --region us-east-1 | \
docker login --username AWS --password-stdin public.ecr.aws

构建并运行:

1
2
docker build --platform linux/amd64 -t flask-lambda .
docker run -p 8080:8080 flask-lambda

测试:

1
2
3
4
5
6
7
8
9
10
curl http://localhost:8080/
# {"status":"healthy"}

curl "http://localhost:8080/api/hello?name=Lambda"
# {"message":"Hello, Lambda!"}

curl -X POST http://localhost:8080/api/echo \
-H "Content-Type: application/json" \
-d '{"msg":"hello"}'
# {"received":{"msg":"hello"}}

本地没问题,接下来部署到 AWS。

第五步:推送镜像到 ECR

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 设置变量
AWS_REGION="ap-northeast-1"
AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
ECR_REPO="flask-lambda"

# 创建 ECR 仓库
aws ecr create-repository --repository-name $ECR_REPO --region $AWS_REGION

# 登录 ECR
aws ecr get-login-password --region $AWS_REGION | \
docker login --username AWS --password-stdin \
$AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com

# 打标签并推送
docker tag flask-lambda:latest \
$AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO:latest

docker push $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO:latest

第六步:创建 Lambda 函数

首先创建 IAM 角色:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 创建信任策略
cat > trust-policy.json << 'EOF'
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole"
}]
}
EOF

# 创建角色并附加权限
aws iam create-role \
--role-name flask-lambda-role \
--assume-role-policy-document file://trust-policy.json

aws iam attach-role-policy \
--role-name flask-lambda-role \
--policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

sleep 10 # 等待角色生效

创建 Lambda 函数:

1
2
3
4
5
6
7
8
aws lambda create-function \
--function-name flask-api \
--package-type Image \
--code ImageUri=$AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO:latest \
--role arn:aws:iam::$AWS_ACCOUNT_ID:role/flask-lambda-role \
--timeout 30 \
--memory-size 512 \
--region $AWS_REGION

第七步:配置 API Gateway

打开 API Gateway 控制台,创建 REST API:

  1. 点击「创建 API」→ 选择「REST API」→「构建」
  2. 输入 API 名称,如 flask-api
  3. 创建资源:
    • 点击「创建资源」
    • 勾选「代理资源」
    • 资源路径填 {proxy+}
    • 点击「创建资源」
  4. 设置集成:
    • 集成类型选「Lambda 函数」
    • 勾选「Lambda 代理集成」
    • 选择你的 Lambda 函数 flask-api
  5. 同样为根路径 / 创建 ANY 方法,集成到同一个 Lambda
  6. 点击「部署 API」,阶段名填 v1

部署完成后,你会得到一个调用 URL,类似:

1
https://xxxxxx.execute-api.ap-northeast-1.amazonaws.com/v1

第八步:测试 API

1
2
3
4
5
6
7
API_URL="https://xxxxxx.execute-api.ap-northeast-1.amazonaws.com/v1"

curl $API_URL/
# {"status":"healthy"}

curl "$API_URL/api/hello?name=Serverless"
# {"message":"Hello, Serverless!"}

🎉 大功告成!你的 Flask API 已经运行在 Lambda 上了。

进阶:IAM 认证

如果你的 API 需要认证,可以在 API Gateway 中启用 IAM 认证。调用时需要对请求进行 SigV4 签名:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
import requests

def signed_request(method, url, data=None, headers=None):
session = boto3.Session()
credentials = session.get_credentials()

headers = headers or {}
request = AWSRequest(method=method, url=url, data=data, headers=headers)
# 注意:API Gateway 用 execute-api,Function URL 用 lambda
SigV4Auth(credentials, "execute-api", "ap-northeast-1").add_auth(request)

return requests.request(
method=method,
url=url,
headers=dict(request.headers),
data=data
)

resp = signed_request("GET", f"{API_URL}/api/hello")
print(resp.json())

更新部署

代码更新后,只需重新构建镜像并更新 Lambda:

1
2
3
4
5
6
7
docker build --platform linux/amd64 -t flask-lambda .
docker tag flask-lambda:latest $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO:latest
docker push $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO:latest

aws lambda update-function-code \
--function-name flask-api \
--image-uri $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO:latest

冷启动优化

首次请求可能需要 1-2 秒(冷启动),可以通过以下方式优化:

  1. 启用 Provisioned Concurrency:预热实例,消除冷启动
  2. 减小镜像体积:使用 slim 基础镜像,减少依赖
  3. 启用异步初始化:设置环境变量 AWS_LWA_ASYNC_INIT=true

总结

使用 Lambda Web Adapter,你可以:

  • 保持 Flask 代码不变
  • 使用熟悉的 Gunicorn 生产配置
  • 同一镜像本地和云端都能运行
  • 享受 Serverless 的弹性伸缩和按需付费

参考资料

使用 OpenList 将 S3 转换为 WebDAV

Amazon S3 是一种高可扩展、低延迟的对象存储服务,广泛用于存储和管理数据。尽管目前有多个工具来将 S3 与其他存储解决方案(如 Storage Gateway,EMRFS 或者 S3FS 等)集成,但今天我们介绍一个新的方法,通过使用 OpenList,将 S3 存储转换为 WebDAV,简化文件管理和访问。

本文将引导你通过 Docker Compose 启动 OpenList,并将其与 Amazon S3 配置,以便通过 WebDAV 协议进行访问。

阅读更多

EC2 没有绑定 EIP,重启后 IP 会变?DDNS-GO 自动更新你的域名

在 IPV4 即将枯竭的年代,云服务商的 EIP 也开始收费了。对于没有弹性公网 IP(EIP)的云服务器,我们可以通过 DDNS-GO 实现动态域名解析。本文介绍了如何使用 Docker 快速部署 DDNS-GO,并借助 DDNS 实现动态域名绑定,从而让服务器即使公网 IP 变化,也能够实时更新域名解析记录,这样只需要使用域名访问,不再需要在控制台查看。


安装 Docker

使用官方安装脚本快速安装 Docker:

1
2
curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh
阅读更多

组队参加 Amazon Q Idea 1000,我们的作品上了 AWS 峰会

Amazon Transit Gateway (TGW) 是一个强大的网络连接服务,用于在不同的 VPC(虚拟私有云)之间实现高效互联。本文将指导您如何创建和配置 TGW,以便实现跨账户和跨区域的 VPC 互联。

VPC Peering 的局限性

点对点连接:VPC Peering 是一个点对点的连接,每次只能连接两个 VPC。如果需要连接多个 VPC,需要为每对 VPC 单独设置 Peering 连接,也就是我们常说的不能进行路由的传递,需要打通的 VPC 很多的时候会非常的麻烦。

阅读更多

基于 AWS IAM Identity Center 的 SAML 配置,实现单点登录阿里云

在企业多云环境日益普及的今天,很多组织不仅使用 AWS(亚马逊云科技) 作为主要的计算与存储平台,同时也会使用 阿里云 来满足本地合规、地域性需求或价格优势。如何在多个云平台之间实现 统一身份认证,避免用户维护多个账号与密码,已经成为企业安全与运维中的核心问题。

AWS IAM Identity Center(身份中心,原 AWS SSO)作为 AWS 官方提供的集中式身份认证与访问管理服务,可以作为企业的 主身份提供商(IdP)。通过 SAML 2.0 协议,它能够将认证结果传递给其他云服务商(如阿里云),让用户在 AWS 完成一次身份验证后,直接进入阿里云控制台,而无需再次登录。这就是所谓的 跨云单点登录(Single Sign-On, SSO)

阅读更多

组队参加 Amazon Q Idea 1000,我们的作品上了 AWS 峰会

最近生成式 AI 的比赛很多,也报名了 AWS Idea1000 的比赛,作品登录上了 AWS 的峰会。

我们的产品名称是“拍立懂”,团队名称是 “凌晨三点的夜”。主要是拍照识别商品成分、品牌资质,分析价格合理性,为老年人提供购买决策建议;通过实时语音聊天交互亲人语音陪伴老年人购物,满足空巢老人的情感空缺。

我们的项目团队汇聚了文化创意创业者、资深互联网产品经理、互联网技术博主与 AI 前端工程师等多元背景,形成从商业策略到产品落地的完整闭环。团队成员对创新技术和用户价值怀有共同的热情,彼此协作、优势互补,致力于在商业模式和技术实现上持续突破,为项目注入持久动力。

阅读更多

如何在 AWS EC2 上部署 Isaac Sim

网上已经有一些关于在阿里云和腾讯云上部署 Isaac Sim 的教程,本文将带大家了解如何在 AWS EC2 上部署 NVIDIA Isaac Sim 仿真平台,并以 A10G GPU(g5.2xlarge 实例)为例进行实战操作。


一、环境说明

  • GPU 类型:A10G(适用于 RTX 渲染)
  • 实例类型g5.2xlarge
  • 操作系统镜像(AMI)
    Deep Learning OSS Nvidia Driver AMI GPU PyTorch 2.7 (Ubuntu 22.04)
    👉 该镜像自带 NVIDIA 驱动、CUDA、Docker、nvidia-docker,无需手动安装

阅读更多

手把手搞定报名亚马逊科技认证

亚马逊云科技认证考试为我们这些技术从业者提供了提升专业技能的机会。无论选择线上还是线下考试,每种方式都有其独特的优势和挑战。选择合适的考试方式将帮助我们更好地展示自己的技术水平。以下是我对不同考试方式的优缺点介绍,以及各科目的考试代码。希望这些信息能帮助大家在备考过程中做出最优选择。

阅读更多
Your browser is out-of-date!

Update your browser to view this website correctly.&npsb;Update my browser now

×