Implementar e configurar o agente Story

Tarefa 1: Implementar o Agente de História

O Agente de História é um agente Strands que, dado um Game e uma lista de Actions como contexto, irá progredir uma narrativa. Configuraremos o agente para interagir com nosso Inventory MCP Server e gerenciar os itens disponíveis de um jogador.

Nota

Para este tutorial, implementaremos o Agente de forma “stateless”, onde cada invocação é uma sessão independente incluindo todo o histórico de mensagens. Para produção, considere integrar com Bedrock AgentCore Memory. Para mais informações, consulte a página usando AgentCore Memory com Strands.

Implementação do Agente

Para implementar nosso agente, atualize os seguintes arquivos em 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

+

import os
from contextlib import contextmanager

import boto3
from aws_lambda_powertools.utilities import parameters
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()


def _get_inventory_mcp_arn() -> str:
    """Resolve the Inventory MCP server ARN from runtime config (AppConfig)."""
    config = parameters.get_app_config(
        name="connection",
        environment="default",
        application=os.environ["RUNTIME_CONFIG_APP_ID"],
        transform="json",
    )
    return config["agentRuntimes"]["InventoryMcpServer"]


@contextmanager
def get_agent(player_name: str, genre: str, session_id: str):
    mcp_client = AgentCoreMCPClient.with_iam_auth(
        agent_runtime_arn=_get_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 aws_lambda_powertools.utilities import parameters
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()


def _get_inventory_mcp_arn() -> str:
    """Resolve the Inventory MCP server ARN from runtime config (AppConfig)."""
    config = parameters.get_app_config(
        name="connection",
        environment="default",
        application=os.environ["RUNTIME_CONFIG_APP_ID"],
        transform="json",
    )
    return config["agentRuntimes"]["InventoryMcpServer"]


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

Esta configuração implementa:

• Extração do jogador, gênero e ações do payload do agente,
• Construção de um cliente que o Agente pode usar para invocar nosso MCP Server com autenticação SigV4, e
• Construção do agente com um prompt de sistema e as ferramentas do MCP Server.

Tarefa 2: Implantação e testes

Construir o código

Para construir o código:

pnpm

pnpm build

yarn

yarn build

npm

npm run build

bun

bun build

Dica

Se encontrar erros de lint, execute o seguinte comando para corrigi-los automaticamente.

pnpm

pnpm lint

yarn

yarn lint

npm

npm run lint

bun

bun lint

Implantar sua aplicação

Para implantar sua aplicação, execute o seguinte 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/*"

Esta implantação levará aproximadamente 2 minutos para ser concluída.

Após a conclusão da implantação, você verá saídas similares às seguintes (alguns valores foram redigidos):

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

Testar seu Agente

Você pode testar seu Agente de duas formas:

• Iniciando uma instância local do servidor Agent e invocando-o usando curl, ou
• Chamando a API implantada usando curl com um token JWT.

Local

Inicie seu servidor Agent local executando o seguinte comando:

pnpm

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

yarn

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

npm

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

bun

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

Cuidado

Use o valor de saída RuntimeConfigApplicationId do CDK deploy para a variável de ambiente RUNTIME_CONFIG_APP_ID, e sua região AWS implantada para AWS_REGION.

Cuidado

Use a saída InventoryMcpArn do CDK deploy acima, e especifique sua AWS_REGION alvo.

Uma vez que o servidor Agent esteja em execução (você não verá nenhuma saída), invoque-o executando o seguinte comando:

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

Implantado

Para testar o agente implantado, você precisará autenticar com Cognito e obter um token JWT. Primeiro, configure suas variáveis de ambiente:

# Defina seu Cognito User Pool ID e Client ID das saídas do CDK
export POOL_ID="<UserPoolId das saídas do CDK>"
export CLIENT_ID="<UserPoolClientId das saídas do CDK>"
export REGION="<sua-regiao>"

Crie um usuário de teste e obtenha um token de autenticação:

# Desabilite MFA para simplificar a criação do usuário
aws cognito-idp set-user-pool-mfa-config \
  --mfa-configuration OFF \
  --user-pool-id $POOL_ID

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

# Definir Senha Permanente (substitua por algo mais seguro!)
aws cognito-idp admin-set-user-password \
  --user-pool-id $POOL_ID \
  --username "test" \
  --password "PermanentPass123-" \
  --region $REGION \
  --permanent > /dev/null

# Autenticar Usuário e capturar 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)

Invoque o agente implantado usando a URL do runtime do Bedrock AgentCore:

# Defina a ARN do Story Agent das saídas do CDK
export AGENT_ARN="<StoryAgentArn das saídas do CDK>"

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

# Construa a URL de invocação
export AGENT_URL="https://bedrock-agentcore.$REGION.amazonaws.com/runtimes/$ENCODED_ARN/invocations?qualifier=DEFAULT"

# Invoque o 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

A codificação URL é necessária para formatar corretamente a ARN para o endpoint da API do Bedrock AgentCore.

Se o comando for executado com sucesso, você começará a ver o texto inicial da história sendo transmitido como JSON Lines:

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

Parabéns. Você construiu e implantou seu primeiro Agente Strands no Bedrock AgentCore Runtime! 🎉🎉🎉