Implement and configure the Story agent

Task 1: Implement the Story Agent

The Story Agent is a Strands agent which, given a Game and a list of Actions for context, will progress a story. We will configure the agent to interact with our Inventory MCP Server to manage a player’s available items.

Python Implementation

For this tutorial, we will implement the Agent in a “stateless” fashion, where each invocation is an independent session including the full message history. For production, consider integrating with Bedrock AgentCore Memory. For more information, refer to the using AgentCore Memory with Strands page.

Agent implementation

To implement our agent, update the following files in 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()],
        )

Since the connection generator already wired up the MCP client in Module 1, these changes focus on:

• Removing the sample tool and unused imports,
• Adding the player_name and genre parameters to the get_agent function, and
• Customizing the system prompt for our dungeon adventure game.

Task 2: Deployment and testing

Build the code

To build the code:

pnpm

pnpm build

yarn

yarn build

npm

npm run build

bun

bun build

Fixing Lint Errors

If you encounter any lint errors, run the following command to automatically fix them.

pnpm

pnpm lint

yarn

yarn lint

npm

npm run lint

bun

bun lint

Deploy your application

To deploy your application, run the following command:

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

This deployment will take around 2 minutes to complete.

Once the deployment completes, you will see outputs similar to the following (some values have been redacted):

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

Test your Agent

You can test your Agent by either:

• Starting a local instance of the Agent server and invoking it using curl, or
• Calling the deployed API using curl with a JWT token.

Local

Start your local Agent server by running the following command:

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

Use CDK Output Values

Use the CDK deploy output value of RuntimeConfigApplicationId for the RUNTIME_CONFIG_APP_ID environment variable, and your deployed AWS region for AWS_REGION.

Use CDK Output Values

Use the CDK deploy InventoryMcpArn output in the above, and specify your target AWS_REGION.

Once the Agent server is up and running (you won’t see any output), call it by running the following command:

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

Deployed

To test the deployed agent, you’ll need to authenticate with Cognito and obtain a JWT token. First, set up your environment variables:

# Set your Cognito User Pool ID and Client ID from the CDK outputs
export POOL_ID="<UserPoolId from CDK outputs>"
export CLIENT_ID="<UserPoolClientId from CDK outputs>"
export REGION="<your-region>"

Create a test user and obtain an authentication token:

# Disable MFA to simplify user creation
aws cognito-idp set-user-pool-mfa-config \
  --mfa-configuration OFF \
  --user-pool-id $POOL_ID

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

# Set Permanent Password (replace with something more secure!)
aws cognito-idp admin-set-user-password \
  --user-pool-id $POOL_ID \
  --username "test" \
  --password "PermanentPass123-" \
  --region $REGION \
  --permanent > /dev/null

# Authenticate User and capture ID 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)

Invoke the deployed agent using the Bedrock AgentCore runtime URL:

# Set the Story Agent ARN from CDK outputs
export AGENT_ARN="<StoryAgentArn from CDK outputs>"

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

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

# Invoke the 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"}'

URL Encoding

The URL encoding is necessary to properly format the ARN for the Bedrock AgentCore API endpoint.

If the command runs successfully, you should start to see the initial story text being streamed as JSON Lines:

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

Congratulations. You have built and deployed your first Strands Agent on Bedrock AgentCore Runtime! 🎉🎉🎉