跳转到内容

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 - 要添加 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 项目构建目标与配置

    部署Strands智能体时生成以下文件:

    • 文件夹packages/common/constructs/src
      • 文件夹app
        • 文件夹agents
          • 文件夹<project-name>
            • <project-name>.ts 部署智能体的CDK构造
            • Dockerfile CDK构造使用的透传Docker文件

    工具是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-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服务,可以使用connection生成器

    1. 安装 Nx Console VSCode Plugin 如果您尚未安装
    2. 在VSCode中打开Nx控制台
    3. 点击 Generate (UI) 在"Common Nx Commands"部分
    4. 搜索 @aws/nx-plugin - connection
    5. 填写必需参数
      • 点击 Generate

      关于连接设置的详细信息,请参考connection生成器指南

      对于其他MCP服务,请参考Strands文档

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

      生成器使用FastAPI为Strands智能体创建HTTP服务器。FastAPI提供了一个现代、高性能的Python API构建框架,具有自动API文档生成和类型验证功能。

      生成的服务器包含:

      • 配置CORS中间件的FastAPI应用
      • 错误处理中间件
      • OpenAPI模式生成
      • 健康检查端点(/ping
      • 智能体调用端点(/invocations

      使用Pydantic自定义调用输入和输出

      Section titled “使用Pydantic自定义调用输入和输出”

      智能体的调用端点使用Pydantic模型来定义和验证请求及响应模式。您可以在main.py中自定义这些模型以满足智能体需求。

      默认的InvokeInput模型接受提示词和会话ID。

      from pydantic import BaseModel
      class InvokeInput(BaseModel):
      prompt: str
      session_id: str

      您可以扩展此模型以包含智能体所需的任何额外字段。

      对于流式响应,生成器提供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: int

      FastAPI有一个关于原生支持的开放功能请求

      生成器包含对Bedrock AgentCore Python SDK的依赖,用于PingStatus常量。如果需要,可以直接使用BedrockAgentCoreApp替代FastAPI,但请注意会失去类型安全性。

      SDK功能详情请参阅文档

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

      Terminal window
      pnpm nx agent-serve your-project

      该命令使用uv run通过Bedrock AgentCore Python SDK运行智能体。

      如果您为 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/*',
      ],
      }),
      );
      }
      }

      生成器提供了一个 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);
      }
      }

      当您选择 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 UserPoolUserPoolClient

      为构建Bedrock AgentCore Runtime部署包,项目中添加了bundle目标:

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

      同时添加智能体专用的docker目标:

      通过Dockerfile中的自动配置,使用AWS分布式OpenTelemetry(ADOT)实现可观测性。

      在CloudWatch控制台选择”GenAI Observability”查看追踪数据,需启用事务搜索

      详细配置参考AgentCore可观测性文档

      要调用通过<your-agent-name>-serve目标本地运行的智能体,可以向本地智能体运行端口的/invocations发送简单的POST请求。例如,使用curl

      Terminal window
      curl -N -X POST http://localhost:8081/invocations \
      -d '{"prompt": "what is 3 + 5?", "session_id": "abcdefghijklmnopqrstuvwxyz0123456789"}' \
      -H "Content-Type: application/json"

      要调用部署到 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,
      });
      }
      }

      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)签名。

      Terminal window
      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'
      点击此处了解配置上述acurl命令的更多详情

      对于Cognito认证,在Authorization标头中传递Cognito访问令牌:

      Terminal window
      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命令获取访问令牌,例如:

      Terminal window
      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