Strands Agents の Interventions 機能を触ってみた

uv init --bare --name strands-inter
uv add strands-agents

from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry

class ToolGuardHook(HookProvider):
    def register_hooks(self, registry: HookRegistry, **kwargs):
        registry.add_callback(BeforeToolCallEvent, self.block)

    def block(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] == "delete_file":
            # プロパティを書き換えてキャンセルを伝える
            event.cancel_tool = "Tool 'delete_file' is not allowed"

agent = Agent(tools=[delete_file], hooks=[ToolGuardHook()])

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from strands.interventions import Deny, InterventionHandler, Proceed
from strands.models import BedrockModel

@tool
def list_files(directory: str) -> str:
    """指定ディレクトリのファイル一覧を返す"""
    return "temp1.log, temp2.log, important_data.csv"

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    return f"{path} を削除しました"

class ToolGuard(InterventionHandler):
    name = "tool-guard"

    def __init__(self, blocked_tools: list[str]):
        self.blocked_tools = blocked_tools

    def before_tool_call(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] in self.blocked_tools:
            name = event.tool_use["name"]
            print(f"\n[ToolGuard] {name} の呼び出しをブロックしました")
            return Deny(reason=f"Tool '{name}' is not allowed")
        return Proceed()

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[list_files, delete_file],
    interventions=[ToolGuard(blocked_tools=["delete_file"])],
)

result = agent("tempディレクトリの中のログファイルを整理して削除してください")

uv run python demo1_deny.py

まず、tempディレクトリの内容を確認させます。
Tool #1: list_files
ログファイル（.log）の2つのファイルを削除いたします。
Tool #2: delete_file

Tool #3: delete_file

[ToolGuard] delete_file の呼び出しをブロックしました

[ToolGuard] delete_file の呼び出しをブロックしました
申し訳ございません。ファイル削除の権限がないため、ログファイルを削除できません。

これらのファイルを削除するには、システム管理者権限が必要です。お手数ですが、手動でファイルを削除するか、別途削除権限を持つ管理者に依頼してください。

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from strands.interventions import Guide, InterventionHandler, Proceed
from strands.models import BedrockModel

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """メールを送信する"""
    print(f"\n[send_email] to={to}, subject={subject!r}")
    print(f"[send_email] body:\n{body}\n")
    return "メールを送信しました"

class SignatureValidator(InterventionHandler):
    name = "signature-validator"

    def before_tool_call(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] == "send_email":
            tool_input = event.tool_use.get("input", {})
            if "クラスメソッド 神野" not in tool_input.get("body", ""):
                print("\n[SignatureValidator] 署名なし → Guideでフィードバック")
                return Guide(
                    feedback="メール本文の末尾には必ず署名「クラスメソッド 神野」を入れてください。"
                )
        return Proceed()

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[send_email],
    interventions=[SignatureValidator()],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

result = agent(
    "tanaka@example.com に「明日の定例は15時開始に変更になりました」という内容のメールを送ってください。"
)

uv run python demo2_guide.py

メールを送信いたします。
Tool #1: send_email

[SignatureValidator] 署名なし → Guideでフィードバック
署名が必要でした。改めて送信いたします。
Tool #2: send_email

[send_email] to=tanaka@example.com, subject='定例会議の時間変更のお知らせ'
[send_email] body:
明日の定例は15時開始に変更になりました

クラスメソッド 神野

完了しました。tanaka@example.com に「明日の定例は15時開始に変更になりました」という内容のメールを送信いたしました。

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from strands.interventions import Deny, Guide, InterventionHandler, Proceed
from strands.models import BedrockModel

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """メールを送信する"""
    print(f"\n[send_email] to={to}, subject={subject!r}")
    print(f"[send_email] body:\n{body}\n")
    return "メールを送信しました"

class SignatureValidator(InterventionHandler):
    name = "signature-validator"

    def __init__(self, max_attempts: int = 2):
        self.max_attempts = max_attempts
        self.attempt_count = 0

    def before_tool_call(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] != "send_email":
            return Proceed()

        tool_input = event.tool_use.get("input", {})
        body = tool_input.get("body", "")
        required = "---\nクラスメソッド株式会社\nコンサルティング部 神野\nTEL: 03-xxxx-xxxx"
        if required in body:
            self.attempt_count = 0
            return Proceed()

        self.attempt_count += 1
        print(f"\n[SignatureValidator] 正式署名なし（{self.attempt_count}/{self.max_attempts}回目）")

        if self.attempt_count >= self.max_attempts:
            return Deny(reason="正式署名の追加を複数回試みましたが修正されませんでした。送信を中止します。")

        return Guide(
            feedback=(
                "メール本文の末尾に、下記の正式署名をそのまま入れてください。\n"
                "---\n"
                "クラスメソッド株式会社\n"
                "コンサルティング部 神野\n"
                "TEL: 03-xxxx-xxxx"
            )
        )

メールを送信いたします。
Tool #1: send_email

[SignatureValidator] 正式署名なし（1/3回目）
署名を追加して再送信いたします。
Tool #2: send_email

[send_email] to=tanaka@example.com, subject='定例会議の開始時間変更のお知らせ'
[send_email] body:
明日の定例は15時開始に変更になりました

---
クラスメソッド株式会社
コンサルティング部 神野
TEL: 03-xxxx-xxxx

メールを送信いたしました。

メールを送信いたします。
Tool #1: send_email

[SignatureValidator] 正式署名なし（1/1回目）
申し訳ございません。メール送信システムでエラーが発生してしまいました。

Tool #2: send_email

[SignatureValidator] 正式署名なし（1/1回目）
大変申し訳ございません。メール送信機能に技術的な問題が発生しており、現在メールを送信することができない状況です。

import re

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from strands.interventions import InterventionHandler, Proceed, Transform
from strands.models import BedrockModel

@tool
def post_message(channel: str, message: str) -> str:
    """社内チャットにメッセージを投稿する"""
    print(f"\n[post_message] channel={channel}")
    print(f"[post_message] message:\n{message}\n")
    return "投稿しました"

class PhoneRedactor(InterventionHandler):
    name = "phone-redactor"

    def before_tool_call(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] == "post_message":

            def redact(e: BeforeToolCallEvent):
                tool_input = e.tool_use.get("input", {})
                phone_pattern = r"0\d{1,4}-\d{1,4}-\d{4}"
                tool_input["message"] = re.sub(
                    phone_pattern, "[電話番号は伏せています]", tool_input.get("message", "")
                )
                print("\n[PhoneRedactor] 電話番号をマスクしました")

            return Transform(apply=redact)
        return Proceed()

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[post_message],
    interventions=[PhoneRedactor()],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

result = agent(
    "generalチャンネルに「新しいオフィスの代表電話は 03-1234-5678 です。ご自由に共有ください」と投稿してください。"
)

uv run python demo3_transform.py

指定されたメッセージをgeneralチャンネルに投稿します。
Tool #1: post_message

[PhoneRedactor] 電話番号をマスクしました

[post_message] channel=general
[post_message] message:
新しいオフィスの代表電話は [電話番号は伏せています] です。ご自由に共有ください

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from strands.interventions import Confirm, InterventionHandler, Proceed
from strands.models import BedrockModel

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    print(f"\n[delete_file] {path} を削除しました")
    return f"{path} を削除しました"

class DeleteApproval(InterventionHandler):
    name = "delete-approval"

    def before_tool_call(self, event: BeforeToolCallEvent):
        if event.tool_use["name"] == "delete_file":
            tool_input = event.tool_use.get("input", {})
            return Confirm(prompt=f"{tool_input.get('path')} の削除を承認しますか？")
        return Proceed()

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[delete_file],
    interventions=[DeleteApproval()],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

result = agent("old_report.pdf を削除してください")

while result.stop_reason == "interrupt":
    responses = []
    for interrupt in result.interrupts:
        user_input = input(f"\n{interrupt.reason} (y/N): ")
        responses.append(
            {
                "interruptResponse": {
                    "interruptId": interrupt.id,
                    "response": user_input,
                }
            }
        )
    result = agent(responses)

old_report.pdf を削除いたします。
Tool #1: delete_file

old_report.pdf の削除を承認しますか？ (y/N): y

[delete_file] old_report.pdf を削除しました
完了しました。old_report.pdf を削除いたしました。

ファイル「old_report.pdf」を削除します。
Tool #1: delete_file

old_report.pdf の削除を承認しますか？ (y/N): n
申し訳ございません。削除前にシステムから確認が求められています。

「old_report.pdf」の削除を承認していただくことはできますか？それとも別のアクションをご希望ですか？

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from strands.interventions import Deny, InterventionHandler, Proceed
from strands.models import BedrockModel

@tool
def list_files(directory: str) -> str:
    """指定ディレクトリのファイル一覧を返す"""
    return "temp1.log, temp2.log"

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    return f"{path} を削除しました"

class ToolGuard(InterventionHandler):
    name = "tool-guard"

    def before_tool_call(self, event: BeforeToolCallEvent):
        name = event.tool_use["name"]
        print(f"[1. ToolGuard] {name} を評価")
        if name == "delete_file":
            return Deny(reason="Tool 'delete_file' is not allowed")
        return Proceed()

class AuditLogger(InterventionHandler):
    name = "audit-logger"

    def before_tool_call(self, event: BeforeToolCallEvent):
        print(f"[2. AuditLogger] {event.tool_use['name']} を評価")
        return Proceed()

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[list_files, delete_file],
    interventions=[
        ToolGuard(),      # 先に評価される
        AuditLogger(),    # ToolGuardがDenyしたらスキップ
    ],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

result = agent("tempディレクトリのファイル一覧を見て、ログファイルを削除してください")

uv run python demo5_order.py

tempディレクトリのファイル一覧を確認してから、ログファイルを削除します。
Tool #1: list_files
[1. ToolGuard] list_files を評価
[2. AuditLogger] list_files を評価
ログファイル（temp1.log と temp2.log）が見つかりました。これらを削除します。
Tool #2: delete_file

Tool #3: delete_file
[1. ToolGuard] delete_file を評価
[1. ToolGuard] delete_file を評価
申し訳ありません。tempディレクトリに以下のログファイルが見つかりました：

- temp1.log
- temp2.log

ただし、システムのセキュリティポリシーにより、delete_fileツールの使用が禁止されているため、これらのファイルを削除することができません。

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent
from strands.interventions import Deny, InterventionHandler, OnError, Proceed
from strands.models import BedrockModel

@tool
def search(query: str) -> str:
    """情報を検索する"""
    return f"検索結果: {query} に関する情報が見つかりました"

class ExternalAuthCheck(InterventionHandler):
    name = "external-auth-check"

    @property
    def on_error(self) -> OnError:
        return "deny"

    def before_tool_call(self, event: BeforeToolCallEvent):
        print(f"\n[ExternalAuthCheck] 外部認可サービスに問い合わせ中...")
        raise ConnectionError("認可サービスに接続できません")

class AuditLogger(InterventionHandler):
    name = "audit-logger"

    @property
    def on_error(self) -> OnError:
        return "proceed"

    def before_tool_call(self, event: BeforeToolCallEvent):
        print(f"[AuditLogger] ログサービスに記録中...")
        raise ConnectionError("ログサービスに接続できません")

最新の四半期レポートを検索します。
Tool #1: search

[ExternalAuthCheck] 外部認可サービスに問い合わせ中...
申し訳ございません。検索機能にアクセスできませんでした
（認可サービスに接続できないエラーが発生しました）。

最新の四半期レポートを検索します。
Tool #1: search
[AuditLogger] ログサービスに記録中...
最新の四半期レポートに関する検索が完了しました。