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.

Note

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

Cela permettra de configurer les éléments suivants :

• L’extraction du joueur, du genre et des actions depuis le payload de l’agent,
• La construction d’un client que l’agent peut utiliser pour invoquer notre serveur MCP avec une authentification SigV4, et
• La construction de l’agent avec un prompt système et les outils du serveur MCP.

Tâche 2 : Déploiement et tests

Compiler le code

Pour compiler le code :

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

Astuce

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

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

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

Vous pouvez tester votre API 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

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

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 :

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

# Authentifier l'utilisateur et capturer le token ID
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')

Attention

Utilisez la sortie UserIdentityUserIdentityUserPoolIdXXX pour POOL_ID et assurez-vous d’avoir le Client ID correspondant depuis les sorties de votre déploiement CDK.

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

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

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 voir un flux d’événements similaire à :

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}}}

...

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