Jogo de Dungeons de IA Agêntica

Módulo 3: Implementação do Agente de História

Cuidado

Certifique-se que sua conta AWS habilitou acesso ao modelo padrão da Strands dentro do Bedrock para sua região alvo seguindo os passos do guia oficial. Na data desta escrita, o modelo é o Anthropic Claude Sonnet 4.

O Agente de História é um agente Strands que, dado um Game e uma lista de Actions como contexto, irá progredir uma narrativa. Configuraremos o agente para interagir com nosso Inventory MCP Server e gerenciar os itens disponíveis do jogador.

Nota

Para simplificar este tutorial, implementamos o Agente de forma “stateless” onde cada invocação é uma sessão independente incluindo todo o histórico. Para produção, considere integrar com Bedrock AgentCore Memory. Mais documentação sobre uso com Strands está disponível aqui.

Implementação do Agente

Vamos implementar nosso agente. Atualize os seguintes arquivos em packages/story/dungeon_adventure_story/agent:

main.py

+

import os
from bedrock_agentcore.runtime import BedrockAgentCoreApp

from .agent import get_agent

app = BedrockAgentCoreApp()

PORT = int(os.environ.get("PORT", "8080"))

@app.entrypoint
async def invoke(payload, context):
    """Handler for agent invocation"""
    player_name = payload.get("playerName")
    genre = payload.get("genre")
    actions = payload.get("actions")

    messages = [{"role": "user", "content": [{"text": "Continue or create a new story..."}]}]
    for action in actions:
        messages.append({"role": action["role"], "content": [{"text": action["content"]}]})

    with get_agent(player_name, genre, session_id=context.session_id) as agent:
        stream = agent.stream_async(messages)
        async for event in stream:
            print(event)
            yield (event)


if __name__ == "__main__":
    app.run(port=PORT)

+-

import os
from bedrock_agentcore.runtime import BedrockAgentCoreApp

from .agent import get_agent

app = BedrockAgentCoreApp()

PORT = int(os.environ.get("PORT", "8080"))

@app.entrypoint
async def invoke(payload, context):
    """Handler for agent invocation"""
    prompt = payload.get(
        "prompt", "No prompt found in input, please guide the user "
        "to create a json payload with prompt key"
    )
    player_name = payload.get("playerName")
    genre = payload.get("genre")
    actions = payload.get("actions")

    with get_agent(session_id=context.session_id) as agent:
        stream = agent.stream_async(prompt)
    messages = [{"role": "user", "content": [{"text": "Continue or create a new story..."}]}]
    for action in actions:
        messages.append({"role": action["role"], "content": [{"text": action["content"]}]})

    with get_agent(player_name, genre, session_id=context.session_id) as agent:
        stream = agent.stream_async(messages)
        async for event in stream:
            print(event)
            yield (event)


if __name__ == "__main__":
    app.run()
    app.run(port=PORT)

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

Esta configuração implementa:

• Extração do jogador, gênero e ações do payload do agente
• Construção de um cliente para o Agente invocar o MCP Server com autenticação SigV4
• Criação do agente com prompt de sistema e ferramentas do MCP Server

Implantação e Testes

Primeiro, vamos construir o código:

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

Dica

Se encontrar erros de lint, execute este comando para correção automática:

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

Implante a aplicação com:

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

Esta implantação levará aproximadamente 2 minutos.

Após conclusão, você verá saídas similares a estas (valores redigidos):

dungeon-adventure-infra-sandbox-Application
dungeon-adventure-infra-sandbox-Application: deploying... [2/2]

 ✅  dungeon-adventure-infra-sandbox-Application

✨  Tempo de implantação: 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

Podemos testar nossa API de duas formas:

• Iniciando uma instância local do servidor Agent e invocando com curl
• Chamando a API implantada usando curl com token JWT

Local

Inicie o servidor local com:

pnpm

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

yarn

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

npm

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

bun

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

Cuidado

Use o output InventoryMcpArn do CDK deploy acima, e especifique sua AWS_REGION alvo.

Com o servidor rodando (sem output visível), invoque com:

curl -N -X POST http://127.0.0.1:9999/invocations \
  -d '{"genre":"superhero", "actions":[], "playerName":"UnnamedHero"}' \
  -H "Content-Type: application/json" \
  -H "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id: abcdefghijklmnopqrstuvwxyz-123456789"

Implantado

Para testar o agente implantado, autentique via Cognito para obter token JWT. Primeiro configure variáveis:

export POOL_ID="<UserPoolId dos outputs do CDK>"
export CLIENT_ID="<UserPoolClientId dos outputs do CDK>"
export REGION="<sua-regiao>"

Crie usuário teste e obtenha token:

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

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')

Cuidado

Use o output UserIdentityUserIdentityUserPoolIdXXX para POOL_ID e confirme o Client ID correto.

Invoque o agente implantado:

export AGENT_ARN="<StoryAgentArn dos outputs do CDK>"
export ENCODED_ARN=$(echo $AGENT_ARN | sed 's/:/%3A/g' | sed 's/\//%2F/g')
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"}'

Nota

A codificação URL é necessária para formatar corretamente a ARN no endpoint da API.

Se executado com sucesso, você verá eventos sendo transmitidos como:

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

...

Parabéns! Você implementou e implantou seu primeiro Agente Strands no Bedrock AgentCore Runtime! 🎉🎉🎉