8月28日、AWSが「Extend Amazon Bedrock Guardrails to Tool Interactions Using the Strands Agents SDK」と題した記事を公開した。この記事では、Strands Agents SDKのライフサイクルフックを使い、Amazon Bedrock Guardrailsの保護範囲をモデル境界からツール呼び出し境界まで拡張する実装方法について詳しく紹介されている。以下に、その内容を紹介する。

モデル境界だけでは守れない理由

Amazon Bedrock Guardrailsは、モデルへの入力プロンプトとモデルからの出力レスポンスを検査する。しかしAIエージェントは、モデルを呼び出すだけでなく、外部ツールを実行し、MCP（Model Context Protocol）サーバーと通信し、RAGパイプラインからデータを取得する。これらはすべてモデル境界の外側で起きるため、モデルレベルのガードレールは届かない。

具体的に穴が開く箇所は4つある：

• ツールパラメータが未検査で渡る。モデルが決定したパラメータに個人情報（PII）やポリシー違反のコンテンツが含まれていても、そのままツールが実行される
• 外部データが無検証で流入する。ツールレスポンスやMCP出力がモデルレベルのガードレールより先にエージェントの挙動に影響を与える
• 誤情報がエージェントの推論を汚染する。外部ソースから取得した不正確・誤誘導的なコンテンツを、エージェントが権威ある情報として扱う（金融・医療・法律分野では特に危険）
• マルチエージェント構成でバッドデータが伝播する。上流エージェントのポリシー違反コンテンツがツール層を通じて下流エージェントに流れ込む

3つのバリデーションチェックポイント

この記事が提案するのは、信頼境界ごとに3か所でガードレールを適用するアーキテクチャだ。

Checkpoint 2が最も重要だ。モデルはすでに「何を送るか」を決定済みだが、その内容が安全かどうかは誰も確認していない。このタイミングでフックを挟むことで、実際のツール実行前にキャンセルできる。

検証強度はチェックポイントごとに調整できる。Checkpoint 1はPII検出・コンテンツフィルタリング・トピック制約を含むフルのガードレールを使い、Checkpoint 2はツールごとにカスタムルールを適用するか、正規表現やスキーマ検証などのローカルチェックで軽くする。Checkpoint 3は外部データを返すツール（Webサーチなど）に絞って不適切コンテンツ検出を重点的に行う、という使い分けが推奨されている。

実装：GuardrailHookクラス

実装の核心はGuardrailHookクラスだ。Strands Agents SDKのHookProviderを継承し、3つのライフサイクルイベントにコールバックを登録する。内部ではboto3でApplyGuardrail APIを呼び出す。

import boto3
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import (
    BeforeInvocationEvent,
    BeforeToolCallEvent,
    AfterToolCallEvent,
)

class GuardrailHook(HookProvider):

    def __init__(self, guardrail_id, guardrail_version, region_name, tool_names=None):
        self.client = boto3.client("bedrock-runtime", region_name=region_name)
        self.guardrail_id = guardrail_id
        self.guardrail_version = guardrail_version
        self.tool_names = tool_names  # None = apply to all tools

    def register_hooks(self, registry: HookRegistry, **kwargs):
        registry.add_callback(BeforeInvocationEvent, self.validate_inbound)
        registry.add_callback(BeforeToolCallEvent, self.validate_input)
        registry.add_callback(AfterToolCallEvent, self.validate_output)

    def _check(self, content, source="INPUT"):
        response = self.client.apply_guardrail(
            guardrailIdentifier=self.guardrail_id,
            guardrailVersion=self.guardrail_version,
            source=source,
            content=[{"text": {"text": content}}],
        )
        return response["action"] != "GUARDRAIL_INTERVENED"

    # Checkpoint 1
    async def validate_inbound(self, event: BeforeInvocationEvent):
        for msg in reversed(event.messages):
            if msg.get("role") == "user":
                for block in msg.get("content", []):
                    text = block.get("text", "")
                    if text and not self._check(text):
                        event.messages.clear()
                        event.messages.append({
                            "role": "user",
                            "content": [{"text": "Request blocked by safety guardrail."}],
                        })
                        return
                break

    # Checkpoint 2
    async def validate_input(self, event: BeforeToolCallEvent):
        if self.tool_names and event.tool_use.get("name") not in self.tool_names:
            return
        tool_input = event.tool_use.get("input", {})
        for param_value in tool_input.values():
            if isinstance(param_value, str) and not self._check(param_value):
                event.cancel_tool = "This request was blocked by a safety guardrail."
                return

    # Checkpoint 3
    async def validate_output(self, event: AfterToolCallEvent):
        if self.tool_names and event.tool_use.get("name") not in self.tool_names:
            return
        content_parts = [
            block["text"]
            for block in event.result.get("content", [])
            if "text" in block
        ]
        content = "\n".join(content_parts)
        if content and not self._check(content, source="OUTPUT"):
            event.result = {
                "toolUseId": event.result["toolUseId"],
                "status": "error",
                "content": [{"text": "Content blocked by safety guardrail."}],
            }

既存のツールやエージェントロジックへの変更は一切不要。Agentコンストラクタのhooksパラメータに渡すだけで有効になる。

ツールごとにガードレールを使い分ける

tool_namesパラメータを使えば、特定のツールだけにフックを適用できる。例えばWebサーチツールには厳格なコンテンツフィルタリングとPII検出を組み合わせたガードレールを、顧客データツールにはPII検出特化のガードレールを、それぞれ別のフックとして登録できる。

web_search_hook = GuardrailHook(
    guardrail_id="gr-websearch-id",
    guardrail_version="1",
    region_name="us-east-1",
    tool_names={"web_search"},  # web_searchのみ
)

customer_data_hook = GuardrailHook(
    guardrail_id="gr-customer-id",
    guardrail_version="1",
    region_name="us-east-1",
    tool_names={"get_customer_data"},  # get_customer_dataのみ
)

この設計は最小権限の原則をツール層に適用したものとも言える。すべてに同一のガードレールを掛けるのではなく、リスクプロファイルに応じた制御が可能になる。

前提条件

• AWS アカウントおよびAmazon Bedrockへのアクセス
• Amazon Bedrock Guardrailsの設定済みガードレール（作成方法）
• Python 3.11以上
• pip install strands-agents
• bedrock:ApplyGuardrailおよびbedrock:InvokeModel権限を持つAWS認証情報

Strandsエージェントをゼロから構築する場合は、公式のStrands Agents SDK技術解説が参考になる。

詳細はExtend Amazon Bedrock Guardrails to Tool Interactions Using the Strands Agents SDKを参照していただきたい。