Bỏ qua để đến nội dung
@aws/nx-plugin
Blog

Python Strands Agent

Tạo một Python Strands Agent để xây dựng các AI agent với các công cụ, và tùy chọn triển khai nó lên Amazon Bedrock AgentCore Runtime.

Strands là gì?

Phần tiêu đề “Strands là gì?”

Strands là một framework Python nhẹ, sẵn sàng cho production để xây dựng các AI agent. Các tính năng chính bao gồm:

  • Nhẹ và có thể tùy chỉnh: Vòng lặp agent đơn giản không cản trở bạn
  • Sẵn sàng cho production: Khả năng quan sát đầy đủ, theo dõi và các tùy chọn triển khai cho quy mô lớn
  • Không phụ thuộc vào model và provider: Hỗ trợ nhiều model khác nhau từ các provider khác nhau
  • Công cụ được cộng đồng đóng góp: Bộ công cụ mạnh mẽ do cộng đồng đóng góp
  • Hỗ trợ đa agent: Các kỹ thuật nâng cao như nhóm agent và agent tự động
  • Chế độ tương tác linh hoạt: Hỗ trợ hội thoại, streaming và non-streaming

Cách sử dụng

Phần tiêu đề “Cách sử dụng”

Tạo một Strands Agent

Phần tiêu đề “Tạo một Strands Agent”

Bạn có thể tạo một Python Strands Agent theo hai cách:

  1. Install the Nx Console VSCode Plugin if you haven't already
  2. Open the Nx Console in VSCode
  3. Click Generate (UI) in the "Common Nx Commands" section
  4. Search for @aws/nx-plugin - py#strands-agent
  5. Fill in the required parameters
    • Click Generate

    Tùy chọn

    Phần tiêu đề “Tùy chọn”
    Parameter Type Default Description
    project Required 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 Inherit The preferred IaC provider. By default this is inherited from your initial selection.

    Kết quả của Generator

    Phần tiêu đề “Kết quả của Generator”

    Generator sẽ thêm các file sau vào Python project hiện có của bạn:

    • Thư mụcyour-project/
      • Thư mụcyour_module/
        • Thư mụcagent/ (hoặc tên tùy chỉnh nếu được chỉ định)
          • __init__.py Khởi tạo Python package
          • agent.py Định nghĩa agent chính với các công cụ mẫu
          • main.py Entry point cho Bedrock AgentCore Runtime
          • agentcore_mcp_client.py Client factory hữu ích để gọi các MCP server cũng được host trên Bedrock AgentCore runtime
          • Dockerfile Entry point để host agent của bạn (bị loại trừ khi computeType được đặt thành None)
      • pyproject.toml Được cập nhật với các dependencies của Strands
      • project.json Được cập nhật với các target serve agent

    Hạ tầng

    Phần tiêu đề “Hạ tầng”

    Vì generator này cung cấp infrastructure as code dựa trên iacProvider bạn đã chọn, nó sẽ tạo một dự án trong packages/common bao gồm các CDK constructs hoặc Terraform modules liên quan.

    Dự án infrastructure as code chung được cấu trúc như sau:

    • Thư mụcpackages/common/constructs
      • Thư mụcsrc
        • Thư mụcapp/ Constructs cho infrastructure cụ thể của một dự án/generator
          • …
        • Thư mụccore/ Constructs chung được tái sử dụng bởi các constructs trong app
          • …
        • index.ts Entry point xuất các constructs từ app
      • project.json Các build targets và cấu hình của dự án

    Để triển khai Strands Agent của bạn, các file sau được tạo:

    • Thư mụcpackages/common/constructs/src
      • Thư mụcapp
        • Thư mụcagents
          • Thư mục<project-name>
            • <project-name>.ts CDK construct để triển khai agent của bạn
            • Dockerfile File docker passthrough được sử dụng bởi CDK construct
      • Thư mụccore
        • Thư mụcagent-core
          • runtime.ts CDK construct chung để triển khai lên Bedrock AgentCore Runtime

    Làm việc với Strands Agent của bạn

    Phần tiêu đề “Làm việc với Strands Agent của bạn”

    Thêm công cụ

    Phần tiêu đề “Thêm công cụ”

    Các công cụ là các hàm mà AI agent có thể gọi để thực hiện các hành động. Framework Strands sử dụng phương pháp dựa trên decorator đơn giản để định nghĩa các công cụ.

    Bạn có thể thêm các công cụ mới trong file agent.py:

    from strands import Agent, tool
    

    @tool
    def calculate_sum(numbers: list[int]) -> int:
        """Calculate the sum of a list of numbers"""
        return sum(numbers)
    

    @tool
    def get_weather(city: str) -> str:
        """Get weather information for a city"""
        # Your weather API integration here
        return f"Weather in {city}: Sunny, 25°C"
    

    # Add tools to your agent
    agent = Agent(
        system_prompt="You are a helpful assistant with access to various tools.",
        tools=[calculate_sum, get_weather],
    )

    Framework Strands tự động xử lý:

    • Xác thực kiểu dựa trên type hints của hàm
    • Tạo JSON schema cho tool calling
    • Xử lý lỗi và định dạng response

    Sử dụng các công cụ có sẵn

    Phần tiêu đề “Sử dụng các công cụ có sẵn”

    Strands cung cấp một bộ sưu tập các công cụ có sẵn thông qua package strands-tools:

    from strands_tools import current_time, http_request, file_read
    

    agent = Agent(
        system_prompt="You are a helpful assistant.",
        tools=[current_time, http_request, file_read],
    )

    Cấu hình Model

    Phần tiêu đề “Cấu hình Model”

    Mặc định, các Strands agent sử dụng Claude 4 Sonnet, nhưng bạn có thể tùy chỉnh model provider. Xem tài liệu Strands về model providers để biết các tùy chọn cấu hình:

    from strands import Agent
    from strands.models import BedrockModel
    

    # Create a BedrockModel
    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)

    Sử dụng MCP Servers

    Phần tiêu đề “Sử dụng MCP Servers”

    Bạn có thể thêm các công cụ từ MCP servers vào Strands agent của bạn.

    Để sử dụng các MCP Server mà bạn đã tạo bằng generator py#mcp-server hoặc ts#mcp-server (hoặc các server khác được host trên Bedrock AgentCore Runtime), một client factory được tạo cho bạn trong agentcore_mcp_client.py.

    Bạn có thể cập nhật phương thức get_agent trong agent.py để tạo các MCP client và thêm công cụ. Ví dụ sau đây cho thấy cách thực hiện điều này với xác thực IAM (SigV4):

    agent.py
    import os
    from contextlib import contextmanager
    

    import boto3
    from strands import Agent
    

    from .agentcore_mcp_client import AgentCoreMCPClient
    

    # Obtain the region an credentials
    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],
            )

    Với ví dụ xác thực IAM ở trên, chúng ta cần cấu hình hai thứ trong hạ tầng của mình. Thứ nhất, chúng ta cần thêm biến môi trường mà agent của chúng ta đang sử dụng cho ARN của AgentCore Runtime của MCP server, và thứ hai chúng ta cần cấp quyền cho agent của chúng ta để gọi MCP server. Điều này có thể đạt được như sau:

    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);
      }
    }

    Thêm nữa

    Phần tiêu đề “Thêm nữa”

    Để có hướng dẫn chuyên sâu hơn về viết Strands agent, hãy tham khảo tài liệu Strands.

    FastAPI Server

    Phần tiêu đề “FastAPI Server”

    Generator sử dụng FastAPI để tạo HTTP server cho Strands Agent của bạn. FastAPI cung cấp một web framework hiện đại, nhanh để xây dựng API với Python, với tài liệu API tự động và xác thực kiểu.

    Server được tạo bao gồm:

    • Thiết lập ứng dụng FastAPI với CORS middleware
    • Middleware xử lý lỗi
    • Tạo OpenAPI schema
    • Health check endpoint (/ping)
    • Agent invocation endpoint (/invocations)

    Tùy chỉnh Invoke Inputs và Outputs với Pydantic

    Phần tiêu đề “Tùy chỉnh Invoke Inputs và Outputs với Pydantic”

    Invocation endpoint của agent sử dụng các model Pydantic để định nghĩa và xác thực các schema request và response. Bạn có thể tùy chỉnh các model này trong main.py để phù hợp với yêu cầu của agent.

    Định nghĩa Input Models

    Phần tiêu đề “Định nghĩa Input Models”

    Model InvokeInput mặc định chấp nhận một prompt và session ID.

    from pydantic import BaseModel
    

    class InvokeInput(BaseModel):
        prompt: str
        session_id: str

    Bạn có thể mở rộng model này để bao gồm bất kỳ trường bổ sung nào mà agent của bạn cần.

    Định nghĩa Output Models

    Phần tiêu đề “Định nghĩa Output Models”

    Đối với streaming responses, type annotation trả về trên endpoint của bạn tương ứng với kiểu của mỗi giá trị được yield bởi generator function của bạn. Mặc định, agent yield các string chứa văn bản response của agent khi nó stream trở lại từ Strands:

    @app.post("/invocations", openapi_extra={"x-streaming": True})
    async def invoke(input: InvokeInput) -> str:
        """Entry point for agent invocation"""
        return StreamingResponse(handle_invoke(input), media_type="text/event-stream")

    Bạn có thể định nghĩa một Pydantic model để yield dữ liệu có cấu trúc thay thế:

    from pydantic import BaseModel
    

    class StreamChunk(BaseModel):
        content: str
        timestamp: str
        token_count: int
    

    @app.post("/invocations", openapi_extra={"x-streaming": True})
    async def invoke(input: InvokeInput) -> StreamChunk:
        return StreamingResponse(handle_invoke(input), media_type="application/json")

    Bedrock AgentCore Python SDK

    Phần tiêu đề “Bedrock AgentCore Python SDK”

    Generator bao gồm một dependency trên Bedrock AgentCore Python SDK cho các hằng số PingStatus. Nếu muốn, việc sử dụng BedrockAgentCoreApp thay vì FastAPI là khá đơn giản, tuy nhiên lưu ý rằng type-safety sẽ bị mất.

    Bạn có thể tìm thêm chi tiết về khả năng của SDK trong tài liệu tại đây.

    Chạy Strands Agent của bạn

    Phần tiêu đề “Chạy Strands Agent của bạn”

    Phát triển Local

    Phần tiêu đề “Phát triển Local”

    Generator cấu hình một target có tên <your-agent-name>-serve, khởi động Strands Agent của bạn locally để phát triển và kiểm thử.

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

    Lệnh này sử dụng uv run để thực thi Strands Agent của bạn bằng Bedrock AgentCore Python SDK.

    Triển khai Strands Agent của bạn lên Bedrock AgentCore Runtime

    Phần tiêu đề “Triển khai Strands Agent của bạn lên Bedrock AgentCore Runtime”

    Infrastructure as Code

    Phần tiêu đề “Infrastructure as Code”

    Nếu bạn chọn BedrockAgentCoreRuntime cho computeType, hạ tầng CDK hoặc Terraform liên quan được tạo mà bạn có thể sử dụng để triển khai Strands Agent của bạn lên Amazon Bedrock AgentCore Runtime.

    Một CDK construct được tạo cho bạn, được đặt tên dựa trên name bạn đã chọn khi chạy generator, hoặc <ProjectName>Agent theo mặc định.

    Bạn có thể sử dụng CDK construct này trong một ứng dụng 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.role.addToPolicy(
          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/*'],
          }),
        );
      }
    }

    Bundle và Docker Targets

    Phần tiêu đề “Bundle và Docker Targets”

    Để build Strands Agent của bạn cho Bedrock AgentCore Runtime, một target bundle được thêm vào project của bạn, thực hiện:

    • Export các Python dependencies của bạn vào file requirements.txt bằng uv export
    • Cài đặt dependencies cho nền tảng đích (aarch64-manylinux2014) bằng uv pip install

    Một target docker cụ thể cho Strands Agent của bạn cũng được thêm vào, thực hiện:

    Xác thực

    Phần tiêu đề “Xác thực”

    IAM

    Phần tiêu đề “IAM”

    Mặc định, Strands Agent của bạn sẽ được bảo mật bằng xác thực IAM, chỉ cần triển khai nó mà không có bất kỳ tham số nào:

    import { MyProjectAgent } from ':my-scope/common-constructs';
    

    export class ExampleStack extends Stack {
      constructor(scope: Construct, id: string) {
        new MyProjectAgent(this, 'MyProjectAgent');
      }
    }

    Bạn có thể cấp quyền truy cập để gọi agent của bạn trên Bedrock AgentCore Runtime bằng phương thức grantInvoke, ví dụ:

    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 JWT Authentication

    Phần tiêu đề “Cognito JWT Authentication”

    Dưới đây minh họa cách cấu hình xác thực Cognito cho agent của bạn.

    Để cấu hình xác thực JWT, bạn có thể truyền thuộc tính authorizerConfiguration cho agent construct của bạn. Đây là một ví dụ cấu hình một Cognito user pool và client để bảo mật agent:

    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],
            },
          },
        });
      }
    }

    Khả năng quan sát

    Phần tiêu đề “Khả năng quan sát”

    Agent của bạn được tự động cấu hình với khả năng quan sát bằng cách sử dụng AWS Distro for Open Telemetry (ADOT), bằng cách cấu hình auto-instrumentation trong Dockerfile của bạn.

    Bạn có thể tìm thấy các trace trong AWS Console CloudWatch, bằng cách chọn “GenAI Observability” trong menu. Lưu ý rằng để các trace được điền, bạn sẽ cần bật Transaction Search.

    Để biết thêm chi tiết, hãy tham khảo tài liệu AgentCore về observability.