1. Strands Agents SDK 初探用 Python 构建 AWS 服务调用代理最近在探索如何快速构建一个能调用 AWS 服务的 AI 代理时发现了 Strands Agents SDK 这个开源工具包。作为一个长期在 AWS 生态工作的开发者我对这种能简化 AI 代理开发的框架特别感兴趣。Strands 最吸引我的地方在于它原生支持 AWS 服务集成同时保持了代码的简洁性 - 只需要几行 Python 代码就能创建一个功能完整的 AI 代理。1.1 为什么选择 Strands Agents SDK在评估了多个 AI 代理框架后Strands 脱颖而出有几个关键原因AWS 原生集成直接支持 Amazon Bedrock 等 AWS AI 服务无需额外配置生产就绪内置了错误处理、日志记录和监控功能简洁的 API用装饰器定义工具函数直观易用开源灵活可以根据需要扩展和定制特别适合需要快速构建原型又要考虑生产部署的场景。下面我就带大家从零开始用 Strands 构建一个能调用 AWS 服务的 AI 代理。2. 环境准备与 SDK 安装2.1 基础环境配置在开始之前我们需要准备好开发环境Python 环境建议使用 Python 3.9 或更高版本AWS 凭证配置好 AWS CLI 或 boto3 的访问凭证虚拟环境推荐使用 venv 或 conda 创建隔离环境# 创建并激活虚拟环境 python -m venv strands-env source strands-env/bin/activate # Linux/Mac # strands-env\Scripts\activate # Windows2.2 安装 Strands Agents SDK安装过程非常简单只需要一个 pip 命令pip install strands-agents同时安装 AWS 相关的依赖pip install boto3注意如果遇到权限问题可以加上--user参数或在虚拟环境中安装3. 构建第一个 AWS 服务调用代理3.1 创建基础代理结构我们先创建一个最简单的代理框架from strands import Agent, tool tool def list_s3_buckets() - list: 列出当前账户下的所有S3存储桶 import boto3 s3 boto3.client(s3) response s3.list_buckets() return [bucket[Name] for bucket in response[Buckets]] # 创建代理实例 aws_agent Agent( tools[list_s3_buckets], system_prompt你是一个AWS服务助手可以帮助用户管理AWS资源 ) # 测试代理 response aws_agent(我有哪些S3存储桶) print(response)这个简单的例子展示了 Strands 的核心概念tool装饰器将普通函数转换为代理可用的工具Agent类是代理的核心负责协调工具调用和对话管理工具函数可以自由使用任何 Python 库包括 boto33.2 添加更多 AWS 服务功能让我们扩展代理的功能加入更多 AWS 服务tool def ec2_instance_status(instance_id: str) - dict: 获取EC2实例的状态信息 ec2 boto3.client(ec2) response ec2.describe_instances(InstanceIds[instance_id]) return { state: response[Reservations][0][Instances][0][State][Name], type: response[Reservations][0][Instances][0][InstanceType], launch_time: response[Reservations][0][Instances][0][LaunchTime].isoformat() } tool def lambda_invoke(function_name: str, payload: dict None) - dict: 调用Lambda函数 lambda_client boto3.client(lambda) response lambda_client.invoke( FunctionNamefunction_name, Payloadjson.dumps(payload) if payload else b ) return { status_code: response[StatusCode], payload: json.loads(response[Payload].read().decode(utf-8)) } # 更新代理配置 aws_agent Agent( tools[list_s3_buckets, ec2_instance_status, lambda_invoke], system_prompt你是一个全功能的AWS服务助手 )现在代理可以处理三种 AWS 服务S3、EC2 和 Lambda。每个工具函数都有清晰的文档字符串这很重要因为 Strands 会使用这些描述来指导代理如何调用工具。4. 高级功能与最佳实践4.1 添加权限控制和验证在生产环境中我们需要确保代理的操作是安全的from strands.hooks import BeforeToolCallEvent def validate_aws_operations(event: BeforeToolCallEvent): 验证AWS操作请求 if event.tool_use[name] lambda_invoke: function_name event.tool_use[input].get(function_name, ) if production in function_name.lower(): event.cancel_tool 不允许调用生产环境Lambda函数 aws_agent Agent( tools[list_s3_buckets, ec2_instance_status, lambda_invoke], hooks[validate_aws_operations], system_prompt你是一个安全的AWS服务助手 )这个钩子函数会在每次工具调用前执行我们可以在这里添加各种安全检查。4.2 结构化输出让代理返回结构化的数据方便后续处理from pydantic import BaseModel, Field class EC2InstanceReport(BaseModel): instance_id: str Field(..., description实例ID) state: str Field(..., description运行状态) uptime: str Field(..., description运行时长) recommendations: list[str] Field(default_factorylist, description优化建议) tool def get_ec2_report(instance_id: str) - EC2InstanceReport: 生成EC2实例的详细报告 status ec2_instance_status(instance_id) # 计算运行时长等更多信息... return EC2InstanceReport( instance_idinstance_id, statestatus[state], uptime2天3小时, recommendations[可以考虑使用更小的实例类型] ) # 使用结构化输出 response aws_agent( 为实例i-123456789生成详细报告, structured_output_modelEC2InstanceReport ) report response.structured_output print(f实例状态: {report.state})4.3 错误处理与重试AWS 服务调用可能会遇到各种临时性问题我们需要妥善处理from tenacity import retry, stop_after_attempt, wait_exponential tool retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def describe_rds_instance(instance_id: str) - dict: 获取RDS实例详情 rds boto3.client(rds) try: response rds.describe_db_instances(DBInstanceIdentifierinstance_id) return response[DBInstances][0] except Exception as e: print(f获取RDS实例失败: {str(e)}) raise这个例子使用了 tenacity 库来实现指数退避重试是处理 AWS API 调用的最佳实践。5. 部署与监控5.1 本地测试与调试Strands 提供了方便的调试工具from strands.hooks import AfterToolCallEvent def log_tool_calls(event: AfterToolCallEvent): 记录所有工具调用 print(f工具调用: {event.tool_use[name]}) print(f输入: {event.tool_use[input]}) print(f结果: {event.result[status]}) aws_agent Agent( tools[...], hooks[validate_aws_operations, log_tool_calls], trace_attributes{ service: aws-agent, env: development } )5.2 部署到 AWS Lambda将代理部署为无服务器函数import json def lambda_handler(event, context): agent_input event.get(input, ) response aws_agent(agent_input) return { statusCode: 200, body: json.dumps({ output: str(response), structured_output: response.structured_output.dict() if hasattr(response, structured_output) else None }) }可以使用 AWS SAM 或 CDK 来部署这个 Lambda 函数。6. 实际应用案例6.1 自动化运维机器人tool def restart_ec2_instance(instance_id: str) - str: 重启EC2实例 ec2 boto3.client(ec2) ec2.reboot_instances(InstanceIds[instance_id]) return f实例 {instance_id} 重启命令已发送 tool def scale_ec2_auto_scaling_group(group_name: str, capacity: int) - str: 调整自动扩展组容量 autoscaling boto3.client(autoscaling) autoscaling.set_desired_capacity( AutoScalingGroupNamegroup_name, DesiredCapacitycapacity ) return f扩展组 {group_name} 容量已设置为 {capacity} ops_agent Agent( tools[restart_ec2_instance, scale_ec2_auto_scaling_group, ec2_instance_status], system_prompt你是一个AWS运维助手可以执行常见的运维操作, hooks[validate_aws_operations] )6.2 成本优化顾问tool def get_cost_and_usage(start_date: str, end_date: str) - dict: 获取指定时间范围内的AWS成本数据 ce boto3.client(ce) response ce.get_cost_and_usage( TimePeriod{Start: start_date, End: end_date}, GranularityDAILY, Metrics[UnblendedCost] ) return response cost_agent Agent( tools[get_cost_and_usage, list_s3_buckets, ec2_instance_status], system_prompt你是一个AWS成本优化助手可以帮助分析和管理云成本 )7. 性能优化技巧7.1 并发工具调用Strands 支持并行执行多个工具调用from strands import ParallelToolGroup tool def check_s3_bucket_encryption(bucket_name: str) - bool: 检查S3存储桶是否启用了加密 s3 boto3.client(s3) try: result s3.get_bucket_encryption(Bucketbucket_name) return True except: return False tool def check_s3_bucket_logging(bucket_name: str) - bool: 检查S3存储桶是否启用了日志记录 s3 boto3.client(s3) try: result s3.get_bucket_logging(Bucketbucket_name) return bool(result.get(LoggingEnabled)) except: return False security_agent Agent( tool_groups{ s3_security_checks: ParallelToolGroup( tools[check_s3_bucket_encryption, check_s3_bucket_logging], description并行执行S3安全检查 ) }, system_prompt你是一个安全合规检查助手 ) # 使用方式 response security_agent(检查my-bucket的安全配置)7.2 缓存常用数据减少不必要的 API 调用from functools import lru_cache lru_cache(maxsize100) tool def get_vpc_details(vpc_id: str) - dict: 获取VPC详情带缓存 ec2 boto3.client(ec2) response ec2.describe_vpcs(VpcIds[vpc_id]) return response[Vpcs][0]8. 安全最佳实践8.1 最小权限原则确保代理使用的 IAM 角色只拥有必要权限{ Version: 2012-10-17, Statement: [ { Effect: Allow, Action: [ s3:ListAllMyBuckets, ec2:DescribeInstances, lambda:InvokeFunction ], Resource: * } ] }8.2 敏感数据保护from strands.hooks import BeforeToolCallEvent def redact_sensitive_data(event: BeforeToolCallEvent): 过滤敏感数据 if event.tool_use[name] lambda_invoke: if password in str(event.tool_use[input]): event.cancel_tool 请求中包含敏感字段 secure_agent Agent( tools[lambda_invoke], hooks[redact_sensitive_data] )9. 调试与问题排查9.1 常见问题及解决方案问题现象可能原因解决方案权限拒绝错误IAM 角色权限不足检查并更新代理的 IAM 策略工具调用超时AWS API 响应慢增加超时设置或实现重试逻辑结构化输出不符合预期Pydantic 模型定义不完整检查模型字段定义和类型提示代理不理解用户请求工具描述不够清晰完善工具函数的文档字符串9.2 调试技巧启用详细日志import logging logging.basicConfig(levellogging.DEBUG)使用 Strands 的追踪功能response aws_agent(列出我的S3存储桶, traceTrue) print(response.trace) # 查看完整的决策过程逐步测试工具函数# 直接调用工具函数测试 result list_s3_buckets() print(result)10. 扩展与集成10.1 与 Amazon Bedrock 集成from strands.llms import Bedrock aws_agent Agent( tools[...], llmBedrock(model_idanthropic.claude-v2), system_prompt... )10.2 创建多代理系统cost_agent Agent(...) security_agent Agent(...) tool def consult_experts(question: str) - str: 咨询专家团队 cost_response cost_agent(question) security_response security_agent(question) return f成本团队建议: {cost_response}\n安全团队建议: {security_response} chief_agent Agent( tools[consult_experts, ...], system_prompt你是首席AWS顾问可以协调各个专家团队 )通过这种方式我们可以构建复杂的多代理系统每个代理专注于特定领域。在实际项目中我发现 Strands Agents SDK 特别适合快速构建 AWS 相关的自动化工具和智能助手。它的设计既考虑了开发效率又兼顾了生产环境的需求。最让我印象深刻的是它的错误处理和监控能力这在运维场景中至关重要。