Implementare e configurare l'agente Story

Task 1: Implementare lo 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 di un giocatore.

Implementazione Python

Per questo tutorial, implementeremo l’Agente in modo “senza stato”, dove ogni invocazione è una sessione indipendente che include l’intera cronologia dei messaggi. Per ambienti di produzione, considera l’integrazione con Bedrock AgentCore Memory. Per ulteriori informazioni, consulta la pagina using AgentCore Memory with Strands.

Implementazione dell’agente

Per implementare il nostro agente, aggiorna i seguenti file in packages/story/dungeon_adventure_story/agent:

main.py

+

import uuid

import uvicorn
from bedrock_agentcore.runtime.models import PingStatus
from pydantic import BaseModel

from .agent import get_agent
from .init import JsonStreamingResponse, app


class Action(BaseModel):
    role: str
    content: str


class InvokeInput(BaseModel):
    playerName: str
    genre: str
    actions: list[Action]


class StreamChunk(BaseModel):
    content: str


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 StreamChunk(content=content)
            elif event.get("event", {}).get("messageStop") is not None:
                yield StreamChunk(content="\n")


@app.post(
    "/invocations",
    response_class=JsonStreamingResponse,
    responses={200: JsonStreamingResponse.openapi_response(StreamChunk, "Stream of agent response chunks")},
)
async def invoke(input: InvokeInput) -> JsonStreamingResponse:
    """Entry point for agent invocation"""
    return JsonStreamingResponse(handle_invoke(input))


@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 pydantic import BaseModel

from .agent import get_agent
from .init import JsonStreamingResponse, app


class Action(BaseModel):
    role: str
    content: str


class InvokeInput(BaseModel):
    prompt: str
    session_id: str
    playerName: str
    genre: str
    actions: list[Action]


class StreamChunk(BaseModel):
    content: str


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)
            text = event.get("event", {}).get("contentBlockDelta", {}).get("delta", {}).get("text")
            if text is not None:
                yield StreamChunk(content=text)
            content = event.get("event", {}).get("contentBlockDelta", {}).get("delta", {}).get("text")
            if content is not None:
                yield StreamChunk(content=content)
            elif event.get("event", {}).get("messageStop") is not None:
                yield StreamChunk(content="\n")


@app.post(
    "/invocations",
    response_class=JsonStreamingResponse,
    responses={200: JsonStreamingResponse.openapi_response(StreamChunk, "Stream of agent response chunks")},
)
async def invoke(input: InvokeInput) -> JsonStreamingResponse:
    """Entry point for agent invocation"""
    return JsonStreamingResponse(handle_invoke(input))


@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

+

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClient
from strands import Agent


@contextmanager
def get_agent(player_name: str, genre: str, session_id: str):
    inventory_mcp_server = InventoryMcpServerClient.create(session_id=session_id)
    with (
        inventory_mcp_server,
    ):
        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=[*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


# Define a custom tool
@tool
def subtract(a: int, b: int) -> int:
    return a - b


@contextmanager
def get_agent(session_id: str):
def get_agent(player_name: str, genre: str, session_id: str):
    inventory_mcp_server = InventoryMcpServerClient.create(session_id=session_id)
    with (
        inventory_mcp_server,
    ):
        yield Agent(
            system_prompt="""
You are a mathematical wizard.
Use your tools for mathematical tasks.
Refer to tools as your 'spellbook'.
            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=[subtract, current_time, *inventory_mcp_server.list_tools_sync()],
            tools=[*inventory_mcp_server.list_tools_sync()],
        )

Poiché il generatore di connessioni ha già configurato il client MCP nel Modulo 1, queste modifiche si concentrano su:

• Rimuovere lo strumento di esempio e le importazioni non utilizzate,
• Aggiungere i parametri player_name e genre alla funzione get_agent, e
• Personalizzare il prompt di sistema per il nostro gioco di avventura nel dungeon.

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 invocandola con curl, oppure
• Chiamando l’API deployata usando curl con un token JWT.

Local

Avvia il server Agent locale eseguendo il seguente comando:

pnpm

RUNTIME_CONFIG_APP_ID=xxxx AWS_REGION=<region> pnpm nx agent-serve dungeon_adventure.story

yarn

RUNTIME_CONFIG_APP_ID=xxxx AWS_REGION=<region> yarn nx agent-serve dungeon_adventure.story

npm

RUNTIME_CONFIG_APP_ID=xxxx AWS_REGION=<region> npx nx agent-serve dungeon_adventure.story

bun

RUNTIME_CONFIG_APP_ID=xxxx AWS_REGION=<region> bunx nx agent-serve dungeon_adventure.story

Usa i valori di output CDK

Utilizza il valore di output del deploy CDK RuntimeConfigApplicationId per la variabile d’ambiente RUNTIME_CONFIG_APP_ID, e la tua regione AWS deployata per AWS_REGION.

Attenzione

Utilizza l’output InventoryMcpArn del deploy CDK sopra, e specifica la tua AWS_REGION target.

Una volta che il server Agent è attivo e funzionante (non vedrai alcun output), invocalo eseguendo il seguente comando:

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

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
curl -N -X POST "$AGENT_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

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 il testo iniziale della storia in streaming come JSON Lines:

{"content":"You are "}
{"content":"a new superhero "}
{"content":"in the bustling metropolis of Metro City..."}

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