Python Strands 代理
生成一个用于构建AI智能体工具的Python Strands Agent,并可选部署到Amazon Bedrock AgentCore Runtime。
什么是Strands?
Section titled “什么是Strands?”Strands是一个轻量级、生产就绪的Python框架,用于构建AI智能体。主要特性包括:
- 轻量且可定制:简洁的智能体循环设计,不干扰开发
- 生产就绪:完整的可观测性、追踪和规模化部署选项
- 模型与供应商无关:支持多种不同供应商的模型
- 社区驱动工具:强大的社区贡献工具集
- 多智能体支持:支持智能体团队和自主智能体等高级技术
- 灵活交互模式:支持对话式、流式和非流式交互
生成Strands智能体
Section titled “生成Strands智能体”可通过两种方式生成Python Strands智能体:
- 安装 Nx Console VSCode Plugin 如果您尚未安装
- 在VSCode中打开Nx控制台
- 点击
Generate (UI)在"Common Nx Commands"部分 - 搜索
@aws/nx-plugin - py#strands-agent - 填写必需参数
- 点击
Generate
pnpm nx g @aws/nx-plugin:py#strands-agentyarn nx g @aws/nx-plugin:py#strands-agentnpx nx g @aws/nx-plugin:py#strands-agentbunx nx g @aws/nx-plugin:py#strands-agent您还可以执行试运行以查看哪些文件会被更改
pnpm nx g @aws/nx-plugin:py#strands-agent --dry-runyarn nx g @aws/nx-plugin:py#strands-agent --dry-runnpx nx g @aws/nx-plugin:py#strands-agent --dry-runbunx nx g @aws/nx-plugin:py#strands-agent --dry-run| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| project 必需 | string | - | 要添加 Strands Agent 的项目 |
| computeType | string | BedrockAgentCoreRuntime | 托管 Strands Agent 的计算类型 |
| name | string | - | Strands Agent 的名称(默认:agent) |
| auth | string | IAM | 用于与 Strands Agent 进行身份验证的方法。可选择 IAM(默认)或 Cognito。 |
| iacProvider | string | Inherit | 首选的 IaC 提供商。默认情况下,这将继承自您的初始选择。 |
生成器将在现有Python项目中添加以下文件:
文件夹your-project/
文件夹your_module/
文件夹agent/ (可自定义名称)
- __init__.py Python包初始化文件
- init.py 配置CORS和错误处理中间件的FastAPI应用设置
- agent.py 主智能体定义(含示例工具)
- main.py Bedrock AgentCore Runtime入口点
- Dockerfile 智能体托管入口文件(当
computeType设为None时不生成)
- pyproject.toml 更新Strands依赖
- project.json 更新服务目标配置
由于该生成器会根据您选择的 iacProvider 以基础设施即代码的形式输出,它将在 packages/common 目录下创建一个包含相关 CDK 构造体或 Terraform 模块的项目。
通用的基础设施即代码项目结构如下:
文件夹packages/common/constructs
文件夹src
文件夹app/ 针对特定项目/生成器的基础设施构造体
- …
文件夹core/ 被
app目录构造体重用的通用构造体- …
- index.ts 导出
app目录构造体的入口文件
- project.json 项目构建目标与配置
文件夹packages/common/terraform
文件夹src
文件夹app/ 针对特定项目/生成器的 Terraform 模块
- …
文件夹core/ 被
app目录模块重用的通用模块- …
- project.json 项目构建目标与配置
部署Strands智能体时生成以下文件:
文件夹packages/common/constructs/src
文件夹app
文件夹agents
文件夹<project-name>
- <project-name>.ts 部署智能体的CDK构造
- Dockerfile CDK构造使用的透传Docker文件
文件夹packages/common/terraform/src
文件夹app
文件夹agents
文件夹<project-name>
- <project-name>.tf 部署智能体的Terraform模块
文件夹core
文件夹agent-core
- runtime.tf 部署到Bedrock AgentCore Runtime的通用模块
使用Strands智能体
Section titled “使用Strands智能体”工具是AI智能体可调用的功能函数。Strands框架使用基于装饰器的简单方式定义工具。
在agent.py文件中添加新工具:
from strands import Agent, tool
@tooldef calculate_sum(numbers: list[int]) -> int: """计算数字列表的总和""" return sum(numbers)
@tooldef get_weather(city: str) -> str: """获取城市天气信息""" # 此处集成天气API return f"{city}天气:晴,25°C"
# 将工具添加到智能体agent = Agent( system_prompt="您是一个拥有多种工具的智能助手。", tools=[calculate_sum, get_weather],)Strands框架自动处理:
- 基于函数类型提示的类型验证
- 工具调用的JSON模式生成
- 错误处理和响应格式化
使用预置工具
Section titled “使用预置工具”通过strands-tools包使用预置工具:
from strands_tools import current_time, http_request, file_read
agent = Agent( system_prompt="您是一个智能助手。", tools=[current_time, http_request, file_read],)默认使用Claude 4 Sonnet模型,可自定义模型供应商。参考Strands模型供应商文档进行配置:
from strands import Agentfrom strands.models import BedrockModel
# 创建Bedrock模型实例bedrock_model = BedrockModel( model_id="anthropic.claude-sonnet-4-20250514-v1:0", region_name="us-west-2", temperature=0.3,)
agent = Agent(model=bedrock_model)使用MCP服务
Section titled “使用MCP服务”可通过添加MCP服务工具扩展智能体功能。
对于使用py#mcp-server或ts#mcp-server生成器创建的MCP服务,可以使用connection生成器。
- 安装 Nx Console VSCode Plugin 如果您尚未安装
- 在VSCode中打开Nx控制台
- 点击
Generate (UI)在"Common Nx Commands"部分 - 搜索
@aws/nx-plugin - connection - 填写必需参数
- 点击
Generate
pnpm nx g @aws/nx-plugin:connectionyarn nx g @aws/nx-plugin:connectionnpx nx g @aws/nx-plugin:connectionbunx nx g @aws/nx-plugin:connection关于连接设置的详细信息,请参考connection生成器指南。
对于其他MCP服务,请参考Strands文档。
深入指南请参考Strands官方文档。
FastAPI服务器
Section titled “FastAPI服务器”生成器使用FastAPI为Strands智能体创建HTTP服务器。FastAPI提供了一个现代、高性能的Python API构建框架,具有自动API文档生成和类型验证功能。
生成的服务器包含:
- 配置CORS中间件的FastAPI应用
- 错误处理中间件
- OpenAPI模式生成
- 健康检查端点(
/ping) - 智能体调用端点(
/invocations)
使用Pydantic自定义调用输入和输出
Section titled “使用Pydantic自定义调用输入和输出”智能体的调用端点使用Pydantic模型来定义和验证请求及响应模式。您可以在main.py中自定义这些模型以满足智能体需求。
定义输入模型
Section titled “定义输入模型”默认的InvokeInput模型接受提示词和会话ID。
from pydantic import BaseModel
class InvokeInput(BaseModel): prompt: str session_id: str您可以扩展此模型以包含智能体所需的任何额外字段。
定义输出模型
Section titled “定义输出模型”对于流式响应,生成器提供JsonStreamingResponse,它会自动将Pydantic模型序列化为JSON Lines格式(application/jsonl)。该格式兼容OpenAPI 3.2的流式规范,并与生成的TypeScript客户端无缝协作。
默认情况下,智能体yield包含智能体响应文本的StreamChunk对象:
class StreamChunk(BaseModel): content: str您可以自定义StreamChunk模型以满足需求:
from pydantic import BaseModel
class StreamChunk(BaseModel): content: str timestamp: str token_count: intFastAPI有一个关于原生支持的开放功能请求。
Bedrock AgentCore Python SDK
Section titled “Bedrock AgentCore Python SDK”生成器包含对Bedrock AgentCore Python SDK的依赖,用于PingStatus常量。如果需要,可以直接使用BedrockAgentCoreApp替代FastAPI,但请注意会失去类型安全性。
SDK功能详情请参阅文档。
运行Strands智能体
Section titled “运行Strands智能体”生成器配置了名为<your-agent-name>-serve的目标,用于本地开发和测试:
pnpm nx agent-serve your-projectyarn nx agent-serve your-projectnpx nx agent-serve your-projectbunx nx agent-serve your-project该命令使用uv run通过Bedrock AgentCore Python SDK运行智能体。
部署到Bedrock AgentCore Runtime
Section titled “部署到Bedrock AgentCore Runtime”基础设施即代码
Section titled “基础设施即代码”如果您为 computeType 选择了 BedrockAgentCoreRuntime,将生成相关的 CDK 或 Terraform 基础设施,您可以使用它将 Strands Agent 部署到 Amazon Bedrock AgentCore Runtime。
系统会为您的代理生成一个 CDK 构造,其名称基于您运行生成器时选择的 name,或默认为 <ProjectName>Agent。
您可以在 CDK 应用程序中使用此 CDK 构造:
import { MyProjectAgent } from ':my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { // Add the agent to your stack const agent = new MyProjectAgent(this, 'MyProjectAgent');
// Grant permissions to invoke the relevant models in bedrock agent.agentCoreRuntime.addToRolePolicy( new PolicyStatement({ actions: [ 'bedrock:InvokeModel', 'bedrock:InvokeModelWithResponseStream', ], // You can scope the below down to the specific models you use resources: [ 'arn:aws:bedrock:*:*:foundation-model/*', 'arn:aws:bedrock:*:*:inference-profile/*', ], }), ); }}系统会为您生成一个 Terraform 模块,其名称基于您运行生成器时选择的 name,或默认为 <ProjectName>-agent。
您可以在 Terraform 项目中使用此 terraform 模块:
# Agentmodule "my_project_agent" { # Relative path to the generated module in the common/terraform project source = "../../common/terraform/src/app/agents/my-project-agent"
# Grant permissions to invoke the relevant models in bedrock additional_iam_policy_statements = [ { Effect = "Allow" Action = [ "bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream" ] # You can scope the below down to the specific models you use Resource = [ "arn:aws:bedrock:*:*:foundation-model/*", "arn:aws:bedrock:*:*:inference-profile/*" ] } ]}生成器提供了一个 auth 选项来配置 Strands Agent 的身份验证。您可以在生成代理时选择 IAM(默认)或 Cognito 身份验证。
默认情况下,您的 Strands Agent 将使用 IAM 身份验证进行保护,只需在不带任何参数的情况下部署它:
import { MyProjectAgent } from ':my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { new MyProjectAgent(this, 'MyProjectAgent'); }}您可以使用 grantInvokeAccess 方法授予调用 Bedrock AgentCore Runtime 上代理的访问权限,例如:
import { MyProjectAgent } from ':my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { const agent = new MyProjectAgent(this, 'MyProjectAgent'); const lambdaFunction = new Function(this, ...);
agent.grantInvokeAccess(lambdaFunction); }}# Agentmodule "my_project_agent" { # Relative path to the generated module in the common/terraform project source = "../../common/terraform/src/app/agents/my-project-agent"}要授予调用代理的访问权限,您需要添加如下策略,引用 module.my_project_agent.agent_core_runtime_arn 输出:
{ Effect = "Allow" Action = [ "bedrock-agentcore:InvokeAgentRuntime" ] Resource = [ module.my_project_agent.agent_core_runtime_arn, "${module.my_project_agent.agent_core_runtime_arn}/*" ]}Cognito 身份验证
Section titled “Cognito 身份验证”当您选择 Cognito 身份验证时,生成器会配置代理使用 Cognito 进行身份验证。
生成的构造接受一个 identity 属性来配置 Cognito 身份验证:
import { MyProjectAgent, UserIdentity } from ':my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { const identity = new UserIdentity(this, 'Identity');
new MyProjectAgent(this, 'MyProjectAgent', { identity, }); }}UserIdentity 构造可以使用 ts#react-website#auth 生成器生成,或者您可以创建自己的 CDK UserPool 和 UserPoolClient。
生成的模块接受 user_pool_id 和 user_pool_client_ids 变量用于 Cognito 身份验证:
module "user_identity" { source = "../../common/terraform/src/core/user-identity"}
module "my_project_agent" { source = "../../common/terraform/src/app/agents/my-project-agent"
user_pool_id = module.user_identity.user_pool_id user_pool_client_ids = [module.user_identity.user_pool_client_id]}打包与Docker目标
Section titled “打包与Docker目标”为构建Bedrock AgentCore Runtime部署包,项目中添加了bundle目标:
- 使用
uv export导出Python依赖到requirements.txt - 使用
uv pip install为目标平台(aarch64-manylinux2014)安装依赖
同时添加智能体专用的docker目标:
- 根据AgentCore运行时协议构建Docker镜像
通过Dockerfile中的自动配置,使用AWS分布式OpenTelemetry(ADOT)实现可观测性。
在CloudWatch控制台选择”GenAI Observability”查看追踪数据,需启用事务搜索。
详细配置参考AgentCore可观测性文档。
调用Strands智能体
Section titled “调用Strands智能体”调用本地服务器
Section titled “调用本地服务器”要调用通过<your-agent-name>-serve目标本地运行的智能体,可以向本地智能体运行端口的/invocations发送简单的POST请求。例如,使用curl:
curl -N -X POST http://localhost:8081/invocations \ -d '{"prompt": "what is 3 + 5?", "session_id": "abcdefghijklmnopqrstuvwxyz0123456789"}' \ -H "Content-Type: application/json"调用已部署的智能体
Section titled “调用已部署的智能体”要调用部署到 Bedrock AgentCore Runtime 的 Agent,您可以使用 URL 编码的运行时 ARN 向 Bedrock AgentCore Runtime 数据平面端点发送 POST 请求。
您可以按如下方式从基础设施中获取运行时 ARN:
import { CfnOutput } from 'aws-cdk-lib';import { MyProjectAgent } from ':my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { const agent = new MyProjectAgent(this, 'MyProjectAgent');
new CfnOutput(this, 'AgentArn', { value: agent.agentCoreRuntime.agentRuntimeArn, }); }}# Agentmodule "my_project_agent" { # Relative path to the generated module in the common/terraform project source = "../../common/terraform/src/app/agents/my-project-agent"}
output "agent_arn" { value = module.my_project_agent.agent_core_runtime_arn}ARN 将具有以下格式:arn:aws:bedrock-agentcore:<region>:<account>:runtime/<agent-runtime-id>。
然后,您可以通过将 : 替换为 %3A 和将 / 替换为 %2F 来对 ARN 进行 URL 编码。
用于调用 agent 的 Bedrock AgentCore Runtime 数据平面 URL 如下:
https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<url-encoded-arn>/invocations调用此 URL 的具体方式取决于所使用的身份验证方法。
对于IAM认证,请求必须使用AWS Signature Version 4(SigV4)签名。
acurl <region> bedrock-agentcore -N -X POST \'https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<url-encoded-arn>/invocations' \-d '{"prompt": "what is 3 + 5?", "session_id": "abcdefghijklmnopqrstuvwxyz0123456789"}' \-H 'Content-Type: application/json'支持Sigv4的curl
你可以将以下脚本添加到你的 .bashrc 文件中(并 source 它),或者将以下内容粘贴到你希望运行命令的同一个终端中。
acurl () { REGION=$1 SERVICE=$2 shift; shift; curl --aws-sigv4 "aws:amz:$REGION:$SERVICE" --user "$(aws configure get aws_access_key_id):$(aws configure get aws_secret_access_key)" -H "X-Amz-Security-Token: $(aws configure get aws_session_token)" "$@"}要发起一个 sigv4 认证的 curl 请求,按如下方式调用 acurl:
acurl <region> <service> <other-curl-arguments>例如:
API Gateway
Section titled “API Gateway”acurl ap-southeast-2 execute-api -X GET https://xxxStreaming Lambda function url
Section titled “Streaming Lambda function url”acurl ap-southeast-2 lambda -N -X POST https://xxx你可以将以下函数添加到你的 PowerShell 配置文件中,或者将以下内容粘贴到你希望运行命令的同一个 PowerShell 会话中。
# PowerShell profile or current sessionfunction acurl { param( [Parameter(Mandatory=$true)][string]$Region, [Parameter(Mandatory=$true)][string]$Service, [Parameter(ValueFromRemainingArguments=$true)][string[]]$CurlArgs )
$AccessKey = aws configure get aws_access_key_id $SecretKey = aws configure get aws_secret_access_key $SessionToken = aws configure get aws_session_token
& curl --aws-sigv4 "aws:amz:$Region`:$Service" --user "$AccessKey`:$SecretKey" -H "X-Amz-Security-Token: $SessionToken" @CurlArgs}要发起一个 sigv4 认证的 curl 请求,使用以下示例调用 acurl:
API Gateway
Section titled “API Gateway”acurl ap-southeast-2 execute-api -X GET https://xxxStreaming Lambda function url
Section titled “Streaming Lambda function url”acurl ap-southeast-2 lambda -N -X POST https://xxxJWT / Cognito认证
Section titled “JWT / Cognito认证”对于Cognito认证,在Authorization标头中传递Cognito访问令牌:
curl -N -X POST 'https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<url-encoded-arn>/invocations' \ -d '{"prompt": "what is 3 + 5?", "session_id": "abcdefghijklmnopqrstuvwxyz0123456789"}' \ -H "Content-Type: application/json" \ -H "Authorization: Bearer <access-token>"可使用AWS CLI的cognito-idp admin-initiate-auth命令获取访问令牌,例如:
aws cognito-idp admin-initiate-auth \ --user-pool-id <user-pool-id> \ --client-id <user-pool-client-id> \ --auth-flow ADMIN_NO_SRP_AUTH \ --auth-parameters USERNAME=<username>,PASSWORD=<password> \ --region <region> \ --query 'AuthenticationResult.AccessToken' \ --output text