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
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 strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)


@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),
)

app = create_strands_app(agui_agent, path="/invocations")

+-

import logging
import os
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 strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)

# 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),
)

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

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

린트 오류 수정

린트 오류가 발생하면 다음 명령어로 자동 수정할 수 있습니다.

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

에이전트 테스트

다음 방법 중 하나로 에이전트를 테스트할 수 있습니다:

• 에이전트 서버 로컬 인스턴스를 시작하고 생성된 agent-chat 타겟을 통해 대화하거나,
• JWT 토큰을 사용하여 배포된 API를 curl로 호출합니다.

Local

생성된 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을 생성하고(이는 차례로 인벤토리 MCP 서버를 로컬에서 부팅합니다), 그런 다음 http://localhost:8081/invocations에 대해 agent-chat-cli AG-UI REPL을 시작합니다. 첫 번째 메시지에서 에이전트에게 영웅의 이름과 장르를 알려주면(예: My name is Alice. Start my zombie adventure.) 스토리가 스트리밍됩니다.

Raw 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가 이전 메시지를 재생합니다.

Deployed

배포된 에이전트를 테스트하려면 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을 Bedrock AgentCore API 엔드포인트에 맞게 올바르게 포맷하려면 URL 인코딩이 필요합니다.

명령이 성공적으로 실행되면 스토리가 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 런타임에 첫 번째 Strands 에이전트를 구축하고 배포했습니다! 🎉🎉🎉