Triển khai và cấu hình Story agent

Nhiệm vụ 1: Triển khai Story Agent

Story Agent là một agent Strands mà khi được cung cấp một Game và danh sách các Action làm ngữ cảnh, sẽ phát triển câu chuyện. Chúng ta sẽ cấu hình agent để tương tác với Inventory MCP Server của chúng ta nhằm quản lý các vật phẩm khả dụng của người chơi.

Ghi chú

Đối với hướng dẫn này, chúng ta sẽ triển khai Agent theo cách “không trạng thái”, trong đó mỗi lần gọi là một phiên độc lập bao gồm toàn bộ lịch sử tin nhắn. Đối với môi trường sản xuất, hãy xem xét tích hợp với Bedrock AgentCore Memory. Để biết thêm thông tin, tham khảo trang sử dụng AgentCore Memory với Strands.

Triển khai Agent

Để triển khai agent của chúng ta, cập nhật các file sau trong packages/story/dungeon_adventure_story/agent:

main.py

+

import uuid

import uvicorn
from bedrock_agentcore.runtime.models import PingStatus
from fastapi.responses import PlainTextResponse, StreamingResponse
from pydantic import BaseModel

from .agent import get_agent
from .init import app


class Action(BaseModel):
    role: str
    content: str


class InvokeInput(BaseModel):
    playerName: str
    genre: str
    actions: list[Action]


async def handle_invoke(input: InvokeInput):
    """Streaming handler for agent invocation"""
    messages = [{"role": "user", "content": [{"text": "Continue or create a new story..."}]}]
    for action in input.actions:
        messages.append({"role": action.role, "content": [{"text": action.content}]})

    with get_agent(input.playerName, input.genre, session_id=str(uuid.uuid4())) as agent:
        stream = agent.stream_async(messages)
        async for event in stream:
            print(event)
            content = event.get("event", {}).get("contentBlockDelta", {}).get("delta", {}).get("text")
            if content is not None:
                yield content
            elif event.get("event", {}).get("messageStop") is not None:
                yield "\n"


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


@app.get("/ping")
def ping() -> str:
    # TODO: if running an async task, return PingStatus.HEALTHY_BUSY
    return PingStatus.HEALTHY


if __name__ == "__main__":
    uvicorn.run("dungeon_adventure_story.agent.main:app", port=8080)

+-

import uuid

import uvicorn
from bedrock_agentcore.runtime.models import PingStatus
from fastapi.responses import PlainTextResponse, StreamingResponse
from pydantic import BaseModel

from .agent import get_agent
from .init import app


class Action(BaseModel):
    role: str
    content: str


class InvokeInput(BaseModel):
    prompt: str
    session_id: str
    playerName: str
    genre: str
    actions: list[Action]


async def handle_invoke(input: InvokeInput):
    """Streaming handler for agent invocation"""
    with get_agent(session_id=input.session_id) as agent:
        stream = agent.stream_async(input.prompt)
    messages = [{"role": "user", "content": [{"text": "Continue or create a new story..."}]}]
    for action in input.actions:
        messages.append({"role": action.role, "content": [{"text": action.content}]})

    with get_agent(input.playerName, input.genre, session_id=str(uuid.uuid4())) as agent:
        stream = agent.stream_async(messages)
        async for event in stream:
            print(event)
            content = event.get("event", {}).get("contentBlockDelta", {}).get("delta", {}).get("text")
            if content is not None:
                yield content
            elif event.get("event", {}).get("messageStop") is not None:
                yield "\n"


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


@app.get("/ping")
def ping() -> str:
    # TODO: if running an async task, return PingStatus.HEALTHY_BUSY
    return PingStatus.HEALTHY


if __name__ == "__main__":
    uvicorn.run("dungeon_adventure_story.agent.main:app", port=8080)

agent.py

+

import os
from contextlib import contextmanager

import boto3
from strands import Agent

from .agentcore_mcp_client import AgentCoreMCPClient

# Obtain the region and credentials
region = os.environ["AWS_REGION"]
boto_session = boto3.Session(region_name=region)
credentials = boto_session.get_credentials()


@contextmanager
def get_agent(player_name: str, genre: str, session_id: str):
    mcp_client = AgentCoreMCPClient.with_iam_auth(
        agent_runtime_arn=os.environ["INVENTORY_MCP_ARN"],
        credentials=credentials,
        region=region,
        session_id=session_id,
    )
    with mcp_client:
        yield Agent(
            system_prompt=f"""
You are running a text adventure game in the genre <genre>{genre}</genre> for player <player>{player_name}</player>.
Construct a scenario and give the player decisions to make.
Use the tools to manage the player's inventory as items are obtained or lost.
When adding, removing or updating items in the inventory, always list items to check the current state,
and be careful to match item names exactly. Item names in the inventory must be Title Case.
Ensure you specify a suitable emoji when adding items if available.
When starting a game, populate the inventory with a few initial items. Items should be a key part of the narrative.
Keep responses under 100 words.
""",
            tools=[*mcp_client.list_tools_sync()],
        )

+-

import os
from contextlib import contextmanager

from strands import Agent, tool
from strands_tools import current_time
import boto3
from strands import Agent

from .agentcore_mcp_client import AgentCoreMCPClient

# Define a custom tool
@tool
def add(a: int, b: int) -> int:
    return a + b
# Obtain the region and 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):
    yield Agent(
        system_prompt="""
You are an addition wizard.
Use the 'add' tool for addition tasks.
Refer to tools as your 'spellbook'.
""",
        tools=[add, current_time],
def get_agent(player_name: str, genre: str, session_id: str):
    mcp_client = AgentCoreMCPClient.with_iam_auth(
        agent_runtime_arn=os.environ["INVENTORY_MCP_ARN"],
        credentials=credentials,
        region=region,
        session_id=session_id,
    )
    with mcp_client:
        yield Agent(
            system_prompt=f"""
You are running a text adventure game in the genre <genre>{genre}</genre> for player <player>{player_name}</player>.
Construct a scenario and give the player decisions to make.
Use the tools to manage the player's inventory as items are obtained or lost.
When adding, removing or updating items in the inventory, always list items to check the current state,
and be careful to match item names exactly. Item names in the inventory must be Title Case.
Ensure you specify a suitable emoji when adding items if available.
When starting a game, populate the inventory with a few initial items. Items should be a key part of the narrative.
Keep responses under 100 words.
""",
            tools=[*mcp_client.list_tools_sync()],
        )

Điều này sẽ cấu hình những thứ sau:

• Trích xuất player, genre và actions từ agent payload,
• Xây dựng một client mà Agent có thể sử dụng để gọi MCP server của chúng ta với SigV4 Authentication, và
• Xây dựng agent với system prompt và các công cụ của MCP server.

Nhiệm vụ 2: Triển khai và kiểm thử

Build code

Để build code:

pnpm

pnpm nx run-many --target build --all

yarn

yarn nx run-many --target build --all

npm

npx nx run-many --target build --all

bun

bunx nx run-many --target build --all

Mẹo

Nếu bạn gặp bất kỳ lỗi lint nào, hãy chạy lệnh sau để tự động sửa chúng.

pnpm

pnpm nx run-many --target lint --configuration=fix --all

yarn

yarn nx run-many --target lint --configuration=fix --all

npm

npx nx run-many --target lint --configuration=fix --all

bun

bunx nx run-many --target lint --configuration=fix --all

Triển khai ứng dụng của bạn

Để triển khai ứng dụng của bạn, chạy lệnh sau:

pnpm

pnpm nx deploy infra dungeon-adventure-infra-sandbox/*

yarn

yarn nx deploy infra dungeon-adventure-infra-sandbox/*

npm

npx nx deploy infra dungeon-adventure-infra-sandbox/*

bun

bunx nx deploy infra dungeon-adventure-infra-sandbox/*

Quá trình triển khai này sẽ mất khoảng 2 phút để hoàn thành.

Khi quá trình triển khai hoàn tất, bạn sẽ thấy các output tương tự như sau (một số giá trị đã được ẩn):

dungeon-adventure-infra-sandbox-Application
dungeon-adventure-infra-sandbox-Application: deploying... [2/2]

 ✅  dungeon-adventure-infra-sandbox-Application

✨  Deployment time: 354s

Outputs:
dungeon-adventure-infra-sandbox-Application.ElectroDbTableTableNameXXX = dungeon-adventure-infra-sandbox-Application-ElectroDbTableXXX-YYY
dungeon-adventure-infra-sandbox-Application.GameApiEndpointXXX = https://xxx.execute-api.region.amazonaws.com/prod/
dungeon-adventure-infra-sandbox-Application.GameUIDistributionDomainNameXXX = xxx.cloudfront.net
dungeon-adventure-infra-sandbox-Application.InventoryMcpArn = arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventureventoryMcpServerXXXX-YYYY
dungeon-adventure-infra-sandbox-Application.StoryAgentArn = arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventurecationStoryAgentXXXX-YYYY
dungeon-adventure-infra-sandbox-Application.UserIdentityUserIdentityIdentityPoolIdXXX = region:xxx
dungeon-adventure-infra-sandbox-Application.UserIdentityUserIdentityUserPoolIdXXX = region_xxx

Kiểm thử Agent của bạn

Bạn có thể kiểm thử Agent của mình bằng một trong hai cách:

• Khởi động một instance cục bộ của Agent server và gọi nó bằng curl, hoặc
• Gọi API đã triển khai bằng curl với JWT token.

Local

Khởi động Agent server cục bộ của bạn bằng cách chạy lệnh sau:

pnpm

INVENTORY_MCP_ARN=arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventureventoryMcpServerXXXX-YYYY AWS_REGION=<region> pnpm nx run dungeon_adventure.story:agent-serve

yarn

INVENTORY_MCP_ARN=arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventureventoryMcpServerXXXX-YYYY AWS_REGION=<region> yarn nx run dungeon_adventure.story:agent-serve

npm

INVENTORY_MCP_ARN=arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventureventoryMcpServerXXXX-YYYY AWS_REGION=<region> npx nx run dungeon_adventure.story:agent-serve

bun

INVENTORY_MCP_ARN=arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventureventoryMcpServerXXXX-YYYY AWS_REGION=<region> bunx nx run dungeon_adventure.story:agent-serve

Chú ý

Sử dụng output InventoryMcpArn từ CDK deploy ở trên và chỉ định AWS_REGION mục tiêu của bạn.

Khi Agent server đã chạy (bạn sẽ không thấy bất kỳ output nào), gọi nó bằng cách chạy lệnh sau:

curl -N -X POST http://127.0.0.1:8081/invocations \
  -d '{"genre":"superhero", "actions":[], "playerName":"UnnamedHero"}' \
  -H "Content-Type: application/json"

Deployed

Để kiểm thử agent đã triển khai, bạn cần xác thực với Cognito và lấy JWT token. Đầu tiên, thiết lập các biến môi trường của bạn:

# Đặt Cognito User Pool ID và Client ID từ các output của CDK
export POOL_ID="<UserPoolId từ CDK outputs>"
export CLIENT_ID="<UserPoolClientId từ CDK outputs>"
export REGION="<your-region>"

Tạo một test user và lấy authentication token:

# Tạo User
aws cognito-idp admin-create-user \
  --user-pool-id $POOL_ID \
  --username "testuser" \
  --temporary-password "TempPass123!" \
  --region $REGION \
  --message-action SUPPRESS > /dev/null

# Đặt Permanent Password (thay thế bằng một mật khẩu an toàn hơn!)
aws cognito-idp admin-set-user-password \
  --user-pool-id $POOL_ID \
  --username "testuser" \
  --password "PermanentPass123!" \
  --region $REGION \
  --permanent > /dev/null

# Xác thực User và lấy ID Token
export BEARER_TOKEN=$(aws cognito-idp initiate-auth \
  --client-id "$CLIENT_ID" \
  --auth-flow USER_PASSWORD_AUTH \
  --auth-parameters USERNAME='testuser',PASSWORD='PermanentPass123!' \
  --region $REGION | jq -r '.AuthenticationResult.IdToken')

Chú ý

Sử dụng output UserIdentityUserIdentityUserPoolIdXXX cho POOL_ID và đảm bảo bạn có Client ID tương ứng từ các output triển khai CDK của bạn.

Gọi agent đã triển khai bằng URL runtime của Bedrock AgentCore:

# Đặt Story Agent ARN từ CDK outputs
export AGENT_ARN="<StoryAgentArn từ CDK outputs>"

# URL-encode ARN
export ENCODED_ARN=$(echo $AGENT_ARN | sed 's/:/%3A/g' | sed 's/\//%2F/g')

# Xây dựng invocation URL
export MCP_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"

# Gọi agent
curl -N -X POST "$MCP_URL" \
  -H "authorization: Bearer $BEARER_TOKEN" \
  -H "Content-Type: application/json" \
  -H "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id: abcdefghijklmnopqrstuvwxyz-123456789" \
  -d '{"genre":"superhero", "actions":[], "playerName":"UnnamedHero"}'

Ghi chú

URL encoding là cần thiết để định dạng đúng ARN cho endpoint API của Bedrock AgentCore.

Nếu lệnh chạy thành công, bạn sẽ thấy các events được streaming tương tự như:

data: {"init_event_loop": true}

data: {"start": true}

data: {"start_event_loop": true}

data: {"event": {"messageStart": {"role": "assistant"}}}

data: {"event": {"contentBlockDelta": {"delta": {"text": "Welcome"}, "contentBlockIndex": 0}}}

...

Chúc mừng. Bạn đã xây dựng và triển khai Strands Agent đầu tiên của mình trên Bedrock AgentCore Runtime! 🎉🎉🎉