Gioco di Dungeon con IA Agentiva
Modulo 3: Implementazione dello Story Agent
Sezione intitolata “Modulo 3: Implementazione dello Story Agent”Lo Story Agent è un agente Strands che, dato un Game
e una lista di Action
come contesto, farà progredire una storia. Configureremo l’agente per interagire con il nostro server MCP dell’inventario per gestire gli oggetti disponibili del giocatore.
Implementazione dell’Agente
Sezione intitolata “Implementazione dell’Agente”Implementiamo il nostro agente. Aggiorna i seguenti file in 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()], )
Questa configurazione implementa:
- L’estrazione del giocatore, genere e azioni dal payload dell’agente
- La costruzione di un client che l’Agente può utilizzare per invocare il nostro server MCP con Autenticazione SigV4
- La costruzione dell’agente con un prompt di sistema e gli strumenti del server MCP
Deployment e testing
Sezione intitolata “Deployment e testing”Prima, compiliamo il codice:
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
Ora puoi effettuare il deploy dell’applicazione con:
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/*
Il deployment richiederà circa 2 minuti.
Al completamento, dovresti vedere output simili a questi (alcuni valori sono stati oscurati):
dungeon-adventure-infra-sandbox-Applicationdungeon-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-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
Possiamo testare l’API in due modi:
- Avviando un’istanza locale del server Agent e invocandola con
curl
- Chiamando l’API deployata usando curl con un token JWT
Avvia il server Agent locale con:
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
Quando il server è attivo (nessun output visibile!), invocalo 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"
Per testare l’agente deployato, autenticati con Cognito per ottenere un token JWT. Imposta prima le variabili d’ambiente:
# Imposta User Pool ID e Client ID dagli output CDKexport POOL_ID="<UserPoolId dagli output CDK>"export CLIENT_ID="<UserPoolClientId dagli output CDK>"export REGION="<tua-regione>"
Crea un utente test e ottieni il token:
# Crea utenteaws cognito-idp admin-create-user \ --user-pool-id $POOL_ID \ --username "testuser" \ --temporary-password "TempPass123!" \ --region $REGION \ --message-action SUPPRESS > /dev/null
# Imposta password permanenteaws cognito-idp admin-set-user-password \ --user-pool-id $POOL_ID \ --username "testuser" \ --password "PermanentPass123!" \ --region $REGION \ --permanent > /dev/null
# Autentica e cattura 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')
Invoca l’agente deployato usando l’URL Bedrock AgentCore runtime:
# Imposta Story Agent ARN dagli output CDKexport AGENT_ARN="<StoryAgentArn dagli output CDK>"
# Codifica URL dell'ARNexport ENCODED_ARN=$(echo $AGENT_ARN | sed 's/:/%3A/g' | sed 's/\//%2F/g')
# Costruisci URL di invocazioneexport MCP_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"
# Invoca l'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"}'
Se il comando ha successo, vedrai eventi in streaming simili 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}}}
...
Complimenti. Hai creato e deployato il tuo primo Strands Agent su Bedrock AgentCore Runtime! 🎉🎉🎉