Skip to main content

AWS Strands Agents

Strands hooks fire before and after each model call and each tool call. Register one hook provider and Alice evaluates the model input, the tool arguments, the tool result and the model output. Block cancels the call with a reason the model can see; mask rewrites the content in place.

Status: Generally available · Evaluates: Prompts, Responses, Tool calls · Vendor: AWS

Model input, tool arguments, tool results, model output.

Note: This example uses the v1 WonderFenceClient interface of the WonderFence SDK.

Setup​

pip install strands-agents wonderfence-sdk

Configuration: ALICE_API_KEY, ALICE_APP_ID

Example​

imports and client (strands_agent_hooks_bank_example.py)

import json
import logging
import os
import uuid
from collections.abc import Callable

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent, tool
from strands.hooks import (
    AfterModelCallEvent,
    AfterToolCallEvent,
    BeforeModelCallEvent,
    BeforeToolCallEvent,
    HookProvider,
    HookRegistry,
)
from strands.models.gemini import GeminiModel

from wonderfence_sdk.client import WonderFenceClient
from wonderfence_sdk.models import Actions, AnalysisContext

logger = logging.getLogger(__name__)

DEMO_USER_ID = "user_id_123"

BANK_BALANCE = {DEMO_USER_ID: 1500.50}

client = WonderFenceClient(provider="aws-bedrock", platform="aws")

hook provider: register + before model call + before tool call

class WonderFenceBankingHook(HookProvider):
    """Hook provider that integrates WonderFence safety evaluation for banking tools."""

    def __init__(self, wonderfence_client: WonderFenceClient):
        self.client = wonderfence_client

    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeModelCallEvent, self.on_before_model_call)
        registry.add_callback(AfterModelCallEvent, self.on_after_model_call)
        registry.add_callback(BeforeToolCallEvent, self.on_before_tool_call)
        registry.add_callback(AfterToolCallEvent, self.on_after_tool_call)

    def on_before_model_call(self, event: BeforeModelCallEvent) -> None:
        """Evaluates model input for safety before sending to the model."""
        content = self._extract_messages_content(event)
        context = AnalysisContext(session_id=self._get_session_id(event))

        logger.info("WonderFence model input evaluation", {"preview": content[:100]})

        try:
            result = self.client.evaluate_prompt_sync(content, context)
            if result.action == Actions.BLOCK:
                logger.warning("Model input blocked")
                event.cancel_model_call = "Access Denied: Model input violates content policy."
            elif result.action == Actions.MASK:
                logger.info("Model input sanitized")
            else:
                logger.info("Model input safe")
        except Exception as e:
            logger.error("Model input evaluation error", {"error": str(e)})

    def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
        """Evaluates tool input for safety and blocks unsafe tool calls."""
        tool_name = event.tool_use.get("name", "unknown")
        content = f"Tool: {tool_name}, Input: {json.dumps(event.tool_use.get('input', {}))}"
        context = AnalysisContext(session_id=self._get_session_id(event))

        logger.info("WonderFence tool input evaluation", {"tool": tool_name, "preview": content[:100]})

        try:
            result = self.client.evaluate_prompt_sync(content, context)
            if result.action == Actions.BLOCK:
                logger.warning("Tool input blocked", {"tool": tool_name})
                event.cancel_tool = f"Access Denied: Tool '{tool_name}' input violates content policy."
            elif result.action == Actions.MASK:
                logger.info("Tool input sanitized", {"tool": tool_name})
            else:
                logger.info("Tool input safe", {"tool": tool_name})
        except Exception as e:
            logger.error("Tool input evaluation error", {"tool": tool_name, "error": str(e)})

attach the hook to the agent

def create_bank_agent(tool_functions: list) -> Agent:
    """Initializes and returns the Agent instance with WonderFence hooks."""
    model = GeminiModel(model_id="gemini-2.0-flash-exp", client_args={"api_key": os.getenv("GOOGLE_API_KEY")})
    wonderfence_hook = WonderFenceBankingHook(wonderfence_client=client)

    agent = Agent(
        model=model,
        tools=tool_functions,
        hooks=[wonderfence_hook],
        system_prompt=(
            "You are a specialized Bank Agent. Your primary function is to check "
            "and update user bank balances using the available tools. When asked "
            "about a balance, use 'get_account_balance'. When asked to deposit "
            "or withdraw money, use 'update_account_balance'. The default user ID is 'user_id_123'."
        ),
    )
    return agent

app = BedrockAgentCoreApp()
bank_tools = [get_account_balance, update_account_balance]
bank_agent = create_bank_agent(bank_tools)

Good to know​

The after-call handlers follow the same shape with evaluate_response_sync and rewrite the content on MASK. This example runs on Bedrock AgentCore Runtime and uses a Gemini model; swap the model for yours.