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 基础环境配置

在开始之前，我们需要准备好开发环境：

1. Python 环境 ：建议使用 Python 3.9 或更高版本
2. AWS 凭证 ：配置好 AWS CLI 或 boto3 的访问凭证
3. 虚拟环境 ：推荐使用 venv 或 conda 创建隔离环境

# 创建并激活虚拟环境
python -m venv strands-env
source strands-env/bin/activate  # Linux/Mac
# strands-env\Scripts\activate   # Windows

2.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 库，包括 boto3

3.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(
        FunctionName=function_name,
        Payload=json.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_factory=list, description="优化建议")

@tool
def get_ec2_report(instance_id: str) -> EC2InstanceReport:
    """生成EC2实例的详细报告"""
    status = ec2_instance_status(instance_id)
    # 计算运行时长等更多信息...
    return EC2InstanceReport(
        instance_id=instance_id,
        state=status['state'],
        uptime="2天3小时",
        recommendations=["可以考虑使用更小的实例类型"]
    )

# 使用结构化输出
response = aws_agent(
    "为实例i-123456789生成详细报告",
    structured_output_model=EC2InstanceReport
)
report = response.structured_output
print(f"实例状态: {report.state}")

4.3 错误处理与重试

AWS 服务调用可能会遇到各种临时性问题，我们需要妥善处理：

from tenacity import retry, stop_after_attempt, wait_exponential

@tool
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def describe_rds_instance(instance_id: str) -> dict:
    """获取RDS实例详情"""
    rds = boto3.client('rds')
    try:
        response = rds.describe_db_instances(DBInstanceIdentifier=instance_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(
        AutoScalingGroupName=group_name,
        DesiredCapacity=capacity
    )
    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},
        Granularity='DAILY',
        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(Bucket=bucket_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(Bucket=bucket_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(maxsize=100)
@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 调试技巧

1. 启用详细日志：

import logging
logging.basicConfig(level=logging.DEBUG)

2. 使用 Strands 的追踪功能：

response = aws_agent("列出我的S3存储桶", trace=True)
print(response.trace)  # 查看完整的决策过程

3. 逐步测试工具函数：

# 直接调用工具函数测试
result = list_s3_buckets()
print(result)

10. 扩展与集成

10.1 与 Amazon Bedrock 集成

from strands.llms import Bedrock

aws_agent = Agent(
    tools=[...],
    llm=Bedrock(model_id="anthropic.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 相关的自动化工具和智能助手。它的设计既考虑了开发效率，又兼顾了生产环境的需求。最让我印象深刻的是它的错误处理和监控能力，这在运维场景中至关重要。