跳转到内容

Python Strands 代理

生成一个用于构建AI代理的Python Strands Agent,并可选部署到Amazon Bedrock AgentCore Runtime

Strands是一个轻量级、生产就绪的Python框架,用于构建AI代理。主要特性包括:

  • 轻量且可定制:简洁的代理循环设计,不干扰开发
  • 生产就绪:完整的可观测性、追踪和规模化部署选项
  • 模型与供应商无关:支持多种供应商的不同模型
  • 社区驱动工具:强大的社区贡献工具集
  • 多代理支持:支持代理团队和自主代理等高级技术
  • 灵活交互模式:支持对话式、流式和非流式交互

可通过两种方式生成Python Strands代理:

  1. 安装 Nx Console VSCode Plugin 如果您尚未安装
  2. 在VSCode中打开Nx控制台
  3. 点击 Generate (UI) 在"Common Nx Commands"部分
  4. 搜索 @aws/nx-plugin - py#strands-agent
  5. 填写必需参数
    • 点击 Generate
    参数 类型 默认值 描述
    project 必需 string - The project to add the Strands Agent to
    computeType string BedrockAgentCoreRuntime The type of compute to host your Strands Agent.
    name string - The name of your Strands Agent (default: agent)
    iacProvider string CDK The preferred IaC provider

    生成器将在现有Python项目中添加以下文件:

    • 文件夹your-project/
      • 文件夹your_module/
        • 文件夹agent/ (可自定义名称)
          • __init__.py Python包初始化文件
          • agent.py 主代理定义(含示例工具)
          • main.py Bedrock AgentCore Runtime入口点
          • agentcore_mcp_client.py 用于调用Bedrock AgentCore Runtime上MCP服务的客户端工厂
          • 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 项目构建目标与配置

    部署Strands代理时生成以下文件:

    • 文件夹packages/common/constructs/src
      • 文件夹app
        • 文件夹agents
          • 文件夹<project-name>
            • <project-name>.ts 部署代理的CDK构造
            • Dockerfile CDK构造使用的Docker文件
      • 文件夹core
        • 文件夹agent-core
          • runtime.ts 部署到Bedrock AgentCore Runtime的通用CDK构造

    工具是AI代理可调用的功能函数。Strands框架使用基于装饰器的简单方式定义工具。

    agent.py文件中添加新工具:

    from strands import Agent, tool
    @tool
    def calculate_sum(numbers: list[int]) -> int:
    """计算数字列表的总和"""
    return sum(numbers)
    @tool
    def get_weather(city: str) -> str:
    """获取城市天气信息"""
    # 在此集成天气API
    return f"{city}天气:晴,25°C"
    # 将工具添加到代理
    agent = Agent(
    system_prompt="您是一个拥有多种工具的智能助手。",
    tools=[calculate_sum, get_weather],
    )

    Strands框架自动处理:

    • 基于函数类型提示的类型验证
    • 工具调用的JSON模式生成
    • 错误处理和响应格式化

    Strands通过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 Agent
    from 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服务工具扩展代理功能。

    对于使用py#mcp-serverts#mcp-server生成器创建的MCP服务(或部署在Bedrock AgentCore Runtime上的其他服务),生成器会在agentcore_mcp_client.py中创建客户端工厂。

    agent.py中更新get_agent方法以创建MCP客户端并添加工具。以下示例展示使用IAM(SigV4)认证:

    agent.py
    import os
    from contextlib import contextmanager
    import boto3
    from strands import Agent
    from .agentcore_mcp_client import AgentCoreMCPClient
    # 获取区域和凭证
    region = os.environ["AWS_REGION"]
    boto_session = boto3.Session(region_name=region)
    credentials = boto_session.get_credentials()
    @contextmanager
    def get_agent(session_id: str):
    mcp_client = AgentCoreMCPClient.with_iam_auth(
    agent_runtime_arn=os.environ["MCP_AGENTCORE_RUNTIME_ARN"],
    credentials=credentials,
    region=region,
    session_id=session_id,
    )
    with mcp_client:
    mcp_tools = mcp_client.list_tools_sync()
    yield Agent(
    system_prompt="..."
    tools=[*mcp_tools],
    )

    上述IAM认证示例需在基础设施中配置两点:添加MCP服务AgentCore Runtime ARN环境变量,并授予代理调用MCP服务的权限:

    import { MyProjectAgent, MyProjectMcpServer } from ':my-scope/common-constructs';
    export class ExampleStack extends Stack {
    constructor(scope: Construct, id: string) {
    const mcpServer = new MyProjectMcpServer(this, 'MyProjectMcpServer');
    const agent = new MyProjectAgent(this, 'MyProjectAgent', {
    environment: {
    MCP_AGENTCORE_RUNTIME_ARN: mcpServer.agentCoreRuntime.arn,
    },
    });
    mcpServer.agentCoreRuntime.grantInvoke(agent.agentCoreRuntime);
    }
    }

    深入指南请参考Strands官方文档

    生成器配置了Bedrock AgentCore Python SDK来管理AgentCore代理需要实现的底层HTTP协议。

    SDK功能详情请参阅官方文档

    生成器配置了名为<your-agent-name>-serve的目标,用于本地开发和测试:

    Terminal window
    pnpm nx run your-project:agent-serve

    该命令使用uv run通过Bedrock AgentCore Python SDK运行代理。

    若选择BedrockAgentCoreRuntime作为computeType,将生成对应的CDK或Terraform基础设施代码用于部署到Amazon Bedrock AgentCore Runtime

    默认生成基于生成器name参数的CDK构造(默认<ProjectName>Agent)。

    在CDK应用中使用:

    import { MyProjectAgent } from ':my-scope/common-constructs';
    export class ExampleStack extends Stack {
    constructor(scope: Construct, id: string) {
    // 添加代理到堆栈
    const agent = new MyProjectAgent(this, 'MyProjectAgent');
    // 授予调用Bedrock模型的权限
    agent.agentCoreRuntime.role.addToPolicy(
    new PolicyStatement({
    actions: [
    'bedrock:InvokeModel',
    'bedrock:InvokeModelWithResponseStream',
    ],
    // 可限定具体模型ARN
    resources: ['arn:aws:bedrock:*::foundation-model/*'],
    }),
    );
    }
    }

    为构建Bedrock AgentCore Runtime代理,项目添加了bundle目标:

    • 使用uv export导出Python依赖到requirements.txt
    • 使用uv pip install为目标平台(aarch64-manylinux2014)安装依赖

    同时添加docker目标:

    • 根据Dockerfile构建符合AgentCore运行时契约的镜像

    默认使用IAM认证,直接部署即可:

    import { MyProjectAgent } from ':my-scope/common-constructs';
    export class ExampleStack extends Stack {
    constructor(scope: Construct, id: string) {
    new MyProjectAgent(this, 'MyProjectAgent');
    }
    }

    授予调用权限示例:

    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.agentCoreRuntime.grantInvoke(lambdaFunction);
    }
    }

    以下展示如何配置Cognito认证:

    通过authorizerConfiguration属性配置JWT认证。示例配置Cognito用户池和客户端:

    import { MyProjectAgent } from ':my-scope/common-constructs';
    export class ExampleStack extends Stack {
    constructor(scope: Construct, id: string) {
    const userPool = new UserPool(this, 'UserPool');
    const client = userPool.addClient('Client', {
    authFlows: {
    userPassword: true,
    },
    });
    new MyProjectAgent(this, 'MyProjectAgent', {
    authorizerConfiguration: {
    customJWTAuthorizer: {
    discoveryUrl: `https://cognito-idp.${Stack.of(userPool).region}.amazonaws.com/${userPool.userPoolId}/.well-known/openid-configuration`,
    allowedClients: [client.userPoolClientId],
    },
    },
    });
    }
    }

    通过AWS OpenTelemetry发行版(ADOT)自动配置可观测性,在Dockerfile中配置自动检测。

    可在CloudWatch控制台的”GenAI Observability”菜单查看追踪数据。需启用事务搜索以显示数据。

    更多细节请参考AgentCore可观测性文档