はじめに
こんばんは、mirukyです。
以前、Strands Agents SDKの全体像をまとめた記事を書きました。
「3行でエージェントが動く」という手軽さは伝わったかと思いますが、実際に本番で使おうとすると、次の疑問にぶつかります。
- ツールの暴走をどう止めるのか
- エージェントの振る舞いをどこまで制御できるのか
- 状態管理はどうなっているのか
Strands Agents SDKには、これらの問題に対処するための 内部メカニズム が3つあります。 Hooks 、 Plugin 、 Steering です。前回の記事ではほとんど触れなかった部分ですが、本番投入を考えるなら避けて通れません。
本記事では、この3つのメカニズムを中心に、エージェントの内部動作を深く掘り下げます。後半では、Hooksを使った ツール使用制限の実装 をハンズオン形式で解説します。
本記事のサンプルコードはGitHubリポジトリにまとめています。
本記事の情報は2026年4月9日時点のものです（Python SDK v1.35.0）。
目次
- エージェントループの内部動作
- Hooks：ライフサイクルイベントに介入する
- Plugin：Hooks + Toolsを束ねる拡張機構
- Steering：モデルの振る舞いをリアルタイムで誘導する
- Skills：オンデマンドで指示を読み込む
- State管理の3層構造
- ハンズオン：Hookでツール使用を制限する
1. エージェントループの内部動作
前回の記事ではエージェントループをReAct（Reasoning and Acting）パラダイムの一言で片付けましたが、もう少し踏み込みます。
1-1. ループの基本構造
エージェントループの動作は、公式ドキュメントによれば以下の原則で動きます。
"The agent loop operates on a simple principle: invoke the model, check if it wants to use a tool, execute the tool if so, then invoke the model again with the result. Repeat until the model produces a final response."
（モデルを呼び出し、ツールを使いたいかチェックし、使いたければツールを実行し、その結果をもってモデルを再度呼び出す。モデルが最終応答を出すまで繰り返す。）
重要なのは、各イテレーションで 会話履歴にコンテキストが蓄積 される点です。モデルは元のリクエストだけでなく、これまでに呼び出したすべてのツールとその結果を参照して次のアクションを決定します。
1-2. ループの終了条件
モデルの各呼び出しは Stop Reason を返し、これがループの継続・終了を決定します。
| Stop Reason | 動作 |
|---|---|
| end_turn | モデルが最終応答を生成。正常終了 |
| tool_use | ツール実行が必要。ループ継続 |
| cancelled |
agent.cancel() による外部キャンセル |
| max_tokens | トークン上限に到達。異常終了 |
| stop_sequence | 設定されたストップシーケンスに到達 |
| content_filtered | セーフティメカニズムが応答をブロック |
| guardrail_intervened | ガードレールポリシーが生成を停止 |
1-3. ツール実行時のエラーハンドリング
ツールが失敗した場合、例外がそのまま投げられてループが終了するわけではありません。エラー情報は error result としてモデルに返されます。モデルはそのエラーを見て、代替手段を試みたり、別のアプローチを取ることができます。
これが「モデル駆動」の核心です。ワークフローフレームワークなら開発者がエラーハンドリングのコードを書く必要がありますが、Strandsではモデル自身がリカバリを試みます。
1-4. キャンセルメカニズム
agent.cancel() はスレッドセーフかつ冪等です。以下の2つのチェックポイントで検出されます。
| チェックポイント | 動作 |
|---|---|
| モデル応答のストリーミング中 | 部分出力を破棄 |
| ツール実行前 | ツール呼び出しをスキップし、エラー結果を会話履歴に追加 |
import threading
from strands import Agent
def timeout_watchdog(agent: Agent, timeout: float) -> None:
"""タイムアウト後にエージェントをキャンセルする"""
import time
time.sleep(timeout)
agent.cancel()
agent = Agent()
# 30秒後にキャンセル
watchdog = threading.Thread(target=timeout_watchdog, args=(agent, 30.0))
watchdog.start()
result = agent("大規模データセットを分析してください")
watchdog.join()
if result.stop_reason == "cancelled":
print("タイムアウトによりキャンセルされました")
2. Hooks：ライフサイクルイベントに介入する
Hooksは、エージェントのライフサイクル全体に 型安全かつ合成可能な方法で 介入するための仕組みです。ログ記録、動作変更、バリデーション、メトリクス収集など、エージェントの本番運用に必要なあらゆる横断的関心事をカバーします。
2-1. フックイベント一覧
Strands Agents SDKが提供するフックイベントは以下の通りです。
| イベント | タイミング |
|---|---|
| AgentInitializedEvent | エージェントのコンストラクタ完了時 |
| BeforeInvocationEvent | 新しいリクエストの開始時 |
| AfterInvocationEvent | リクエスト終了時（成功・失敗問わず） |
| MessageAddedEvent | 会話履歴にメッセージ追加時 |
| BeforeModelCallEvent | モデル推論の呼び出し前 |
| AfterModelCallEvent | モデル推論の完了後 |
| BeforeToolCallEvent | ツール実行前 |
| AfterToolCallEvent | ツール実行後 |
マルチエージェント用のイベントも別途あります。
| イベント | タイミング |
|---|---|
| MultiAgentInitializedEvent | オーケストレーター初期化時 |
| BeforeMultiAgentInvocationEvent | オーケストレーター実行開始前 |
| AfterMultiAgentInvocationEvent | オーケストレーター実行完了後 |
| BeforeNodeCallEvent | 個別ノード実行前 |
| AfterNodeCallEvent | 個別ノード実行後 |
2-2. 基本的な使い方
最もシンプルな登録方法は agent.add_hook() です。
from strands import Agent
from strands.hooks import BeforeToolCallEvent, AfterToolCallEvent
agent = Agent()
# 型ヒントからイベントタイプを自動推論
def log_tool_call(event: BeforeToolCallEvent) -> None:
print(f"ツール呼び出し: {event.tool_use['name']}")
def log_tool_result(event: AfterToolCallEvent) -> None:
print(f"ツール完了: {event.tool_use['name']}")
agent.add_hook(log_tool_call)
agent.add_hook(log_tool_result)
2-3. 変更可能なイベントプロパティ
Hooksの真価は 読み取り専用ではない プロパティにあります。以下のプロパティを変更することで、エージェントの動作を外部から制御できます。
| イベント | プロパティ | 用途 |
|---|---|---|
| AfterModelCallEvent | retry |
モデル呼び出しのリトライを要求 |
| BeforeToolCallEvent | cancel_tool |
ツール実行をキャンセルしてメッセージを返す |
| BeforeToolCallEvent | selected_tool |
実行するツールを別のツールに差し替える |
| BeforeToolCallEvent | tool_use |
ツールのパラメータを実行前に変更 |
| AfterToolCallEvent | result |
ツールの実行結果を変更 |
| AfterToolCallEvent | retry |
ツール実行のリトライを要求 |
| AfterInvocationEvent | resume |
後続の呼び出しを自動トリガー |
たとえば、 BeforeToolCallEvent.cancel_tool に文字列を設定すると、そのツールの実行はスキップされ、代わりにその文字列がモデルへのフィードバックとして返されます。
2-4. コールバックの実行順序
Before/Afterのペアになるイベントでは、Afterイベントのコールバックは 逆順 で実行されます。これにより、適切なクリーンアップが保証されます。
BeforeInvocationEvent: callback_A → callback_B → callback_C
AfterInvocationEvent: callback_C → callback_B → callback_A
2-5. Invocation State経由のデータ共有
Hooksからカスタムデータにアクセスする方法として、 Invocation State があります。エージェント呼び出し時にキーワード引数で渡したデータが、すべてのフックから参照可能です。
from strands import Agent
from strands.hooks import BeforeToolCallEvent
def log_with_context(event: BeforeToolCallEvent) -> None:
user_id = event.invocation_state.get("user_id", "unknown")
session_id = event.invocation_state.get("session_id")
print(f"User {user_id} in session {session_id}: {event.tool_use['name']}")
agent = Agent()
agent.add_hook(log_with_context)
# 呼び出し時にコンテキストを渡す
result = agent(
"データを処理してください",
user_id="user123",
session_id="sess456",
)
JSON非直列化可能なオブジェクト（DBコネクションなど）も渡せます。
3. Plugin：Hooks + Toolsを束ねる拡張機構
Pluginは、HooksとToolsを一つのクラスにまとめて再利用可能にする仕組みです。v1.28.0で導入され、v1.30.0でSteeringとSkillsがPlugin上に構築されました。
3-1. Pluginの基本構造
from strands import Agent, tool
from strands.plugins import Plugin, hook
from strands.hooks import BeforeToolCallEvent, AfterToolCallEvent
class LoggingPlugin(Plugin):
"""すべてのツール呼び出しをログ出力するPlugin"""
name = "logging-plugin"
@hook
def log_before_tool(self, event: BeforeToolCallEvent) -> None:
print(f"[LOG] ツール呼び出し: {event.tool_use['name']}")
print(f"[LOG] 入力: {event.tool_use['input']}")
@hook
def log_after_tool(self, event: AfterToolCallEvent) -> None:
print(f"[LOG] ツール完了: {event.tool_use['name']}")
@tool
def debug_print(self, message: str) -> str:
"""デバッグメッセージを出力します。
Args:
message: 出力するメッセージ
"""
print(f"[DEBUG] {message}")
return f"出力完了: {message}"
# Pluginを渡すだけで、HooksもToolsも自動登録
agent = Agent(plugins=[LoggingPlugin()])
3-2. 内部で何が起きているか
Pluginをエージェントにアタッチすると、以下の順序で処理されます。
-
Plugin基底クラスが@hookと@toolで装飾されたメソッドを自動検出 -
@hookメソッドをエージェントのHookレジストリに登録 -
@toolメソッドをエージェントのツールリストに追加 -
init_agent(agent)メソッドを呼び出してカスタム初期化を実行
3-3. 状態を持つPlugin
Pluginはインスタンス変数で状態を保持できます。永続化が必要な場合は、 Agent State を使います。
from strands.plugins import Plugin, hook
from strands.hooks import BeforeToolCallEvent
class MetricsPlugin(Plugin):
name = "metrics-plugin"
def init_agent(self, agent) -> None:
if "metrics_call_count" not in agent.state:
agent.state.set("metrics_call_count", 0)
@hook
def count_calls(self, event: BeforeToolCallEvent) -> None:
current = event.agent.state.get("metrics_call_count", 0)
event.agent.state.set("metrics_call_count", current + 1)
agent = Agent(plugins=[MetricsPlugin()])
agent("何か処理してください")
print(f"ツール呼び出し回数: {agent.state.get('metrics_call_count')}")
4. Steering：モデルの振る舞いをリアルタイムで誘導する
SteeringはPlugin上に構築された機能で、v1.30.0で正式リリースされました。モデルの出力を リアルタイムで評価し、フィードバックを注入 します。
4-1. 従来のプロンプティングの問題
複雑なタスク（30ステップ以上など）をエージェントに任せる場合、すべての指示をシステムプロンプトに前もって詰め込む必要があります。これにより以下の問題が生じます。
- プロンプト肥大化 ：コンテキストウィンドウを圧迫し、推論に使えるトークンが減る
- 指示の無視 ：大量の指示の中から必要なものを取り出す精度が下がる
- ハルシネーション ：指示の一部を曲解し、定義していない振る舞いをする
4-2. Steeringのアプローチ
Steeringは モジュラープロンプティング というアプローチを取ります。全指示を前もって渡すのではなく、エージェントの行動に応じて必要な指示をリアルタイムで注入します。
Steering Handlerは2つのポイントでエージェントに介入します。
【Tool Steering（ツール実行前）】
エージェントがツールを呼び出す直前に、その呼び出しが適切かを評価します。3つのアクションを返せます。
| アクション | 動作 |
|---|---|
| Proceed | そのままツールを実行 |
| Guide | ツール実行をキャンセルし、フィードバックを注入 |
| Interrupt | ツール実行を一時停止し、人間の入力を待つ |
【Model Steering（モデル応答後）】
モデルが応答を生成した後、その内容を評価します。
| アクション | 動作 |
|---|---|
| Proceed | 応答をそのまま受け入れる |
| Guide | 応答を破棄し、ガイダンスを注入して再生成 |
4-3. 自然言語によるSteering
LLMSteeringHandler を使えば、ビジネスルールを自然言語で定義できます。
from strands import Agent, tool
from strands.vended_plugins.steering import LLMSteeringHandler
@tool
def send_email(recipient: str, subject: str, message: str) -> str:
"""メールを送信します。
Args:
recipient: 宛先メールアドレス
subject: 件名
message: 本文
"""
return f"メール送信完了: {recipient}"
handler = LLMSteeringHandler(
system_prompt="""
メールの送信前に、以下のルールを検証してください：
- メッセージのトーンがプロフェッショナルであること
- 攻撃的・否定的な表現が含まれていないこと
- フレンドリーな挨拶で始まっていること
ルールに違反している場合は、改善のフィードバックを提供してください。
"""
)
agent = Agent(
tools=[send_email],
plugins=[handler]
)
# 怒りのメールを送ろうとしても、Steeringが介入して丁寧な表現に修正させる
response = agent("会議をキャンセルし続けるクライアントに怒りのメールを送って")
この例では、エージェントが send_email ツールを呼び出す直前にSteeringが介入し、メッセージのトーンが不適切であればツール実行をキャンセルして修正を促します。
4-4. ワークフローフレームワークとの違い
| アプローチ | 特徴 |
|---|---|
| ワークフローフレームワーク | 離散的なステップと制御フローを事前定義。要件変更時にロジック全体を再構築 |
| 従来のプロンプティング | すべての指示を一つのプロンプトに前もって投入。30ステップ超で破綻 |
| Steering | コンテキストに応じたガイダンスを必要な瞬間に注入。適応的推論を維持 |
5. Skills：オンデマンドで指示を読み込む
SkillsもPlugin上に構築された機能です。エージェントのシステムプロンプトに全指示を前もって投入するのではなく、必要な時だけ読み込む仕組みです。
5-1. Skillsの動作フロー
- Discovery（検出） ：Plugin初期化時にスキルのメタデータ（名前と説明）を読み取り、XMLブロックとしてシステムプロンプトに注入
-
Activation（起動） ：エージェントがスキルを必要と判断した時、
skillsツールを呼び出して完全な指示をロード - Execution（実行） ：ロードされた指示に従ってタスクを実行
from strands import Agent, AgentSkills
from strands_tools import file_read, shell
plugin = AgentSkills(skills="./skills/")
agent = Agent(
plugins=[plugin],
tools=[file_read, shell],
)
5-2. SKILL.mdフォーマット
スキルは SKILL.md ファイルを含むディレクトリとして定義します。
---
name: pdf-processing
description: PDFファイルからテキストとテーブルを抽出する
allowed-tools: file_read shell
---
# PDF処理
PDFからコンテンツを抽出する際は以下の手順に従ってください：
1. `shell` で `scripts/extract.py` を実行
2. `file_read` で出力を確認
3. 抽出内容を要約して報告
5-3. Steeringとの使い分け
| アプローチ | 適した場面 |
|---|---|
| システムプロンプト | 少量の常に必要な指示 |
| Skills | 多数の専門ドメイン。必要な時だけ指示をロード |
| Steering | 動的な文脈依存ガイダンスとバリデーション |
| マルチエージェント | 根本的に異なる役割やモデルが必要な場合 |
6. State管理の3層構造
Strands Agentsの状態は3層で管理されています。理解が曖昧だとデバッグに苦労するため、ここで整理します。
6-1. 3層の概要
| 層 | スコープ | モデルへの入力 | 用途 |
|---|---|---|---|
| Conversation History | セッション全体 | あり | ユーザーとアシスタントのメッセージ履歴 |
| Agent State | 複数リクエスト間 | なし | アプリケーション固有のキーバリューストア |
| Request State | 単一リクエスト内 | なし | イベントループのサイクル間で保持 |
6-2. Conversation History
最も基本的な状態です。 agent.messages でアクセスできます。
from strands import Agent
agent = Agent()
agent("こんにちは")
# すべての会話履歴にアクセス
print(agent.messages)
会話履歴はモデルの推論に直接使われるため、コンテキストウィンドウを圧迫します。 SlidingWindowConversationManager がデフォルトで使用され、古いメッセージを自動的に削除します。
6-3. Agent State
モデルには渡されない、アプリケーション固有の状態です。JSON直列化可能な値のみ保存できます。
from strands import Agent
agent = Agent(state={"user_preferences": {"theme": "dark"}})
# 値の取得・設定・削除
agent.state.set("last_action", "login")
print(agent.state.get("last_action"))
agent.state.delete("last_action")
Agent Stateはツール内からも ToolContext 経由でアクセスできるため、ツール間でのデータ共有に適しています。
from strands import Agent, tool, ToolContext
@tool(context=True)
def track_action(action: str, tool_context: ToolContext):
"""ユーザーのアクションを記録します。
Args:
action: 記録するアクション
"""
count = tool_context.agent.state.get("action_count") or 0
tool_context.agent.state.set("action_count", count + 1)
tool_context.agent.state.set("last_action", action)
return f"アクション '{action}' を記録。合計: {count + 1}"
6-4. Request State
単一リクエストのイベントループ内でのみ有効な状態です。コールバックハンドラ間でのデータ共有に使います。
7. ハンズオン：Hookでツール使用を制限する
ここからは実際にコードを動かします。 ツールの呼び出し回数を制限するHook を実装し、暴走するエージェントを制御する方法を学びます。
7-1. セットアップ
GitHubリポジトリをクローンして始めるのが最も簡単です。
git clone https://github.com/miruky/strands-agents-hooks-demo.git
cd strands-agents-hooks-demo
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
デフォルトのモデルプロバイダーはAmazon Bedrock（Claude Sonnet 4、us-west-2）です。AWSクレデンシャルの設定と、Bedrockモデルアクセスの有効化が必要です。
7-2. 制限なしの場合の問題
まず、制限なしのエージェントを見てみましょう。ここではデモ用にモックツールを定義します。
from strands import Agent, tool
@tool
def check_server(target: str) -> str:
"""サーバーの状態を確認します。
Args:
target: 確認対象（例: "cpu", "memory", "disk", "network", "processes"）
"""
mock_responses = {
"cpu": "CPU使用率: 45.2%, コア数: 8, ロードアベレージ: 1.23, 2.05, 1.87",
"memory": "メモリ使用率: 72.1%, 合計: 16GB, 使用中: 11.5GB, 空き: 4.5GB",
"disk": "ディスク使用率: 58.3%, 合計: 512GB, 使用中: 298GB, 空き: 214GB",
"network": "ネットワーク: eth0 up, RX: 1.2GB, TX: 890MB, 接続数: 142",
"processes": "プロセス数: 287, Running: 3, Sleeping: 282, Zombie: 2",
}
return mock_responses.get(target.lower(), f"'{target}' の情報: 正常稼働中")
@tool
def run_diagnostic(command: str) -> str:
"""診断コマンドを実行します。
Args:
command: 実行する診断コマンド
"""
return f"コマンド '{command}' の実行結果: 異常なし"
agent = Agent(
system_prompt="あなたはシステム管理者です。",
tools=[check_server, run_diagnostic]
)
agent("サーバーの状態を徹底的に調査してください")
このエージェントはツールを何十回も呼び出す可能性があります。「徹底的に調査」という曖昧な指示に対して、エージェントはあらゆる対象を繰り返しチェックし、診断コマンドを際限なく試行します。各呼び出しにはモデルの推論コストがかかり、実行時間も伸びます。
7-3. ツール使用制限Hookの実装
公式ドキュメントのCookbookにある LimitToolCounts パターンを参考に実装します。
from threading import Lock
from strands.hooks import (
HookProvider,
HookRegistry,
BeforeInvocationEvent,
BeforeToolCallEvent,
AfterToolCallEvent,
)
class ToolUsageGuard(HookProvider):
"""ツールの呼び出し回数を制限し、使用状況を記録するHook"""
def __init__(self, max_counts: dict[str, int], default_max: int = 10):
self.max_counts = max_counts
self.default_max = default_max
self.tool_counts: dict[str, int] = {}
self.total_calls = 0
self._lock = Lock()
def register_hooks(self, registry: HookRegistry) -> None:
registry.add_callback(BeforeInvocationEvent, self.reset_counts)
registry.add_callback(BeforeToolCallEvent, self.check_limit)
registry.add_callback(AfterToolCallEvent, self.log_result)
def reset_counts(self, event: BeforeInvocationEvent) -> None:
with self._lock:
self.tool_counts = {}
self.total_calls = 0
def check_limit(self, event: BeforeToolCallEvent) -> None:
tool_name = event.tool_use["name"]
with self._lock:
count = self.tool_counts.get(tool_name, 0) + 1
self.tool_counts[tool_name] = count
max_count = self.max_counts.get(tool_name, self.default_max)
if count > max_count:
event.cancel_tool = (
f"ツール '{tool_name}' は上限 {max_count} 回に達しました。"
f"これ以上このツールを呼び出さないでください。"
f"現在の情報で回答を生成してください。"
)
def log_result(self, event: AfterToolCallEvent) -> None:
with self._lock:
self.total_calls += 1
status = event.result.get("status", "unknown")
print(f" [{self.total_calls}] {event.tool_use['name']}: {status}")
def get_summary(self) -> dict:
return {
"total_calls": self.total_calls,
"per_tool": dict(self.tool_counts),
}
7-4. 使用例
# check_serverは3回まで、run_diagnosticは2回まで、その他は5回までに制限
guard = ToolUsageGuard(
max_counts={
"check_server": 3,
"run_diagnostic": 2,
},
default_max=5,
)
agent = Agent(
system_prompt="あなたはシステム管理者です。",
tools=[check_server, run_diagnostic],
hooks=[guard],
)
result = agent("サーバーの状態を徹底的に調査してください")
# 使用状況を出力
print(f"\n使用状況: {guard.get_summary()}")
check_server が3回、 run_diagnostic が2回を超えた時点で cancel_tool が発動し、モデルは「これ以上呼び出さないでください」というフィードバックを受け取ります。モデルはそれまでに得た情報で最終応答を生成します。
7-5. ポイント
- cancel_tool に設定した文字列は、モデルへのフィードバックとしてそのまま渡されます。明確に「呼び出すな」と指示することが重要です
- reset_counts で毎リクエスト開始時にカウントをリセットしています。これにより、同じエージェントインスタンスを再利用できます
- Lock を使ってスレッドセーフにしています。並行ツール実行（デフォルト動作）でも安全です
cancel_tool はツール実行をスキップしますが、モデルが指示に従わず再度同じツールを呼び出そうとする可能性もあります。より厳密な制御が必要な場合は、回数超過後にツールを selected_tool で無害なダミーツールに差し替えるアプローチも検討してください。
おわりに
ここまでお読みいただきありがとうございます。
Strands Agents SDKは「3行で動く」手軽さの裏側に、 Hooks 、 Plugin 、 Steering という3つの制御メカニズムを備えています。Hooksはエージェントループのあらゆるポイントに型安全に介入でき、Pluginはそれらを再利用可能なパッケージにまとめ、Steeringはモデルの振る舞いをリアルタイムで誘導します。
前回の記事が「Strands Agentsで何ができるか」を示すものだったとすれば、本記事は「Strands Agentsをどう制御するか」を示すものです。エージェントの自律性を活かしつつ、暴走させないための仕組みをぜひ活用してみてください。
ではまた、お会いしましょう。
参考リンク
Strands Agents SDK 公式
- Agent Loop - Strands Agents Docs
- Hooks - Strands Agents Docs
- Plugins - Strands Agents Docs
- Steering - Strands Agents Docs
- Skills - Strands Agents Docs
- State Management - Strands Agents Docs