Story エージェントの実装と設定

タスク1: ストーリーエージェントの実装

ストーリーエージェントはStrandsのエージェントで、GameとActionのリストをコンテキストとして物語を進行します。インベントリMCPサーバーと連携してプレイヤーの所持アイテムを管理するよう設定します。

エージェント実装

エージェントを実装するには、packages/story/dungeon_adventure_story/agent配下のファイルを更新します：

main.py
agent.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)

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()],
        )

この実装では以下を構成します：

エージェントペイロードからプレイヤー情報、ジャンル、アクションを抽出、
SigV4認証でMCPサーバーを呼び出すためのクライアントの構築、および
システムプロンプトとMCPサーバーツールを備えたエージェントの作成。

タスク2: デプロイとテスト

コードのビルド

コードをビルドするには：

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

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

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

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

リンターエラーが発生した場合、以下のコマンドで自動修正できます。

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

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

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

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

アプリケーションのデプロイ

アプリケーションをデプロイするには、以下のコマンドを実行します：

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

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

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

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

デプロイは約2分で完了します。

デプロイが完了すると、以下のような出力が表示されます（一部の値は編集済み）：

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

APIのテスト

APIのテスト方法は2通りあります：

エージェントサーバーをローカルで起動しcurlで呼び出す、または
デプロイ済みAPIをJWTトークン付きのcurlで呼び出す。

ローカル
デプロイ済み

以下のコマンドでローカルエージェントサーバーを起動します：

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

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

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

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

エージェントサーバー起動後（出力は表示されません）、以下のコマンドで呼び出します：

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

デプロイ済みエージェントのテストには、Cognito認証によるJWTトークン取得が必要です。まず環境変数を設定します：

# CDK出力からCognito User Pool IDとClient IDを設定
export POOL_ID="<CDK出力のUserPoolId>"
export CLIENT_ID="<CDK出力のUserPoolClientId>"
export REGION="<リージョン>"

テストユーザーを作成し、認証トークンを取得します：

# ユーザー作成
aws cognito-idp admin-create-user \
  --user-pool-id $POOL_ID \
  --username "testuser" \
  --temporary-password "TempPass123!" \
  --region $REGION \
  --message-action SUPPRESS > /dev/null

# パスワード設定（本番環境ではより安全なパスワードを使用してください）
aws cognito-idp admin-set-user-password \
  --user-pool-id $POOL_ID \
  --username "testuser" \
  --password "PermanentPass123!" \
  --region $REGION \
  --permanent > /dev/null

# 認証とIDトークン取得
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')

Bedrock AgentCoreランタイムURLを使用してデプロイ済みエージェントを呼び出します：

# CDK出力からStory Agent ARNを設定
export AGENT_ARN="<CDK出力のStoryAgentArn>"

# ARNをURLエンコード
export ENCODED_ARN=$(echo $AGENT_ARN | sed 's/:/%3A/g' | sed 's/\//%2F/g')

# 呼び出しURLを構築
export MCP_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"

# エージェント呼び出し
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"}'

コマンドが正常に実行されると、以下のようなストリーミングイベントが表示されます：

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

...

おめでとうございます。Bedrock AgentCore Runtime上に初めてのStrandsエージェントを構築・デプロイできました！ 🎉🎉🎉