Implementare e configurare l'agente Story

Task 1: Implementare lo Story Agent

Lo Story Agent è un agente Strands generato con --protocol=AG-UI nel Modulo 1, in modo che l’interfaccia utente possa ricevere lo streaming da esso tramite il protocollo Agent-User Interaction via CopilotKit. Utilizza il server MCP dell’inventario per gestire gli oggetti del giocatore e l’S3SessionManager integrato di Strands per persistere la cronologia delle conversazioni nel bucket delle sessioni che abbiamo fornito nel Modulo 2.

Implementazione dell’agente

Aggiorna i seguenti file in packages/story/dungeon_adventure_story/agent:

main.py

+

import logging
import os
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 strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)


@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-serve-local` (`SERVE_LOCAL=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("SERVE_LOCAL") == "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),
)

app = create_strands_app(agui_agent, path="/invocations")

+-

import logging
import os
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 strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)

# 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-serve-local` (`SERVE_LOCAL=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("SERVE_LOCAL") == "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),
)

# Create FastAPI app with AG-UI endpoint and health check
app = create_strands_app(agui_agent, path="/invocations")

agent.py

+

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClient
from strands import Agent


@contextmanager
def get_agent():
    inventory_mcp_server = InventoryMcpServerClient.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.
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.
Items should be a key part of the narrative.
Keep responses under 100 words.
""",
            tools=[*inventory_mcp_server.list_tools_sync()],
        )

+-

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClient
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 = InventoryMcpServerClient.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.
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.
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()],
        )

Le modifiche sono:

• main.py aggiunge un session_manager_provider che crea un S3SessionManager per ogni thread_id quando deployato, e ricade su un FileSessionManager su disco sotto /tmp/strands-sessions quando eseguito con agent-serve-local (SERVE_LOCAL=true). In deployment, il bucket S3 è lo stesso da cui legge il queryActions della Game API, così il browser può ricostruire le trascrizioni alla rivisitazione; localmente l’agente funziona completamente offline contro il server MCP locale senza chiamate AWS.
• agent.py rimuove lo strumento di esempio subtract e sostituisce il prompt di sistema con uno da dungeon-master che invita il primo messaggio dell’utente a dichiarare il nome del giocatore e il genere, e utilizza gli strumenti del server MCP dell’inventario.

Task 2: Deployment e testing

Compilare il codice

Per compilare il codice:

pnpm

pnpm build

yarn

yarn build

npm

npm run build

bun

bun build

Correzione errori di lint

Se incontri errori di linting, esegui il seguente comando per correggerli automaticamente.

pnpm

pnpm lint

yarn

yarn lint

npm

npm run lint

bun

bun lint

Effettuare il deploy dell’applicazione

Per effettuare il deploy dell’applicazione, esegui il seguente comando:

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

Il deployment richiederà circa 2 minuti per completarsi.

Una volta completato il deployment, vedrai output simili ai seguenti (alcuni valori sono stati oscurati):

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.RuntimeConfigApplicationId = xxxx
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

Testare il tuo Agent

Puoi testare il tuo Agent in due modi:

• Avviando un’istanza locale del server Agent e chattando con esso tramite il target generato agent-chat, oppure
• Chiamando l’API deployata usando curl con un token JWT.

Local

Apri un REPL interattivo contro una copia dell’agente AG-UI servita localmente utilizzando il target generato agent-chat:

pnpm

pnpm nx agent-chat story

yarn

yarn nx agent-chat story

npm

npx nx agent-chat story

bun

bunx nx agent-chat story

agent-chat ha dependsOn: ['agent-serve-local'], quindi non è necessario avviare il server dell’agente separatamente — Nx avvierà agent-serve-local (che a sua volta avvia il server MCP dell’inventario localmente), quindi lancerà il REPL AG-UI agent-chat-cli contro http://localhost:8081/invocations. Il tuo primo messaggio dovrebbe dire all’agente il nome del tuo eroe e il genere (per esempio: My name is Alice. Start my zombie adventure.) e la storia verrà trasmessa in streaming.

curl grezzo

Se preferisci una chiamata HTTP grezza contro il server locale, la forma RunAgentInput di AG-UI è:

curl -N -X POST http://127.0.0.1:8081/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "threadId": "Alice-zombie00000000000000000000000",
    "runId": "run-1",
    "messages": [{ "id": "m1", "role": "user", "content": "My name is Alice. Start my zombie adventure." }],
    "tools": [], "context": [], "state": {}, "forwardedProps": {}
  }'

Riutilizzare lo stesso threadId nelle chiamate successive continuerà la stessa storia — l’S3SessionManager riproduce i messaggi precedenti.

Deployed

Per testare l’agente deployato, dovrai autenticarti con Cognito e ottenere un token JWT. Prima, imposta le tue variabili d’ambiente:

# Imposta User Pool ID e Client ID dagli output CDK
export POOL_ID="<UserPoolId dagli output CDK>"
export CLIENT_ID="<UserPoolClientId dagli output CDK>"
export REGION="<tua-regione>"

Crea un utente test e ottieni un token di autenticazione:

# Disabilita MFA per semplificare la creazione dell'utente
aws cognito-idp set-user-pool-mfa-config \
  --mfa-configuration OFF \
  --user-pool-id $POOL_ID

# Crea Utente
aws cognito-idp admin-create-user \
  --user-pool-id $POOL_ID \
  --username "test" \
  --temporary-password "TempPass123-" \
  --region $REGION \
  --message-action SUPPRESS > /dev/null

# Imposta Password Permanente (sostituisci con qualcosa di più sicuro!)
aws cognito-idp admin-set-user-password \
  --user-pool-id $POOL_ID \
  --username "test" \
  --password "PermanentPass123-" \
  --region $REGION \
  --permanent > /dev/null

# Autentica l'Utente e cattura l'Access Token
export BEARER_TOKEN=$(aws cognito-idp initiate-auth \
  --client-id "$CLIENT_ID" \
  --auth-flow USER_PASSWORD_AUTH \
  --auth-parameters USERNAME='test',PASSWORD='PermanentPass123-' \
  --region $REGION \
  --query "AuthenticationResult.AccessToken" \
  --output text)

Invoca l’agente deployato usando l’URL Bedrock AgentCore runtime:

# Imposta Story Agent ARN dagli output CDK
export AGENT_ARN="<StoryAgentArn dagli output CDK>"

# Codifica URL dell'ARN
export ENCODED_ARN=$(echo $AGENT_ARN | sed 's/:/%3A/g' | sed 's/\//%2F/g')

# Costruisci l'URL di invocazione
export AGENT_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"

# Invoca l'agente con un payload AG-UI RunAgentInput
curl -N -X POST "$AGENT_URL" \
  -H "authorization: Bearer $BEARER_TOKEN" \
  -H "Content-Type: application/json" \
  -H "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id: Alice-zombie00000000000000000000000" \
  -d '{
    "threadId": "Alice-zombie00000000000000000000000",
    "runId": "run-1",
    "messages": [{ "id": "m1", "role": "user", "content": "My name is Alice. Start my zombie adventure." }],
    "tools": [], "context": [], "state": {}, "forwardedProps": {}
  }'

Codifica URL

La codifica URL è necessaria per formattare correttamente l’ARN per l’endpoint API di Bedrock AgentCore.

Se il comando viene eseguito con successo, dovresti iniziare a vedere la storia trasmessa in streaming come un flusso di eventi AG-UI (Server-Sent Events) — i chunk RUN_STARTED / TEXT_MESSAGE_START / TEXT_MESSAGE_CONTENT avvolgono i token effettivi della storia:

data: {"type":"RUN_STARTED","threadId":"Alice-zombie00000000000000000000000",...}
data: {"type":"TEXT_MESSAGE_START","messageId":"...","role":"assistant"}
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":"Greetings, Alice. "}
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":"The moans grow louder beyond the barricaded door..."}

Complimenti. Hai creato e deployato il tuo primo Strands Agent su Bedrock AgentCore Runtime! 🎉🎉🎉