Story 에이전트 구현 및 구성

작업 1: 스토리 에이전트 구현

스토리 에이전트는 Strands 에이전트로, Game과 컨텍스트용 Action 목록을 입력받아 스토리를 진행합니다. 이 에이전트를 인벤토리 MCP 서버와 연동하도록 구성하여 플레이어의 사용 가능한 아이템을 관리할 수 있게 합니다.

참고

본 튜토리얼에서는 각 호출이 전체 메시지 기록을 포함하는 독립 세션인 “상태 비저장” 방식으로 에이전트를 구현합니다. 프로덕션 환경에서는 Bedrock AgentCore 메모리 통합을 고려하세요. 자세한 내용은 Strands와 AgentCore 메모리 사용 페이지를 참조하세요.

에이전트 구현

에이전트를 구현하려면 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()],
        )

이 구성은 다음을 설정합니다:

• 에이전트 페이로드에서 플레이어, 장르, 액션 추출,
• SigV4 인증으로 MCP 서버를 호출할 수 있는 클라이언트 구성, 그리고
• 시스템 프롬프트와 MCP 서버 도구를 포함한 에이전트 구성.

작업 2: 배포 및 테스트

코드 빌드

코드를 빌드하려면:

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

팁

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

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

애플리케이션 배포

애플리케이션을 배포하려면 다음 명령어를 실행하세요:

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.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를 테스트할 수 있습니다:

• 에이전트 서버 로컬 인스턴스를 시작하고 curl로 호출하거나,
• JWT 토큰을 사용하여 배포된 API를 curl로 호출합니다.

Local

다음 명령어로 로컬 에이전트 서버를 시작하세요:

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

주의

위 명령어에서 CDK 배포 출력값 InventoryMcpArn을 사용하고 대상 AWS_REGION을 지정하세요.

에이전트 서버가 실행되면(콘솔에 출력되지 않음) 다음 명령어로 호출하세요:

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

Deployed

배포된 에이전트를 테스트하려면 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')

주의

POOL_ID에는 CDK 배포 출력의 UserIdentityUserIdentityUserPoolIdXXX를 사용하고 CDK 출력에서 해당 Client ID를 확인하세요.

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

참고

ARN을 Bedrock AgentCore API 엔드포인트에 맞게 올바르게 포맷하려면 URL 인코딩이 필요합니다.

명령이 성공적으로 실행되면 다음과 유사한 스트리밍 이벤트가 표시됩니다:

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 런타임에 첫 번째 Strands 에이전트를 구축하고 배포했습니다! 🎉🎉🎉