diff --git a/src/strands/agent/agent.py b/src/strands/agent/agent.py
index 965969961..c23f4090e 100644
--- a/src/strands/agent/agent.py
+++ b/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()
 
diff --git a/src/strands/event_loop/event_loop.py b/src/strands/event_loop/event_loop.py
index 128ef9ca3..6d489d8e9 100644
--- a/src/strands/event_loop/event_loop.py
+++ b/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)
diff --git a/src/strands/types/exceptions.py b/src/strands/types/exceptions.py
index 7ad49eb24..1c3738dcd 100644
--- a/src/strands/types/exceptions.py
+++ b/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.
 
diff --git a/tests/strands/agent/hooks/test_agent_events.py b/tests/strands/agent/hooks/test_agent_events.py
index 1f09579b0..3313580f8 100644
--- a/tests/strands/agent/hooks/test_agent_events.py
+++ b/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": {},
 }
 
diff --git a/tests/strands/agent/test_agent.py b/tests/strands/agent/test_agent.py
index 680a1d23c..a02883c4d 100644
--- a/tests/strands/agent/test_agent.py
+++ b/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")
diff --git a/tests/strands/event_loop/test_event_loop.py b/tests/strands/event_loop/test_event_loop.py
index f025a81ef..929ab16a3 100644
--- a/tests/strands/event_loop/test_event_loop.py
+++ b/tests/strands/event_loop/test_event_loop.py
@@ -29,6 +29,7 @@
 from strands.types.exceptions import (
     ContextWindowOverflowException,
     EventLoopException,
+    MaxIterationsReachedException,
     MaxTokensReachedException,
     ModelThrottledException,
 )
@@ -159,6 +160,7 @@ def agent(model, system_prompt, messages, tool_registry, thread_pool, hook_regis
     mock._model_state = {}
     mock.trace_attributes = {}
     mock.retry_strategy = ModelRetryStrategy()
+    mock.max_iterations = None
 
     return mock
 
@@ -826,6 +828,7 @@ async def test_request_state_initialization(alist):
     mock_agent._cancel_signal = threading.Event()
     mock_agent.event_loop_metrics.start_cycle.return_value = (0, MagicMock())
     mock_agent.hooks.invoke_callbacks_async = AsyncMock()
+    mock_agent.max_iterations = None
 
     # Call without providing request_state
     stream = strands.event_loop.event_loop.event_loop_cycle(
@@ -1279,3 +1282,142 @@ async def test_error_fallback_returns_none_at_call_site(self):
 
         with pytest.raises(Exception, match="API unavailable"):
             await strands.event_loop.event_loop._estimate_input_tokens(agent)
+
+
+class TestMaxIterations:
+    """Unit tests for the max_iterations guard inside event_loop_cycle."""
+
+    @pytest.mark.asyncio
+    async def test_raises_when_iteration_exceeds_limit(self, agent, model, agenerator, alist):
+        """event_loop_cycle raises MaxIterationsReachedException when the iteration count exceeds max_iterations."""
+        agent.max_iterations = 2
+        # Pre-seed the counter so this call becomes iteration 3 (> 2).
+        invocation_state = {"event_loop_iteration": 2}
+
+        with pytest.raises(MaxIterationsReachedException) as exc_info:
+            stream = strands.event_loop.event_loop.event_loop_cycle(
+                agent=agent,
+                invocation_state=invocation_state,
+            )
+            await alist(stream)
+
+        assert exc_info.value.iterations == 2
+        assert exc_info.value.max_iterations == 2
+
+    @pytest.mark.asyncio
+    async def test_check_occurs_before_model_call(self, agent, model, agenerator, alist):
+        """The limit guard fires before any model call so no extra request is made."""
+        agent.max_iterations = 1
+        # iteration 1 already consumed; next call would be iteration 2 > 1.
+        invocation_state = {"event_loop_iteration": 1}
+
+        with pytest.raises(MaxIterationsReachedException):
+            stream = strands.event_loop.event_loop.event_loop_cycle(
+                agent=agent,
+                invocation_state=invocation_state,
+            )
+            await alist(stream)
+
+        model.stream.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_counter_initialises_to_one_on_first_cycle(self, agent, model, agenerator, alist):
+        """invocation_state['event_loop_iteration'] starts at 1 when no prior value exists."""
+        agent.max_iterations = None
+
+        model.stream.return_value = agenerator(
+            [
+                {"contentBlockDelta": {"delta": {"text": "hi"}}},
+                {"contentBlockStop": {}},
+            ]
+        )
+
+        invocation_state = {}
+        stream = strands.event_loop.event_loop.event_loop_cycle(
+            agent=agent,
+            invocation_state=invocation_state,
+        )
+        await alist(stream)
+
+        assert invocation_state["event_loop_iteration"] == 1
+
+    @pytest.mark.asyncio
+    async def test_counter_increments_each_cycle(self, agent, model, agenerator, alist):
+        """The counter advances by 1 per cycle when called with a pre-existing value."""
+        agent.max_iterations = None
+
+        model.stream.return_value = agenerator(
+            [
+                {"contentBlockDelta": {"delta": {"text": "hi"}}},
+                {"contentBlockStop": {}},
+            ]
+        )
+
+        invocation_state = {"event_loop_iteration": 4}
+        stream = strands.event_loop.event_loop.event_loop_cycle(
+            agent=agent,
+            invocation_state=invocation_state,
+        )
+        await alist(stream)
+
+        assert invocation_state["event_loop_iteration"] == 5
+
+    @pytest.mark.asyncio
+    async def test_none_max_iterations_never_raises(self, agent, model, agenerator, alist):
+        """max_iterations=None disables the guard even at a very high iteration count."""
+        agent.max_iterations = None
+
+        model.stream.return_value = agenerator(
+            [
+                {"contentBlockDelta": {"delta": {"text": "done"}}},
+                {"contentBlockStop": {}},
+            ]
+        )
+
+        invocation_state = {"event_loop_iteration": 9999}
+        stream = strands.event_loop.event_loop.event_loop_cycle(
+            agent=agent,
+            invocation_state=invocation_state,
+        )
+        events = await alist(stream)
+
+        tru_stop_reason, _, _, _, _, _ = events[-1]["stop"]
+        assert tru_stop_reason == "end_turn"
+        assert invocation_state["event_loop_iteration"] == 10000
+
+    @pytest.mark.asyncio
+    async def test_completes_normally_at_exact_limit(self, agent, model, agenerator, alist):
+        """A cycle whose iteration number equals max_iterations is allowed to complete."""
+        agent.max_iterations = 3
+        # iteration 2 already done; this call is iteration 3 == limit (not > limit).
+        invocation_state = {"event_loop_iteration": 2}
+
+        model.stream.return_value = agenerator(
+            [
+                {"contentBlockDelta": {"delta": {"text": "result"}}},
+                {"contentBlockStop": {}},
+            ]
+        )
+
+        stream = strands.event_loop.event_loop.event_loop_cycle(
+            agent=agent,
+            invocation_state=invocation_state,
+        )
+        events = await alist(stream)
+
+        tru_stop_reason, _, _, _, _, _ = events[-1]["stop"]
+        assert tru_stop_reason == "end_turn"
+        assert invocation_state["event_loop_iteration"] == 3
+
+    @pytest.mark.asyncio
+    async def test_exception_message_contains_limit(self, agent, model, agenerator, alist):
+        """The exception message includes the configured max_iterations value."""
+        agent.max_iterations = 5
+        invocation_state = {"event_loop_iteration": 5}
+
+        with pytest.raises(MaxIterationsReachedException, match="max_iterations limit of 5"):
+            stream = strands.event_loop.event_loop.event_loop_cycle(
+                agent=agent,
+                invocation_state=invocation_state,
+            )
+            await alist(stream)
diff --git a/tests/strands/event_loop/test_event_loop_metadata.py b/tests/strands/event_loop/test_event_loop_metadata.py
index e6fe97f39..7c8819f09 100644
--- a/tests/strands/event_loop/test_event_loop_metadata.py
+++ b/tests/strands/event_loop/test_event_loop_metadata.py
@@ -57,6 +57,7 @@ def agent(model, messages, tool_registry, hook_registry):
     mock._cancel_signal = threading.Event()
     mock.trace_attributes = {}
     mock.retry_strategy = ModelRetryStrategy()
+    mock.max_iterations = None
     return mock
 
 
diff --git a/tests/strands/event_loop/test_event_loop_structured_output.py b/tests/strands/event_loop/test_event_loop_structured_output.py
index 2d1150712..de9912d52 100644
--- a/tests/strands/event_loop/test_event_loop_structured_output.py
+++ b/tests/strands/event_loop/test_event_loop_structured_output.py
@@ -54,6 +54,7 @@ def mock_agent():
     agent.trace_attributes = {}
     agent.tool_executor = Mock()
     agent._append_messages = AsyncMock()
+    agent.max_iterations = None
 
     # Set up _interrupt_state properly
     agent._interrupt_state = Mock()
diff --git a/tests/strands/types/test_exceptions.py b/tests/strands/types/test_exceptions.py
index 29f68a7d0..6a8ee5b61 100644
--- a/tests/strands/types/test_exceptions.py
+++ b/tests/strands/types/test_exceptions.py
@@ -5,6 +5,7 @@
 from strands.types.exceptions import (
     ContextWindowOverflowException,
     EventLoopException,
+    MaxIterationsReachedException,
     MaxTokensReachedException,
     MCPClientInitializationError,
     ModelThrottledException,
@@ -100,6 +101,77 @@ def test_exception_raised_properly(self):
         assert str(exc_info.value) == "Token limit exceeded"
 
 
+class TestMaxIterationsReachedException:
+    """Tests for MaxIterationsReachedException class."""
+
+    def test_initialization_with_message_only(self):
+        """Test initialization with only a message; iteration attrs default to None."""
+        message = "Agent reached the max_iterations limit"
+        exception = MaxIterationsReachedException(message)
+
+        assert str(exception) == message
+        assert exception.args[0] == message
+        assert exception.iterations is None
+        assert exception.max_iterations is None
+
+    def test_initialization_with_all_args(self):
+        """Test initialization with message, iterations, and max_iterations."""
+        message = "Limit reached"
+        exception = MaxIterationsReachedException(message, iterations=5, max_iterations=5)
+
+        assert str(exception) == message
+        assert exception.iterations == 5
+        assert exception.max_iterations == 5
+
+    def test_iterations_and_max_iterations_are_keyword_only(self):
+        """iterations and max_iterations must be passed as keyword arguments."""
+        with pytest.raises(TypeError):
+            MaxIterationsReachedException("msg", 3, 5)  # type: ignore[call-arg]
+
+    def test_iterations_defaults_to_none(self):
+        """iterations attribute defaults to None when not provided."""
+        exception = MaxIterationsReachedException("msg", max_iterations=10)
+
+        assert exception.iterations is None
+
+    def test_max_iterations_defaults_to_none(self):
+        """max_iterations attribute defaults to None when not provided."""
+        exception = MaxIterationsReachedException("msg", iterations=3)
+
+        assert exception.max_iterations is None
+
+    def test_inheritance(self):
+        """Test that MaxIterationsReachedException inherits from Exception."""
+        exception = MaxIterationsReachedException("Test message")
+
+        assert isinstance(exception, Exception)
+        assert issubclass(MaxIterationsReachedException, Exception)
+
+    def test_exception_raised_and_caught_properly(self):
+        """Test that the exception can be raised and caught, preserving all attributes."""
+        with pytest.raises(MaxIterationsReachedException) as exc_info:
+            raise MaxIterationsReachedException(
+                "Agent looped too many times",
+                iterations=7,
+                max_iterations=7,
+            )
+
+        assert str(exc_info.value) == "Agent looped too many times"
+        assert exc_info.value.iterations == 7
+        assert exc_info.value.max_iterations == 7
+
+    def test_caught_as_base_exception(self):
+        """Test that MaxIterationsReachedException can be caught as a generic Exception."""
+        caught = None
+        try:
+            raise MaxIterationsReachedException("limit hit", iterations=2, max_iterations=2)
+        except Exception as e:
+            caught = e
+
+        assert isinstance(caught, MaxIterationsReachedException)
+        assert isinstance(caught, Exception)
+
+
 class TestContextWindowOverflowException:
     """Tests for ContextWindowOverflowException class."""
 
@@ -284,6 +356,7 @@ def test_all_exceptions_inherit_from_exception(self):
         """Test that all custom exceptions inherit from Exception."""
         exception_classes = [
             EventLoopException,
+            MaxIterationsReachedException,
             MaxTokensReachedException,
             ContextWindowOverflowException,
             MCPClientInitializationError,
@@ -299,6 +372,7 @@ def test_exception_instances_are_exceptions(self):
         """Test that all exception instances are instances of Exception."""
         exceptions = [
             EventLoopException(ValueError("test")),
+            MaxIterationsReachedException("test"),
             MaxTokensReachedException("test"),
             ContextWindowOverflowException("test"),
             MCPClientInitializationError("test"),
@@ -314,6 +388,7 @@ def test_exceptions_can_be_caught_as_exception(self):
         """Test that all custom exceptions can be caught as generic Exception."""
         exceptions_to_raise = [
             (EventLoopException, ValueError("test"), None),
+            (MaxIterationsReachedException, "test", None),
             (MaxTokensReachedException, "test", None),
             (ContextWindowOverflowException, "test", None),
             (MCPClientInitializationError, "test", None),
@@ -340,6 +415,7 @@ def test_exception_str_representations(self):
         """Test string representations of all exceptions."""
         exceptions = [
             (EventLoopException(ValueError("event loop error")), "event loop error"),
+            (MaxIterationsReachedException("max iterations"), "max iterations"),
             (MaxTokensReachedException("max tokens"), "max tokens"),
             (ContextWindowOverflowException("overflow"), "overflow"),
             (MCPClientInitializationError("init error"), "init error"),
@@ -355,6 +431,7 @@ def test_exception_repr_contains_class_name(self):
         """Test that repr contains the exception class name."""
         exceptions = [
             EventLoopException(ValueError("test")),
+            MaxIterationsReachedException("test"),
             MaxTokensReachedException("test"),
             ContextWindowOverflowException("test"),
             MCPClientInitializationError("test"),