Implementar e configurar o agente Story

Tarefa 1: Implementar o Agente de História

O Agente de História é um agente Strands gerado com --protocol=AG-UI no Módulo 1, para que a UI possa transmitir a partir dele através do protocolo Agent-User Interaction via CopilotKit. Ele usa o Inventory MCP Server para gerenciar os itens do jogador, e o S3SessionManager integrado do Strands para persistir o histórico de conversação no bucket de sessões que provisionamos no Módulo 2.

Implementação do Agente

Atualize os seguintes arquivos em packages/story/dungeon_adventure_story/agent:

main.py
agent.py

import logging
import os
import uuid
from functools import cache
from typing import Any, cast

from ag_ui.core import RunAgentInput
from ag_ui_strands import StrandsAgent, StrandsAgentConfig, create_strands_app
from aws_lambda_powertools.utilities import parameters
from dungeon_adventure_agent_connection import session_id_context
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)

SESSION_ID_HEADER = "x-amzn-bedrock-agentcore-runtime-session-id"


@cache
def _resolve_sessions_bucket() -> str:
    """Read the conversation-history bucket name from runtime config.

    Resolved lazily (and memoised) so `fastapi dev` can import this module
    before ``RUNTIME_CONFIG_APP_ID`` is in the environment.
    """
    application = os.environ.get("RUNTIME_CONFIG_APP_ID")
    if not application:
        raise RuntimeError("RUNTIME_CONFIG_APP_ID is not set — cannot resolve the StorySessions bucket.")
    provider = parameters.AppConfigProvider(environment="default", application=application)
    buckets = cast(dict[str, Any], provider.get("buckets", transform="json"))
    return buckets["StorySessions"]["bucketName"]


def _session_manager_provider(input_data: RunAgentInput) -> SessionManager:
    """Create a session manager keyed by the AG-UI thread_id.

    - In AgentCore (and `agent-serve`), persist to the shared ``StorySessions``
      S3 bucket — the same bucket the Game API reads from to rebuild
      conversation history on revisit.
    - In `agent-dev` (`LOCAL_DEV=true`), persist to a temp
      directory so the agent can run fully offline against the local MCP
      server without any AWS calls.
    """
    session_id = input_data.thread_id or "default"
    if os.environ.get("LOCAL_DEV") == "true":
        return FileSessionManager(session_id=session_id, storage_dir="/tmp/strands-sessions")
    return S3SessionManager(session_id=session_id, bucket=_resolve_sessions_bucket())


# The template Agent is cloned per thread_id by ``StrandsAgent`` — we plug in
# a ``session_manager_provider`` so each thread gets its own session manager
# and conversation history is replayed on subsequent turns and survives agent
# restarts.
_agent_ctx = get_agent()
_agent = _agent_ctx.__enter__()

agui_agent = StrandsAgent(
    agent=_agent,
    name="StoryAgent",
    description="A Strands Agent exposed via the AG-UI protocol.",
    config=StrandsAgentConfig(session_manager_provider=_session_manager_provider),
)


class _SessionIdMiddleware(BaseHTTPMiddleware):
    """Bind the session ID for this request so downstream MCP / A2A clients forward it on outbound calls."""

    async def dispatch(self, request: Request, call_next):
        session_id = request.headers.get(SESSION_ID_HEADER) or str(uuid.uuid4())
        with session_id_context(session_id):
            return await call_next(request)


app = create_strands_app(agui_agent, path="/invocations")
app.add_middleware(_SessionIdMiddleware)

import logging
import os
import uuid
from functools import cache
from typing import Any, cast

from ag_ui_strands import StrandsAgent, create_strands_app
from ag_ui.core import RunAgentInput
from ag_ui_strands import StrandsAgent, StrandsAgentConfig, create_strands_app
from aws_lambda_powertools.utilities import parameters
from dungeon_adventure_agent_connection import session_id_context
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from strands.session import FileSessionManager, S3SessionManager, SessionManager

from .agent import get_agent

logging.basicConfig(level=logging.INFO)

SESSION_ID_HEADER = "x-amzn-bedrock-agentcore-runtime-session-id"

# Create AG-UI agent wrapper

@cache
def _resolve_sessions_bucket() -> str:
    """Read the conversation-history bucket name from runtime config.

    Resolved lazily (and memoised) so `fastapi dev` can import this module
    before ``RUNTIME_CONFIG_APP_ID`` is in the environment.
    """
    application = os.environ.get("RUNTIME_CONFIG_APP_ID")
    if not application:
        raise RuntimeError("RUNTIME_CONFIG_APP_ID is not set — cannot resolve the StorySessions bucket.")
    provider = parameters.AppConfigProvider(environment="default", application=application)
    buckets = cast(dict[str, Any], provider.get("buckets", transform="json"))
    return buckets["StorySessions"]["bucketName"]


def _session_manager_provider(input_data: RunAgentInput) -> SessionManager:
    """Create a session manager keyed by the AG-UI thread_id.

    - In AgentCore (and `agent-serve`), persist to the shared ``StorySessions``
      S3 bucket — the same bucket the Game API reads from to rebuild
      conversation history on revisit.
    - In `agent-dev` (`LOCAL_DEV=true`), persist to a temp
      directory so the agent can run fully offline against the local MCP
      server without any AWS calls.
    """
    session_id = input_data.thread_id or "default"
    if os.environ.get("LOCAL_DEV") == "true":
        return FileSessionManager(session_id=session_id, storage_dir="/tmp/strands-sessions")
    return S3SessionManager(session_id=session_id, bucket=_resolve_sessions_bucket())


# The template Agent is cloned per thread_id by ``StrandsAgent`` — we plug in
# a ``session_manager_provider`` so each thread gets its own session manager
# and conversation history is replayed on subsequent turns and survives agent
# restarts.
_agent_ctx = get_agent()
_agent = _agent_ctx.__enter__()

agui_agent = StrandsAgent(
    agent=_agent,
    name="StoryAgent",
    description="A Strands Agent exposed via the AG-UI protocol.",
    config=StrandsAgentConfig(session_manager_provider=_session_manager_provider),
)


class _SessionIdMiddleware(BaseHTTPMiddleware):
    """Bind the session ID for this request so downstream MCP / A2A clients forward it on outbound calls."""

    async def dispatch(self, request: Request, call_next):
        session_id = request.headers.get(SESSION_ID_HEADER) or str(uuid.uuid4())
        with session_id_context(session_id):
            return await call_next(request)


# Create FastAPI app with AG-UI endpoint and health check
app = create_strands_app(agui_agent, path="/invocations")
app.add_middleware(_SessionIdMiddleware)

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClientStrands, log_model_errors
from strands import Agent


@contextmanager
def get_agent():
    inventory_mcp_server = InventoryMcpServerClientStrands.create()
    with (
        inventory_mcp_server,
    ):
        yield Agent(
            system_prompt="""
You are running a text adventure game for a lone adventurer. When a new story
begins, the first user message will tell you the player's name and the genre
(one of 'medieval', 'zombie', 'superhero'). Greet the player by name, set the
scene in the chosen genre, and populate their inventory with a few starting
items. On subsequent turns, advance the story in response to the player's
actions and keep item state in sync with the narrative.
Use the tools to manage the player's inventory as items are obtained or lost.
Item names in the inventory must be Title Case — match them exactly.
Only use list-inventory-items if you are unsure of the exact item name before modifying it.
IMPORTANT: Only call ONE tool per response turn. Never batch multiple tool calls in a single turn.
Ensure you specify a suitable emoji when adding items if available.
Items should be a key part of the narrative.
Keep responses under 100 words.
""",
            tools=[*inventory_mcp_server.list_tools_sync()],
            hooks=[log_model_errors],
        )

from contextlib import contextmanager

from dungeon_adventure_agent_connection import InventoryMcpServerClientStrands, log_model_errors
from strands import Agent, tool
from strands_tools import current_time
from strands import Agent


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


@contextmanager
def get_agent():
    inventory_mcp_server = InventoryMcpServerClientStrands.create()
    with (
        inventory_mcp_server,
    ):
        yield Agent(
            name="StoryAgent",
            description="StoryAgent Strands Agent",
            system_prompt="""
You are a mathematical wizard.
Use your tools for mathematical tasks.
Refer to tools as your 'spellbook'.
You are running a text adventure game for a lone adventurer. When a new story
begins, the first user message will tell you the player's name and the genre
(one of 'medieval', 'zombie', 'superhero'). Greet the player by name, set the
scene in the chosen genre, and populate their inventory with a few starting
items. On subsequent turns, advance the story in response to the player's
actions and keep item state in sync with the narrative.
Use the tools to manage the player's inventory as items are obtained or lost.
Item names in the inventory must be Title Case — match them exactly.
Only use list-inventory-items if you are unsure of the exact item name before modifying it.
IMPORTANT: Only call ONE tool per response turn. Never batch multiple tool calls in a single turn.
Ensure you specify a suitable emoji when adding items if available.
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()],
            hooks=[log_model_errors],
        )

As alterações são:

main.py adiciona um session_manager_provider que cria um S3SessionManager por thread_id quando implantado, e retorna para um FileSessionManager em disco sob /tmp/strands-sessions quando executado sob agent-dev (LOCAL_DEV=true). Quando implantado, o bucket S3 é o mesmo do qual o queryActions da Game API lê, para que o navegador possa reconstruir transcrições ao revisitar; localmente o agente persiste sessões em disco e se comunica com o servidor MCP local, sem uma implantação.
agent.py remove a ferramenta de exemplo subtract e substitui o prompt de sistema por um de mestre de dungeon que convida a primeira mensagem do usuário a declarar o nome do jogador e o gênero, e usa as ferramentas do Inventory MCP Server.

Tarefa 2: Testar seu Agente localmente

Construir o código

Para construir o código:

pnpm build

yarn build

npm run build

bun build

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

pnpm lint

yarn lint

npm run lint

bun lint

Conversar com seu Agente

O destino agent-chat gerado abre um REPL interativo contra seu agente. Ele é executado de forma independente e se conecta ao seu agente em execução local, então primeiro inicie o servidor local do agente em um terminal:

pnpm nx agent-dev story

yarn nx agent-dev story

npx nx agent-dev story

bunx nx agent-dev story

Em seguida, em um segundo terminal, inicie o chat:

pnpm nx agent-chat story

yarn nx agent-chat story

npx nx agent-chat story

bunx nx agent-chat story

Sua primeira mensagem deve informar ao agente o nome do seu herói e o gênero (por exemplo: My name is Alice. Start my zombie adventure.) e a história será transmitida de volta.

Parabéns. Você construiu e testou seu primeiro Agente Strands localmente! 🎉🎉🎉