* Cutting of agentic loop based on max_iterations configured.

src/strands/agent/agent.py

@@ -146,6 +146,7 @@ def __init__(
         tool_executor: ToolExecutor | None = None,
         retry_strategy: ModelRetryStrategy | _DefaultRetryStrategySentinel | None = _DEFAULT_RETRY_STRATEGY,
         concurrent_invocation_mode: ConcurrentInvocationMode = ConcurrentInvocationMode.THROW,
+        max_iterations: int | None = None,
     ):
         """Initialize the Agent with the specified configuration.
 
@@ -214,9 +215,13 @@ def __init__(
                 Set to "unsafe_reentrant" to skip lock acquisition entirely, allowing concurrent invocations.
                 Warning: "unsafe_reentrant" makes no guarantees about resulting behavior and is provided
                 only for advanced use cases where the caller understands the risks.
+            max_iterations: Maximum number of model invocation cycles allowed per
+                invocation. When the limit is exceeded the event loop is terminated and a
+                ``MaxIterationsReachedException`` is raised, protecting against runaway tool-calling loops
+                and unbounded latency/cost. ``None`` (the default) disables the limit.
 
         Raises:
-            ValueError: If agent id contains path separators.
+            ValueError: If agent id contains path separators or if ``max_iterations`` is not a positive integer.
         """
         self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
         self.messages = messages if messages is not None else []
@@ -265,6 +270,12 @@ def __init__(
         self.record_direct_tool_call = record_direct_tool_call
         self.load_tools_from_directory = load_tools_from_directory
 
+        if max_iterations is not None and (
+            isinstance(max_iterations, bool) or not isinstance(max_iterations, int) or max_iterations <= 0
+        ):
+            raise ValueError("max_iterations must be a positive integer or None")
+        self.max_iterations = max_iterations
+
         # Create internal cancel signal for graceful cancellation using threading.Event
         self._cancel_signal = threading.Event()
 

src/strands/event_loop/event_loop.py

@@ -36,6 +36,7 @@
 from ..types.exceptions import (
     ContextWindowOverflowException,
     EventLoopException,
+    MaxIterationsReachedException,
     MaxTokensReachedException,
     StructuredOutputException,
 )
@@ -161,6 +162,24 @@ async def event_loop_cycle(
     # Initialize cycle state
     invocation_state["event_loop_cycle_id"] = uuid.uuid4()
 
+    # Track number of cycles executed in this invocation. Incremented here so the very first
+    # cycle is iteration 1.
+    iteration = invocation_state.get("event_loop_iteration", 0) + 1
+    invocation_state["event_loop_iteration"] = iteration
+
+    # Enforce the optional max_iterations limit configured on the agent. We check before any
+    # model invocation so an over-limit cycle never makes an additional request to the model.
+    max_iterations = getattr(agent, "max_iterations", None)
+    if max_iterations is not None and iteration > max_iterations:
+        raise MaxIterationsReachedException(
+            (
+                f"Agent reached the configured max_iterations limit of {max_iterations} before producing a final "
+                "response. Increase max_iterations or inspect the conversation history to diagnose the loop."
+            ),
+            iterations=iteration - 1,
+            max_iterations=max_iterations,
+        )
+
     # Initialize state and get cycle trace
     if "request_state" not in invocation_state:
         invocation_state["request_state"] = {}
@@ -270,6 +289,7 @@ async def event_loop_cycle(
             EventLoopException,
             ContextWindowOverflowException,
             MaxTokensReachedException,
+            MaxIterationsReachedException,
         ) as e:
             # These exceptions should bubble up directly rather than get wrapped in an EventLoopException
             tracer.end_span_with_error(cycle_span, str(e), e)

src/strands/types/exceptions.py

@@ -35,6 +35,28 @@ def __init__(self, message: str):
         super().__init__(message)
 
 
+class MaxIterationsReachedException(Exception):
+    """Exception raised when the agent reaches its configured ``max_iterations`` limit.
+
+    The event loop tracks the number of "model turn -> tool call -> tool result" cycles executed during a single
+    agent invocation. When this counter exceeds the limit configured on the agent, the loop is terminated
+    and this exception is raised so callers can handle the situation gracefully (e.g. return a partial response,
+    surface a timeout to the user, or apply a retry policy with a different strategy).
+    """
+
+    def __init__(self, message: str, *, iterations: int | None = None, max_iterations: int | None = None) -> None:
+        """Initialize the exception.
+
+        Args:
+            message: The error message describing the iteration limit.
+            iterations: Number of cycles that were executed before the limit was hit.
+            max_iterations: The configured maximum iteration limit.
+        """
+        self.iterations = iterations
+        self.max_iterations = max_iterations
+        super().__init__(message)
+
+
 class ContextWindowOverflowException(Exception):
     """Exception raised when the context window is exceeded.
 

tests/strands/agent/hooks/test_agent_events.py

@@ -43,6 +43,7 @@ def mock_sleep():
     "event_loop_cycle_id": ANY,
     "event_loop_cycle_span": ANY,
     "event_loop_cycle_trace": ANY,
+    "event_loop_iteration": ANY,
     "request_state": {},
 }
 

tests/strands/agent/test_agent.py

@@ -29,7 +29,7 @@
 from strands.types._events import EventLoopStopEvent, ModelStreamEvent
 from strands.types.agent import ConcurrentInvocationMode
 from strands.types.content import Messages
-from strands.types.exceptions import ConcurrencyException, ContextWindowOverflowException, EventLoopException
+from strands.types.exceptions import ConcurrencyException, ContextWindowOverflowException, EventLoopException, MaxIterationsReachedException
 from strands.types.session import Session, SessionAgent, SessionMessage, SessionType
 from tests.fixtures.mock_session_repository import MockedSessionRepository
 from tests.fixtures.mocked_model_provider import MockedModelProvider
@@ -291,6 +291,22 @@ def test_agent__init__invalid_id(agent_id):
         Agent(agent_id=agent_id)
 
 
+def test_agent__init__max_iterations_defaults_to_none():
+    agent = Agent()
+    assert agent.max_iterations is None
+
+
+def test_agent__init__max_iterations_accepts_positive_int():
+    agent = Agent(max_iterations=5)
+    assert agent.max_iterations == 5
+
+
+@pytest.mark.parametrize("invalid_value", [0, -1, 1.5, "5", True])
+def test_agent__init__max_iterations_rejects_invalid_values(invalid_value):
+    with pytest.raises(ValueError, match="max_iterations must be a positive integer or None"):
+        Agent(max_iterations=invalid_value)
+
+
 def test_agent__call__(
     mock_model,
     system_prompt,
@@ -734,6 +750,7 @@ def test_agent__call__callback(mock_model, agent, callback_handler, agenerator):
             event_loop_cycle_id=unittest.mock.ANY,
             event_loop_cycle_span=unittest.mock.ANY,
             event_loop_cycle_trace=unittest.mock.ANY,
+            event_loop_iteration=unittest.mock.ANY,
             request_state={},
         ),
         unittest.mock.call(event={"contentBlockStop": {}}),
@@ -745,6 +762,7 @@ def test_agent__call__callback(mock_model, agent, callback_handler, agenerator):
             event_loop_cycle_id=unittest.mock.ANY,
             event_loop_cycle_span=unittest.mock.ANY,
             event_loop_cycle_trace=unittest.mock.ANY,
+            event_loop_iteration=unittest.mock.ANY,
             reasoning=True,
             reasoningText="value",
             request_state={},
@@ -756,6 +774,7 @@ def test_agent__call__callback(mock_model, agent, callback_handler, agenerator):
             event_loop_cycle_id=unittest.mock.ANY,
             event_loop_cycle_span=unittest.mock.ANY,
             event_loop_cycle_trace=unittest.mock.ANY,
+            event_loop_iteration=unittest.mock.ANY,
             reasoning=True,
             reasoning_signature="value",
             request_state={},
@@ -770,6 +789,7 @@ def test_agent__call__callback(mock_model, agent, callback_handler, agenerator):
             event_loop_cycle_id=unittest.mock.ANY,
             event_loop_cycle_span=unittest.mock.ANY,
             event_loop_cycle_trace=unittest.mock.ANY,
+            event_loop_iteration=unittest.mock.ANY,
             request_state={},
         ),
         unittest.mock.call(event={"contentBlockStop": {}}),
@@ -2800,3 +2820,210 @@ def test_as_tool_defaults_description_when_agent_has_none():
     tool = agent.as_tool()
 
     assert tool.tool_spec["description"] == "Use the researcher agent as a tool by providing a natural language input"
+
+
+def test_agent_max_iterations_raises_on_runaway_tool_loop():
+    """Agent stops with MaxIterationsReachedException when tool-calling loops past the configured limit."""
+
+    @strands.tools.tool
+    def echo(value: str) -> str:
+        return value
+
+    tool_use_response = {
+        "role": "assistant",
+        "content": [
+            {
+                "toolUse": {
+                    "toolUseId": "loop",
+                    "name": "echo",
+                    "input": {"value": "again"},
+                }
+            }
+        ],
+    }
+    # Repeatedly request the same tool to simulate a runaway agent.
+    mocked_model = MockedModelProvider([tool_use_response, tool_use_response, tool_use_response, tool_use_response])
+
+    agent = Agent(
+        model=mocked_model,
+        tools=[echo],
+        callback_handler=None,
+        max_iterations=2,
+    )
+
+    with pytest.raises(MaxIterationsReachedException) as exc_info:
+        agent("loop forever")
+
+    assert exc_info.value.max_iterations == 2
+    assert exc_info.value.iterations == 2
+
+
+def test_agent_max_iterations_allows_completion_within_limit():
+    """Agent should complete normally when it produces a final answer within the iteration budget."""
+
+    @strands.tools.tool
+    def echo(value: str) -> str:
+        return value
+
+    tool_use_response = {
+        "role": "assistant",
+        "content": [
+            {
+                "toolUse": {
+                    "toolUseId": "t1",
+                    "name": "echo",
+                    "input": {"value": "hi"},
+                }
+            }
+        ],
+    }
+    final_response = {"role": "assistant", "content": [{"text": "all done"}]}
+    mocked_model = MockedModelProvider([tool_use_response, final_response])
+
+    agent = Agent(
+        model=mocked_model,
+        tools=[echo],
+        callback_handler=None,
+        max_iterations=5,
+    )
+
+    result = agent("do the thing")
+
+    assert result.stop_reason == "end_turn"
+    assert result.message["content"][0]["text"] == "all done"
+
+
+def test_agent_max_iterations_counter_resets_between_invocations():
+    """The per-invocation iteration counter must not leak across separate agent calls."""
+
+    responses = [
+        {"role": "assistant", "content": [{"text": "first"}]},
+        {"role": "assistant", "content": [{"text": "second"}]},
+    ]
+    mocked_model = MockedModelProvider(responses)
+
+    agent = Agent(model=mocked_model, callback_handler=None, max_iterations=1)
+
+    result1 = agent("hello")
+    result2 = agent("again")
+
+    assert result1.message["content"][0]["text"] == "first"
+    assert result2.message["content"][0]["text"] == "second"
+
+
+def test_agent_max_iterations_none_does_not_raise():
+    """max_iterations=None (the default) must never raise regardless of tool-call depth."""
+
+    @strands.tools.tool
+    def echo(value: str) -> str:
+        return value
+
+    tool_use_response = {
+        "role": "assistant",
+        "content": [
+            {
+                "toolUse": {
+                    "toolUseId": "t1",
+                    "name": "echo",
+                    "input": {"value": "hi"},
+                }
+            }
+        ],
+    }
+    final_response = {"role": "assistant", "content": [{"text": "done"}]}
+    mocked_model = MockedModelProvider(
+        [tool_use_response, tool_use_response, tool_use_response, tool_use_response, tool_use_response, final_response]
+    )
+
+    agent = Agent(model=mocked_model, tools=[echo], callback_handler=None)
+
+    result = agent("run many cycles")
+
+    assert result.stop_reason == "end_turn"
+    assert result.message["content"][0]["text"] == "done"
+
+
+def test_agent_max_iterations_exception_message_is_informative():
+    """MaxIterationsReachedException message should contain the configured limit."""
+
+    @strands.tools.tool
+    def echo(value: str) -> str:
+        return value
+
+    tool_use_response = {
+        "role": "assistant",
+        "content": [
+            {
+                "toolUse": {
+                    "toolUseId": "t1",
+                    "name": "echo",
+                    "input": {"value": "hi"},
+                }
+            }
+        ],
+    }
+    mocked_model = MockedModelProvider([tool_use_response, tool_use_response, tool_use_response])
+
+    agent = Agent(model=mocked_model, tools=[echo], callback_handler=None, max_iterations=1)
+
+    with pytest.raises(MaxIterationsReachedException, match="max_iterations limit of 1"):
+        agent("keep going")
+
+
+def test_agent_max_iterations_raises_on_second_cycle_when_limit_is_one():
+    """With max_iterations=1, an agent that calls a tool raises on the second cycle attempt."""
+
+    @strands.tools.tool
+    def echo(value: str) -> str:
+        return value
+
+    tool_use_response = {
+        "role": "assistant",
+        "content": [
+            {
+                "toolUse": {
+                    "toolUseId": "t1",
+                    "name": "echo",
+                    "input": {"value": "hi"},
+                }
+            }
+        ],
+    }
+    final_response = {"role": "assistant", "content": [{"text": "done"}]}
+    mocked_model = MockedModelProvider([tool_use_response, final_response])
+
+    agent = Agent(model=mocked_model, tools=[echo], callback_handler=None, max_iterations=1)
+
+    with pytest.raises(MaxIterationsReachedException) as exc_info:
+        agent("one cycle only")
+
+    assert exc_info.value.iterations == 1
+    assert exc_info.value.max_iterations == 1
+
+
+@pytest.mark.asyncio
+async def test_agent_max_iterations_raises_in_async_invocation():
+    """MaxIterationsReachedException must propagate through invoke_async as well."""
+
+    @strands.tools.tool
+    def echo(value: str) -> str:
+        return value
+
+    tool_use_response = {
+        "role": "assistant",
+        "content": [
+            {
+                "toolUse": {
+                    "toolUseId": "t1",
+                    "name": "echo",
+                    "input": {"value": "hi"},
+                }
+            }
+        ],
+    }
+    mocked_model = MockedModelProvider([tool_use_response, tool_use_response, tool_use_response])
+
+    agent = Agent(model=mocked_model, tools=[echo], callback_handler=None, max_iterations=1)
+
+    with pytest.raises(MaxIterationsReachedException):
+        await agent.invoke_async("loop")