Juego de Mazmorra de IA Agéntica
Módulo 3: Implementación del Agente de Historia
Sección titulada «Módulo 3: Implementación del Agente de Historia»El Agente de Historia es un agente de Strands que, dado un Game
y una lista de Action
s como contexto, avanzará una historia. Configuraremos el agente para interactuar con nuestro Inventory MCP Server y gestionar los objetos disponibles del jugador.
Implementación del Agente
Sección titulada «Implementación del Agente»Implementemos nuestro agente. Actualiza los siguientes archivos en packages/story/dungeon_adventure_story/agent
:
import osfrom bedrock_agentcore.runtime import BedrockAgentCoreApp
from .agent import get_agent
app = BedrockAgentCoreApp()
PORT = int(os.environ.get("PORT", "8080"))
@app.entrypointasync 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 osfrom bedrock_agentcore.runtime import BedrockAgentCoreApp
from .agent import get_agent
app = BedrockAgentCoreApp()
PORT = int(os.environ.get("PORT", "8080"))
@app.entrypointasync 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)
import osfrom contextlib import contextmanager
import boto3from strands import Agent
from .agentcore_mcp_client import AgentCoreMCPClient
# Obtain the region and credentialsregion = os.environ["AWS_REGION"]boto_session = boto3.Session(region_name=region)credentials = boto_session.get_credentials()
@contextmanagerdef 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 osfrom contextlib import contextmanager
from strands import Agent, toolfrom strands_tools import current_timeimport boto3from strands import Agent
from .agentcore_mcp_client import AgentCoreMCPClient
# Define a custom tool@tooldef add(a: int, b: int) -> int: return a + b# Obtain the region and credentialsregion = os.environ["AWS_REGION"]boto_session = boto3.Session(region_name=region)credentials = boto_session.get_credentials()
@contextmanagerdef 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()], )
Esto configura:
- Extracción del jugador, género y acciones del payload del agente
- Construcción de un cliente que el Agente puede usar para invocar nuestro servidor MCP con Autenticación SigV4
- Construcción del agente con un prompt del sistema y las herramientas del servidor MCP
Despliegue y pruebas
Sección titulada «Despliegue y pruebas»Primero, compilemos el código base:
pnpm nx run-many --target build --all
yarn nx run-many --target build --all
npx nx run-many --target build --all
bunx nx run-many --target build --all
Ahora puedes desplegar la aplicación ejecutando:
pnpm nx deploy infra dungeon-adventure-infra-sandbox/*
yarn nx deploy infra dungeon-adventure-infra-sandbox/*
npx nx deploy infra dungeon-adventure-infra-sandbox/*
bunx nx deploy infra dungeon-adventure-infra-sandbox/*
Este despliegue tomará aproximadamente 2 minutos en completarse.
Una vez completado el despliegue, verás salidas similares a las siguientes (algunos valores han sido omitidos):
dungeon-adventure-infra-sandbox-Applicationdungeon-adventure-infra-sandbox-Application: deploying... [2/2]
✅ dungeon-adventure-infra-sandbox-Application
✨ Tiempo de despliegue: 354s
Outputs:dungeon-adventure-infra-sandbox-Application.ElectroDbTableTableNameXXX = dungeon-adventure-infra-sandbox-Application-ElectroDbTableXXX-YYYdungeon-adventure-infra-sandbox-Application.GameApiEndpointXXX = https://xxx.execute-api.region.amazonaws.com/prod/dungeon-adventure-infra-sandbox-Application.GameUIDistributionDomainNameXXX = xxx.cloudfront.netdungeon-adventure-infra-sandbox-Application.InventoryMcpArn = arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventureventoryMcpServerXXXX-YYYYdungeon-adventure-infra-sandbox-Application.StoryAgentArn = arn:aws:bedrock-agentcore:region:xxxxxxx:runtime/dungeonadventurecationStoryAgentXXXX-YYYYdungeon-adventure-infra-sandbox-Application.UserIdentityUserIdentityIdentityPoolIdXXX = region:xxxdungeon-adventure-infra-sandbox-Application.UserIdentityUserIdentityUserPoolIdXXX = region_xxx
Podemos probar nuestra API de dos formas:
- Iniciando una instancia local del servidor del Agente e invocándola con
curl
. - Llamando a la API desplegada usando curl con un token JWT.
Inicia el servidor local del Agente ejecutando:
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
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
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
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
Una vez que el servidor del Agente esté en ejecución (¡no verás ninguna salida!), invócalo con:
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"
Para probar el agente desplegado, necesitarás autenticarte con Cognito y obtener un token JWT. Primero configura tus variables de entorno:
# Establece tu User Pool ID y Client ID de Cognito desde los outputs de CDKexport POOL_ID="<UserPoolId from CDK outputs>"export CLIENT_ID="<UserPoolClientId from CDK outputs>"export REGION="<your-region>"
Crea un usuario de prueba y obtén un token de autenticación:
# Crear usuarioaws cognito-idp admin-create-user \ --user-pool-id $POOL_ID \ --username "testuser" \ --temporary-password "TempPass123!" \ --region $REGION \ --message-action SUPPRESS > /dev/null
# Establecer contraseña permanente (¡reemplaza con algo más seguro!)aws cognito-idp admin-set-user-password \ --user-pool-id $POOL_ID \ --username "testuser" \ --password "PermanentPass123!" \ --region $REGION \ --permanent > /dev/null
# Autenticar usuario y capturar tokenexport 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')
Ahora invoca el agente desplegado usando la URL de Bedrock AgentCore Runtime:
# Establece el ARN del Agente de Historia desde los outputs de CDKexport AGENT_ARN="<StoryAgentArn from CDK outputs>"
# Codifica el ARN en URLexport ENCODED_ARN=$(echo $AGENT_ARN | sed 's/:/%3A/g' | sed 's/\//%2F/g')
# Construye la URL de invocaciónexport MCP_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"
# Invoca el agentecurl -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"}'
Si el comando se ejecuta correctamente, verás eventos transmitidos similares a:
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}}}
...
¡Felicidades! Has construido y desplegado tu primer Agente de Strands en Bedrock AgentCore Runtime. 🎉🎉🎉