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.

Nota

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 fastapi.responses import PlainTextResponse, StreamingResponse
from pydantic import BaseModel

from .agent import get_agent
from .init import app


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


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


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


@app.post("/invocations", openapi_extra={"x-streaming": True}, response_class=PlainTextResponse)
async def invoke(input: InvokeInput) -> str:
    """Entry point for agent invocation"""
    return StreamingResponse(handle_invoke(input), media_type="text/event-stream")


@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 fastapi.responses import PlainTextResponse, StreamingResponse
from pydantic import BaseModel

from .agent import get_agent
from .init import app


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


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


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


@app.post("/invocations", openapi_extra={"x-streaming": True}, response_class=PlainTextResponse)
async def invoke(input: InvokeInput) -> str:
    """Entry point for agent invocation"""
    return StreamingResponse(handle_invoke(input), media_type="text/event-stream")


@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

+

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

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, e
• La costruzione dell’agente con un prompt di sistema e gli strumenti del server MCP.

Task 2: Deployment e testing

Compilare il codice

Per compilare il codice:

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

Consiglio

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

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

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.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 la tua API

Puoi testare la tua API 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

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

yarn

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

npm

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

bun

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

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:

# Crea Utente
aws cognito-idp admin-create-user \
  --user-pool-id $POOL_ID \
  --username "testuser" \
  --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 "testuser" \
  --password "PermanentPass123!" \
  --region $REGION \
  --permanent > /dev/null

# Autentica l'Utente e cattura l'ID Token
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')

Attenzione

Utilizza l’output UserIdentityUserIdentityUserPoolIdXXX per POOL_ID e assicurati di avere il Client ID corrispondente dagli output del deployment CDK.

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 MCP_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"

# Invoca l'agente
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

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

Se il comando viene eseguito con 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! 🎉🎉🎉