Story 에이전트 구현 및 구성

작업 1: Story 에이전트 구현

Story 에이전트는 모듈 1에서 --protocol=AG-UI로 생성된 Strands 에이전트로, UI가 CopilotKit을 통해 Agent-User Interaction protocol을 통해 스트리밍할 수 있습니다. 이 에이전트는 Inventory MCP Server를 사용하여 플레이어의 아이템을 관리하고, 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-dev` (`LOCAL_DEV=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("LOCAL_DEV") == "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-dev` (`LOCAL_DEV=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("LOCAL_DEV") == "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 InventoryMcpServerClientStrands, log_model_errors
from strands import Agent


@contextmanager
def get_agent():
    inventory_mcp_server = InventoryMcpServerClientStrands.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()],
            hooks=[log_model_errors],
        )

+-

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClientStrands, log_model_errors
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 = InventoryMcpServerClientStrands.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()],
            hooks=[log_model_errors],
        )

변경 사항은 다음과 같습니다:

• main.py는 배포 시 thread_id당 S3SessionManager를 생성하는 session_manager_provider를 추가하고, agent-dev에서 실행할 때(LOCAL_DEV=true) /tmp/strands-sessions 아래의 온디스크 FileSessionManager로 폴백합니다. 배포 시 S3 버킷은 Game API의 queryActions가 읽는 것과 동일한 버킷이므로 브라우저가 재방문 시 대화 기록을 재구성할 수 있습니다. 로컬에서는 에이전트가 세션을 디스크에 저장하고 배포 없이 로컬 MCP 서버와 통신합니다.
• agent.py는 샘플 subtract 도구를 제거하고 시스템 프롬프트를 던전 마스터 프롬프트로 교체합니다. 이 프롬프트는 첫 번째 사용자 메시지에서 플레이어의 이름과 장르를 명시하도록 유도하며, Inventory MCP Server의 도구를 사용합니다.

작업 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

에이전트와 대화하기

생성된 agent-chat 타겟은 에이전트에 대해 대화형 REPL을 엽니다. 이것은 독립적으로 실행되며 로컬에서 실행 중인 에이전트에 연결되므로, 먼저 한 터미널에서 에이전트의 로컬 서버를 시작하세요:

pnpm

pnpm nx agent-dev story

yarn

yarn nx agent-dev story

npm

npx nx agent-dev story

bun

bunx nx agent-dev story

그런 다음 두 번째 터미널에서 채팅을 시작하세요:

pnpm

pnpm nx agent-chat story

yarn

yarn nx agent-chat story

npm

npx nx agent-chat story

bun

bunx nx agent-chat story

첫 번째 메시지는 에이전트에게 영웅의 이름과 장르를 알려야 합니다(예: My name is Alice. Start my zombie adventure.). 그러면 스토리가 스트리밍되어 돌아옵니다.

AWS 자격 증명 필요

에이전트가 로컬에서 실행되지만 LLM을 위해 Amazon Bedrock을 호출합니다. Strands 기본 모델에 액세스할 수 있는 AWS 자격 증명이 환경에 필요합니다. 자세한 내용은 Strands 모델 공급자 문서를 참조하세요.

축하합니다. 첫 번째 Strands 에이전트를 로컬에서 빌드하고 테스트했습니다! 🎉🎉🎉