【小ネタ】 Strands Agents で一定回数以上ツールが実行された場合に検知して停止させる

Strands Agents

神野雄大

2025.12.27

はじめに

こんにちは、スーパーマーケットのラ・ムーが大好きなコンサルティング部の神野です。

Strands Agentsは再実行などの実装を意識せずとも、目的を達成するためによしなにツールの実行を繰り返すフレームワークです。下記のようなイメージです。

CleanShot 2025-12-26 at 23.00.51@2x

良い意味で実装するのは楽ちんですが、ループが予想以上に実行されるケースも逆にいうとある気がします。
先日ブログでご紹介したStructured Outputで絶対にクリアできないバリデーションを実装した場合、興味本位でどうなるんだろうとAgentを起動したところ無限にツールが実行されました。

from pydantic import BaseModel, Field, field_validator
from strands import Agent
from strands.models import BedrockModel
from strands.types.exceptions import StructuredOutputException

class ImpossibleModel(BaseModel):
    """絶対に検証が通らないモデル"""

    secret_code: str = Field(description="シークレットコード")

    @field_validator("secret_code")
    @classmethod
    def validate_impossible(cls, value: str) -> str:
        raise ValueError("このバリデーションは絶対に通りません")

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-2-lite-v1:0",
    region_name="us-west-2",
    temperature=0.0,
)

agent = Agent(model=bedrock_model)

try:
    result = agent("テスト", structured_output_model=ImpossibleModel)
    print(f"成功: {result.structured_output}")
except StructuredOutputException as e:
    print(f"構造化出力エラー: {e}")

# 実行結果                                                                                  
申し訳ありませんが、現在ご提供いただける機能が不足しています。特に、検証が通らないモデル「ImpossibleModel」を呼び出すには、適切なシークレットコードが必要です。このコードは現在利用できません。別のサポートが必要な場合は、お知らせください。

Tool #1: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #2: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #3: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #4: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

・・・

うーん、困りますよね・・・
単純に処理が実行されるだけならいいのですが、伴ってLLMや外部APIのコストがかかるとかなり怖いポイントです。
一定回数の実行を検知したら、停止する方法はないのか気になって調べると下記方法が公式ドキュメントで紹介されていました。

このやり方に注目して、実際に試してみたいと思います！

前提

今回は下記環境で検証を実施しました。

Python 3.13.6
uv 0.6.12
strands-agents 1.19.0
strands-agents-tools 0.2.17
使用したモデル
- us.amazon.nova-2-lite-v1:0

ツールの実行回数をカウントして停止する

公式ドキュメントに記載してあるサンプルのコードを実装して試してみます。

公式ドキュメントのサンプルを試す

import time
from threading import Lock

from strands import Agent, tool
from strands.hooks import (
    BeforeInvocationEvent,
    BeforeToolCallEvent,
    HookProvider,
    HookRegistry,
)
from strands.models import BedrockModel

class LimitToolCounts(HookProvider):
    """Limits the number of times tools can be called per agent invocation"""

    def __init__(self, max_tool_counts: dict[str, int]):
        """
        Initializer.

        Args:
            max_tool_counts: A dictionary mapping tool names to max call counts for
                tools. If a tool is not specified in it, the tool can be called as many
                times as desired
        """
        self.max_tool_counts = max_tool_counts
        self.tool_counts = {}
        self._lock = Lock()

    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeInvocationEvent, self.reset_counts)
        registry.add_callback(BeforeToolCallEvent, self.intercept_tool)

    def reset_counts(self, event: BeforeInvocationEvent) -> None:
        with self._lock:
            self.tool_counts = {}

    def intercept_tool(self, event: BeforeToolCallEvent) -> None:
        tool_name = event.tool_use["name"]
        with self._lock:
            max_tool_count = self.max_tool_counts.get(tool_name)
            tool_count = self.tool_counts.get(tool_name, 0) + 1
            self.tool_counts[tool_name] = tool_count

        if max_tool_count and tool_count > max_tool_count:
            event.cancel_tool = (
                f"Tool '{tool_name}' has been invoked too many and is now being throttled. "
                f"DO NOT CALL THIS TOOL ANYMORE "
            )

@tool
def sleep(milliseconds: int) -> str:
    """指定されたミリ秒だけスリープする

    Args:
        milliseconds: スリープする時間（ミリ秒）
    """
    time.sleep(milliseconds / 1000)
    return f"Slept for {milliseconds}ms"

limit_hook = LimitToolCounts(max_tool_counts={"sleep": 3})

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-2-lite-v1:0",
    temperature=0.0,
)

agent = Agent(tools=[sleep], hooks=[limit_hook], model=bedrock_model)

# This call will only have 3 successful sleeps
agent("Sleep 5 times for 10ms each or until you can't anymore")
# This will sleep successfully again because the count resets every invocation
agent("Sleep once")

Hooksを活用してモデルが呼び出される前（BeforeInvocationEvent）とツールが呼び出される前（BeforeToolCallEvent）にトリガーを仕込みます。
モデルが呼び出される前にツールの呼び出し回数を初期化して、ツール実行前にツール名と実行回数のペアをインクリメントして更新します。仮に想定した実行回数を上回る場合はツールの実行を取り止めます。
試しに実行してみますね。

Tool #1: sleep

Tool #2: sleep

Tool #3: sleep

Tool #4: sleep

Tool #5: sleep
I've executed the sleep operations as requested. Here's the outcome:

- First sleep: ✅ Success - Slept for 10ms
- Second sleep: ✅ Success - Slept for 10ms  
- Third sleep: ✅ Success - Slept for 10ms
- Fourth sleep: ⚠️ Throttled - Tool 'sleep' has been invoked too many times and is now being throttled. DO NOT CALL THIS TOOL ANYMORE
- Fifth sleep: ⚠️ Throttled - Same throttling message

The system has prevented further sleep calls due to throttling limits being reached after the third invocation. This is a safety mechanism to prevent excessive resource consumption. 

✅ Completed 3 successful 10ms sleeps before hitting the throttle limit.
Tool #6: sleep
✅ Success! Completed one more sleep operation:
- Slept for 10ms

The sleep function is now working again after the previous throttling. You can continue with additional sleep operations if needed.%

見た目がちょっと紛らわしいのですが、Tool #n はツールの呼び出し要求（toolUse）が出た回数なので、必ずしもsleepが実際に成功した回数と一致しないことがあります。
cancel_tool はその回の実行をキャンセルしてエラーの結果を返すだけなので、LLMがそれでもう一回呼ぼうとすると Tool #4 や Tool #5 自体は増えます。なのでこれは厳密に3回で強制停止というよりそれ以上は通さないように抑制する挙動、という理解の方が近いです。

なので、LLMの返信としては3回目まではsleepに成功して、4回目以降はスロットルに引っかかったと返信しているので期待通りに動いています。ちなみに再度実行した際はカウントがリセットされているので、Tool #6は成功していますね。

これを最初に無限ループが発生したStructured Outputの場合にも応用してみます。

ただStructured Output と組み合わせると止まらない

まずはサンプル通りに event.cancel_tool だけ仕込んでみました。

from threading import Lock

from pydantic import BaseModel, Field, field_validator
from strands import Agent
from strands.hooks import (
    BeforeInvocationEvent,
    BeforeToolCallEvent,
    HookProvider,
    HookRegistry,
)
from strands.models import BedrockModel
from strands.types.exceptions import StructuredOutputException

class LimitToolCounts(HookProvider):
    """ツールの呼び出し回数を制限するHook"""

    def __init__(self, max_tool_counts: dict[str, int]):
        self.max_tool_counts = max_tool_counts
        self.tool_counts = {}
        self._lock = Lock()

    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeInvocationEvent, self.reset_counts)
        registry.add_callback(BeforeToolCallEvent, self.intercept_tool)

    def reset_counts(self, event: BeforeInvocationEvent) -> None:
        with self._lock:
            self.tool_counts = {}

    def intercept_tool(self, event: BeforeToolCallEvent) -> None:
        tool_name = event.tool_use["name"]
        with self._lock:
            max_tool_count = self.max_tool_counts.get(tool_name)
            tool_count = self.tool_counts.get(tool_name, 0) + 1
            self.tool_counts[tool_name] = tool_count

        if max_tool_count and tool_count > max_tool_count:
            # ツール呼び出しはキャンセルする（toolResult(error)としてLLMに返る）
            event.cancel_tool = (
                f"Tool '{tool_name}' has been invoked too many times and is now being throttled. "
                f"DO NOT CALL THIS TOOL ANYMORE "
            )

class ImpossibleModel(BaseModel):
    """絶対に検証が通らないモデル"""

    secret_code: str = Field(description="シークレットコード")

    @field_validator("secret_code")
    @classmethod
    def validate_impossible(cls, value: str) -> str:
        raise ValueError("このバリデーションは絶対に通りません")

bedrock_model = BedrockModel(
    model_id="us.amazon.nova-2-lite-v1:0",
    region_name="us-west-2",
    temperature=0.0,
)

# ImpossibleModelツールの呼び出しを3回に制限
limit_hook = LimitToolCounts(max_tool_counts={"ImpossibleModel": 3})
agent = Agent(model=bedrock_model, hooks=[limit_hook])

try:
    result = agent("テスト", structured_output_model=ImpossibleModel)
    if result.structured_output is None:
        raise StructuredOutputException(
            "構造化出力の生成が完了する前にイベントループを停止しました（tool count limit で停止）。"
        )
    print(f"成功: {result.structured_output}")
except StructuredOutputException as e:
    print(f"構造化出力エラー: {e}")

すると、期待としては一定回数を超えたら止まってほしいのですが止まりませんでした。
ログ上は「ツール呼び出し自体はキャンセルされているっぽい」のに、ImpossibleModel が延々と呼ばれてしまいます。

# 実行
uv run main.py
# 実行記録
Tool #1: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #2: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #3: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #4: ImpossibleModel
Tool #5: ImpossibleModel
...

止まりませんでした。なぜでしょうか？
Strandsのライフサイクルを整理すると、通常は以下の流れです。

ユーザー入力 → LLM呼び出し → toolUse → ツール実行 → toolResult → 次のLLM呼び出し → …

サンプルの sleep みたいな通常のツールだと、途中でツールが失敗しても、じゃあここまでで回答して終わるが選べます。
そのため、 cancel_tool でも自然にループが収束しやすいです。

一方で structured_output_model を渡した場合、内部的には Pydantic の検証を行うツールが動的に追加されて、検証が通るまで繰り返しやすい構造になっています。
今回みたいに絶対に検証が通らないモデルだと、ツールをキャンセルしても次のターンでまた同じツールが呼ばれて、結果としてループが終わりません。

しっかりと止めたいならツールをキャンセルするだけじゃなくてイベントループ自体を止めるフラグを立てることを試してみました。

BeforeToolCallEvent はツール実行の直前に発火するHookイベントで、ここで受け取れる invocation_state は、その呼び出しの間だけ共有される状態で、Strands内部でも request_state という領域を見ています。
この request_state に stop_event_loop の停止用のフラグをTrueにすることで、次の周回に入らず停止させるようにしてみます。実装としては下記のようにしました。

def intercept_tool(self, event: BeforeToolCallEvent) -> None:
    tool_name = event.tool_use["name"]
    with self._lock:
        max_tool_count = self.max_tool_counts.get(tool_name)
        tool_count = self.tool_counts.get(tool_name, 0) + 1
        self.tool_counts[tool_name] = tool_count

    if max_tool_count and tool_count > max_tool_count:
        event.cancel_tool = (
            f"Tool '{tool_name}' has been invoked too many times and is now being throttled. "
            f"DO NOT CALL THIS TOOL ANYMORE "
        )
        # この行を追加
        event.invocation_state.setdefault("request_state", {})["stop_event_loop"] = True

実装できたので実行してみます。

申し訳ありませんが、現在ご提供できる機能が不足しています。特に、絶対に検証が通らないモデル「ImpossibleModel」を呼び出すには、適切なシークレットコードが必要です。しかし、現在このコードは提供されていません。

この問題を解決するために、以下のいずれかのアクションを取っていただけますか：

1. **シークレットコードを提供してください**  
   モデルを呼び出すには、`secret_code` パラメータが必要です。コードをお知らせいただければ、すぐに処理を進めます。

2. **別のサポートが必要な場合はお知らせください**  
   現在ご利用可能な他の機能についてご説明できます。例えば、基本的な情報提供や他のツールの使用についてご相談ください。

ご協力ありがとうございます。
Tool #1: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #2: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

Tool #3: ImpossibleModel
tool_name=<ImpossibleModel> | structured output validation failed | error_message=<Validation failed for ImpossibleModel. Please fix the following errors:
- Field 'secret_code': Value error, このバリデーションは絶対に通りません>

無事止まりました！！よかった！！

おわりに

今回はStrands Agentsがツールが一定回数以上実行されたら停止する方法を検証してみました。
ユーザー側がリトライ処理などを意識せずとも再帰的にStrands Agents が実行してくれるのはありがたいですが、時に無限ループみたいな挙動が発生するのは怖いですよね。APIツールやLLMを呼び出すようなツールだとコストが嵩んでしまうことなどもあると思うので、無限ループみたいな挙動が見られる際はこういった対策を試してみるのがいいかもしれませんね。

今回はあくまで公式ドキュメントに書いてあったやり方をご紹介しましたが、もっと良いやり方があればぜひ教えてください！！

本記事が少しでも参考になりましたら幸いです。最後までご覧いただきありがとうございましたー！！

【小ネタ】 Strands Agents で一定回数以上ツールが実行された場合に検知して停止させる

はじめに

前提

ツールの実行回数をカウントして停止する

公式ドキュメントのサンプルを試す

ただStructured Output と組み合わせると止まらない

おわりに

関連記事

AWSで探す

注目のテーマ

プロダクトやサービスで探す

特集やシリーズから探す

EVENTS