Implémenter et configurer l'agent Story

Tâche 1 : Implémenter le Story Agent

Le Story Agent est un agent Strands qui, étant donné un Game et une liste d’Actions comme contexte, fait progresser une histoire. Nous configurerons l’agent pour interagir avec notre Inventory MCP Server afin de gérer les objets disponibles d’un joueur.

Implémentation Python

Pour ce tutoriel, nous allons implémenter l’agent de manière « sans état », où chaque invocation est une session indépendante incluant l’historique complet des messages. Pour la production, envisagez d’intégrer Bedrock AgentCore Memory. Pour plus d’informations, consultez la page utilisation de AgentCore Memory avec Strands.

Implémentation de l’agent

Pour implémenter notre agent, mettez à jour les fichiers suivants dans 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()],
        )

Étant donné que le générateur de connexion a déjà configuré le client MCP dans le Module 1, ces modifications se concentrent sur :

• La suppression de l’outil d’exemple et des imports inutilisés,
• L’ajout des paramètres player_name et genre à la fonction get_agent, et
• La personnalisation du prompt système pour notre jeu d’aventure de donjon.

Tâche 2 : Déploiement et tests

Compiler le code

Pour compiler le code :

pnpm

pnpm build

yarn

yarn build

npm

npm run build

bun

bun build

Correction des erreurs de lint

Si vous rencontrez des erreurs de lint, exécutez la commande suivante pour les corriger automatiquement.

pnpm

pnpm lint

yarn

yarn lint

npm

npm run lint

bun

bun lint

Déployer votre application

Pour déployer votre application, exécutez la commande suivante :

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

Ce déploiement prendra environ 2 minutes.

Une fois le déploiement terminé, vous verrez des sorties similaires à ceci (certaines valeurs ont été masquées) :

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

Tester votre Agent

Vous pouvez tester votre Agent via :

• Le démarrage d’une instance locale du serveur Agent et son invocation à l’aide de curl, ou
• L’appel de l’API déployée en utilisant curl avec un token JWT.

Local

Démarrez votre serveur Agent local en exécutant la commande suivante :

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

Utiliser les valeurs de sortie CDK

Utilisez la valeur de sortie RuntimeConfigApplicationId du déploiement CDK pour la variable d’environnement RUNTIME_CONFIG_APP_ID, et votre région AWS déployée pour AWS_REGION.

Attention

Utilisez la sortie InventoryMcpArn du déploiement CDK ci-dessus, et spécifiez votre AWS_REGION cible.

Une fois le serveur Agent démarré et en cours d’exécution (vous ne verrez aucune sortie), invoquez-le en exécutant la commande suivante :

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

Deployed

Pour tester l’agent déployé, vous devrez vous authentifier avec Cognito et obtenir un token JWT. Tout d’abord, configurez vos variables d’environnement :

# Définissez votre User Pool ID et Client ID Cognito depuis les sorties CDK
export POOL_ID="<UserPoolId depuis les sorties CDK>"
export CLIENT_ID="<UserPoolClientId depuis les sorties CDK>"
export REGION="<votre-région>"

Créez un utilisateur test et obtenez un token d’authentification :

# Désactiver MFA pour simplifier la création d'utilisateur
aws cognito-idp set-user-pool-mfa-config \
  --mfa-configuration OFF \
  --user-pool-id $POOL_ID

# Créer l'utilisateur
aws cognito-idp admin-create-user \
  --user-pool-id $POOL_ID \
  --username "test" \
  --temporary-password "TempPass123-" \
  --region $REGION \
  --message-action SUPPRESS > /dev/null

# Définir un mot de passe permanent (remplacez par quelque chose de plus sécurisé !)
aws cognito-idp admin-set-user-password \
  --user-pool-id $POOL_ID \
  --username "test" \
  --password "PermanentPass123-" \
  --region $REGION \
  --permanent > /dev/null

# Authentifier l'utilisateur et capturer le token d'accès
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)

Invoquez l’agent déployé en utilisant l’URL du runtime Bedrock AgentCore :

# Définir l'ARN du Story Agent depuis les sorties CDK
export AGENT_ARN="<StoryAgentArn depuis les sorties CDK>"

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

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

# Invoquer l'agent
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"}'

Note

L’encodage URL est nécessaire pour formater correctement l’ARN pour le point de terminaison de l’API Bedrock AgentCore.

Si la commande s’exécute avec succès, vous devriez commencer à voir le texte initial de l’histoire en flux continu sous forme de JSON Lines :

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

Félicitations. Vous avez construit et déployé votre premier agent Strands sur Bedrock AgentCore Runtime ! 🎉🎉🎉