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

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

ストーリーエージェントは、モジュール1で--protocol=AG-UIを指定して生成されたStrandsエージェントです。これにより、UIはCopilotKitを介してAgent-User Interactionプロトコルでストリーミングできます。インベントリMCPサーバーを使用してプレイヤーのアイテムを管理し、Strandsの組み込みS3SessionManagerを使用して、モジュール2でプロビジョニングしたセッションバケットに会話履歴を永続化します。

エージェント実装

packages/story/dungeon_adventure_story/agent配下のファイルを更新します：

main.py

+

import logging
import os
import uuid
from functools import cache
from typing import Any, cast

from ag_ui.core import RunAgentInput
from ag_ui_strands import StrandsAgent, StrandsAgentConfig, create_strands_app
from aws_lambda_powertools.utilities import parameters
from dungeon_adventure_agent_connection import session_id_context
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)

SESSION_ID_HEADER = "x-amzn-bedrock-agentcore-runtime-session-id"


@cache
def _resolve_sessions_bucket() -> str:
    """Read the conversation-history bucket name from runtime config.

    Resolved lazily (and memoised) so `fastapi dev` can import this module
    before ``RUNTIME_CONFIG_APP_ID`` is in the environment.
    """
    application = os.environ.get("RUNTIME_CONFIG_APP_ID")
    if not application:
        raise RuntimeError("RUNTIME_CONFIG_APP_ID is not set — cannot resolve the StorySessions bucket.")
    provider = parameters.AppConfigProvider(environment="default", application=application)
    buckets = cast(dict[str, Any], provider.get("buckets", transform="json"))
    return buckets["StorySessions"]["bucketName"]


def _session_manager_provider(input_data: RunAgentInput) -> SessionManager:
    """Create a session manager keyed by the AG-UI thread_id.

    - In AgentCore (and `agent-serve`), persist to the shared ``StorySessions``
      S3 bucket — the same bucket the Game API reads from to rebuild
      conversation history on revisit.
    - In `agent-serve-local` (`SERVE_LOCAL=true`), persist to a temp
      directory so the agent can run fully offline against the local MCP
      server without any AWS calls.
    """
    session_id = input_data.thread_id or "default"
    if os.environ.get("SERVE_LOCAL") == "true":
        return FileSessionManager(session_id=session_id, storage_dir="/tmp/strands-sessions")
    return S3SessionManager(session_id=session_id, bucket=_resolve_sessions_bucket())


# The template Agent is cloned per thread_id by ``StrandsAgent`` — we plug in
# a ``session_manager_provider`` so each thread gets its own session manager
# and conversation history is replayed on subsequent turns and survives agent
# restarts.
_agent_ctx = get_agent()
_agent = _agent_ctx.__enter__()

agui_agent = StrandsAgent(
    agent=_agent,
    name="StoryAgent",
    description="A Strands Agent exposed via the AG-UI protocol.",
    config=StrandsAgentConfig(session_manager_provider=_session_manager_provider),
)


class _SessionIdMiddleware(BaseHTTPMiddleware):
    """Bind the session ID for this request so downstream MCP / A2A clients forward it on outbound calls."""

    async def dispatch(self, request: Request, call_next):
        session_id = request.headers.get(SESSION_ID_HEADER) or str(uuid.uuid4())
        with session_id_context(session_id):
            return await call_next(request)


app = create_strands_app(agui_agent, path="/invocations")
app.add_middleware(_SessionIdMiddleware)

+-

import logging
import os
import uuid
from functools import cache
from typing import Any, cast

from ag_ui_strands import StrandsAgent, create_strands_app
from ag_ui.core import RunAgentInput
from ag_ui_strands import StrandsAgent, StrandsAgentConfig, create_strands_app
from aws_lambda_powertools.utilities import parameters
from dungeon_adventure_agent_connection import session_id_context
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)

SESSION_ID_HEADER = "x-amzn-bedrock-agentcore-runtime-session-id"

# Create AG-UI agent wrapper

@cache
def _resolve_sessions_bucket() -> str:
    """Read the conversation-history bucket name from runtime config.

    Resolved lazily (and memoised) so `fastapi dev` can import this module
    before ``RUNTIME_CONFIG_APP_ID`` is in the environment.
    """
    application = os.environ.get("RUNTIME_CONFIG_APP_ID")
    if not application:
        raise RuntimeError("RUNTIME_CONFIG_APP_ID is not set — cannot resolve the StorySessions bucket.")
    provider = parameters.AppConfigProvider(environment="default", application=application)
    buckets = cast(dict[str, Any], provider.get("buckets", transform="json"))
    return buckets["StorySessions"]["bucketName"]


def _session_manager_provider(input_data: RunAgentInput) -> SessionManager:
    """Create a session manager keyed by the AG-UI thread_id.

    - In AgentCore (and `agent-serve`), persist to the shared ``StorySessions``
      S3 bucket — the same bucket the Game API reads from to rebuild
      conversation history on revisit.
    - In `agent-serve-local` (`SERVE_LOCAL=true`), persist to a temp
      directory so the agent can run fully offline against the local MCP
      server without any AWS calls.
    """
    session_id = input_data.thread_id or "default"
    if os.environ.get("SERVE_LOCAL") == "true":
        return FileSessionManager(session_id=session_id, storage_dir="/tmp/strands-sessions")
    return S3SessionManager(session_id=session_id, bucket=_resolve_sessions_bucket())


# The template Agent is cloned per thread_id by ``StrandsAgent`` — we plug in
# a ``session_manager_provider`` so each thread gets its own session manager
# and conversation history is replayed on subsequent turns and survives agent
# restarts.
_agent_ctx = get_agent()
_agent = _agent_ctx.__enter__()

agui_agent = StrandsAgent(
    agent=_agent,
    name="StoryAgent",
    description="A Strands Agent exposed via the AG-UI protocol.",
    config=StrandsAgentConfig(session_manager_provider=_session_manager_provider),
)


class _SessionIdMiddleware(BaseHTTPMiddleware):
    """Bind the session ID for this request so downstream MCP / A2A clients forward it on outbound calls."""

    async def dispatch(self, request: Request, call_next):
        session_id = request.headers.get(SESSION_ID_HEADER) or str(uuid.uuid4())
        with session_id_context(session_id):
            return await call_next(request)


# Create FastAPI app with AG-UI endpoint and health check
app = create_strands_app(agui_agent, path="/invocations")
app.add_middleware(_SessionIdMiddleware)

agent.py

+

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClient
from strands import Agent


@contextmanager
def get_agent():
    inventory_mcp_server = InventoryMcpServerClient.create()
    with (
        inventory_mcp_server,
    ):
        yield Agent(
            system_prompt="""
You are running a text adventure game for a lone adventurer. When a new story
begins, the first user message will tell you the player's name and the genre
(one of 'medieval', 'zombie', 'superhero'). Greet the player by name, set the
scene in the chosen genre, and populate their inventory with a few starting
items. On subsequent turns, advance the story in response to the player's
actions and keep item state in sync with the narrative.
Use the tools to manage the player's inventory as items are obtained or lost.
Item names in the inventory must be Title Case — match them exactly.
Only use list-inventory-items if you are unsure of the exact item name before modifying it.
IMPORTANT: Only call ONE tool per response turn. Never batch multiple tool calls in a single turn.
Ensure you specify a suitable emoji when adding items if available.
Items should be a key part of the narrative.
Keep responses under 100 words.
""",
            tools=[*inventory_mcp_server.list_tools_sync()],
        )

+-

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClient
from strands import Agent, tool
from strands_tools import current_time
from strands import Agent


@tool
def subtract(a: int, b: int) -> int:
    return a - b


@contextmanager
def get_agent():
    inventory_mcp_server = InventoryMcpServerClient.create()
    with (
        inventory_mcp_server,
    ):
        yield Agent(
            name="StoryAgent",
            description="StoryAgent Strands Agent",
            system_prompt="""
You are a mathematical wizard.
Use your tools for mathematical tasks.
Refer to tools as your 'spellbook'.
You are running a text adventure game for a lone adventurer. When a new story
begins, the first user message will tell you the player's name and the genre
(one of 'medieval', 'zombie', 'superhero'). Greet the player by name, set the
scene in the chosen genre, and populate their inventory with a few starting
items. On subsequent turns, advance the story in response to the player's
actions and keep item state in sync with the narrative.
Use the tools to manage the player's inventory as items are obtained or lost.
Item names in the inventory must be Title Case — match them exactly.
Only use list-inventory-items if you are unsure of the exact item name before modifying it.
IMPORTANT: Only call ONE tool per response turn. Never batch multiple tool calls in a single turn.
Ensure you specify a suitable emoji when adding items if available.
Items should be a key part of the narrative.
Keep responses under 100 words.
""",
            tools=[subtract, current_time, *inventory_mcp_server.list_tools_sync()],
            tools=[*inventory_mcp_server.list_tools_sync()],
        )

変更内容は以下の通りです：

• main.pyは、デプロイ時にthread_idごとにS3SessionManagerを作成し、agent-serve-local（SERVE_LOCAL=true）で実行時には/tmp/strands-sessions配下のディスク上のFileSessionManagerにフォールバックするsession_manager_providerを追加します。デプロイ時、S3バケットはGame APIのqueryActionsが読み取るものと同じであるため、ブラウザは再訪問時にトランスクリプトを再構築できます。ローカルでは、エージェントはAWS呼び出しなしでローカルMCPサーバーに対して完全にオフラインで実行されます。
• agent.pyは、サンプルのsubtractツールを削除し、システムプロンプトをダンジョンマスター用のものに置き換えます。このプロンプトは、最初のユーザーメッセージでプレイヤーの名前とジャンルを述べるよう促し、インベントリMCPサーバーのツールを使用します。

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

コードのビルド

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

pnpm

pnpm build

yarn

yarn build

npm

npm run build

bun

bun build

Lintエラーの修正

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

pnpm

pnpm lint

yarn

yarn lint

npm

npm run lint

bun

bun lint

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

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

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/*"

デプロイは約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.RuntimeConfigApplicationId = xxxx
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

エージェントのテスト

エージェントのテスト方法は2通りあります：

• エージェントサーバーをローカルで起動し、生成されたagent-chatターゲットを介してチャットする、または
• デプロイ済みAPIをJWTトークン付きのcurlで呼び出す。

ローカル

生成されたagent-chatターゲットを使用して、ローカルで提供されるAG-UIエージェントのコピーに対する対話型REPLを開きます：

pnpm

pnpm nx agent-chat story

yarn

yarn nx agent-chat story

npm

npx nx agent-chat story

bun

bunx nx agent-chat story

agent-chatにはdependsOn: ['agent-serve-local']が設定されているため、エージェントサーバーを別途起動する必要はありません。Nxがagent-serve-localを起動し（これによりInventory MCPサーバーもローカルで起動されます）、その後agent-chat-cliを使用してhttp://localhost:8081/invocationsに対するAG-UI REPLを起動します。最初のメッセージでは、エージェントにヒーローの名前とジャンルを伝えてください（例：My name is Alice. Start my zombie adventure.）。ストーリーがストリーミングで返されます。

生のcurl

ローカルサーバーに対して生のHTTP呼び出しを希望する場合、AG-UIのRunAgentInputの形式は以下の通りです：

curl -N -X POST http://127.0.0.1:8081/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "threadId": "Alice-zombie00000000000000000000000",
    "runId": "run-1",
    "messages": [{ "id": "m1", "role": "user", "content": "My name is Alice. Start my zombie adventure." }],
    "tools": [], "context": [], "state": {}, "forwardedProps": {}
  }'

後続の呼び出しで同じthreadIdを再利用すると、同じストーリーが続きます。S3SessionManagerが以前のメッセージを再生します。

デプロイ済み

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

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

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

# ユーザー作成を簡素化するためMFAを無効化
aws cognito-idp set-user-pool-mfa-config \
  --mfa-configuration OFF \
  --user-pool-id $POOL_ID

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

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

# 認証とアクセストークン取得
export BEARER_TOKEN=$(aws cognito-idp initiate-auth \
  --client-id "$CLIENT_ID" \
  --auth-flow USER_PASSWORD_AUTH \
  --auth-parameters USERNAME='test',PASSWORD='PermanentPass123-' \
  --region $REGION \
  --query "AuthenticationResult.AccessToken" \
  --output text)

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 AGENT_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"

# AG-UI RunAgentInputペイロードでエージェントを呼び出し
curl -N -X POST "$AGENT_URL" \
  -H "authorization: Bearer $BEARER_TOKEN" \
  -H "Content-Type: application/json" \
  -H "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id: Alice-zombie00000000000000000000000" \
  -d '{
    "threadId": "Alice-zombie00000000000000000000000",
    "runId": "run-1",
    "messages": [{ "id": "m1", "role": "user", "content": "My name is Alice. Start my zombie adventure." }],
    "tools": [], "context": [], "state": {}, "forwardedProps": {}
  }'

URLエンコード

ARNのURLエンコードは、Bedrock AgentCore APIエンドポイントの正しいフォーマットに必要です。

コマンドが正常に実行されると、AG-UIイベントストリーム（Server-Sent Events）としてストーリーがストリーミングされ始めます。RUN_STARTED / TEXT_MESSAGE_START / TEXT_MESSAGE_CONTENTチャンクが実際のストーリートークンをラップします：

data: {"type":"RUN_STARTED","threadId":"Alice-zombie00000000000000000000000",...}
data: {"type":"TEXT_MESSAGE_START","messageId":"...","role":"assistant"}
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":"Greetings, Alice. "}
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":"The moans grow louder beyond the barricaded door..."}

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