From 8df19f0ab31c30f445b4cf8d4d66fd9d46cf6c3d Mon Sep 17 00:00:00 2001
From: Dean Schmigelski <dbschmigelski+github@gmail.com>
Date: Fri, 23 Jan 2026 14:43:41 -0500
Subject: [PATCH 1/7] feat(hooks): add retry mechanism for tool calls

Add `retry` field to AfterToolCallEvent and BidiAfterToolCallEvent that
allows hook callbacks to trigger tool re-execution. When set to True,
the tool executor discards the current result and invokes the tool again.

This provides the hook infrastructure needed for implementing custom
retry strategies for tool failures, timeouts, or transient errors.
---
 src/strands/experimental/hooks/events.py |  18 ++-
 src/strands/hooks/events.py              |  18 ++-
 src/strands/tools/executors/_executor.py | 185 +++++++++++++----------
 3 files changed, 135 insertions(+), 86 deletions(-)

diff --git a/src/strands/experimental/hooks/events.py b/src/strands/experimental/hooks/events.py
index 081190af3..9310b1729 100644
--- a/src/strands/experimental/hooks/events.py
+++ b/src/strands/experimental/hooks/events.py
@@ -156,6 +156,18 @@ class BidiAfterToolCallEvent(BidiHookEvent):
     Note: This event uses reverse callback ordering, meaning callbacks registered
     later will be invoked first during cleanup.
 
+    Tool Retrying:
+        When ``retry`` is set to True by a hook callback, the tool executor will
+        discard the current tool result and invoke the tool again. This has important
+        implications for streaming consumers:
+
+        - Streaming events from the discarded tool execution will have already been emitted
+          to callers before the retry occurs. Agent invokers consuming streamed events
+          should be prepared to handle this scenario, potentially by tracking retry state
+          or implementing idempotent event processing
+        - The original tool result is thrown away internally and not added to the
+          conversation history
+
     Attributes:
         selected_tool: The tool that was invoked. It may be None if tool lookup failed.
         tool_use: The tool parameters that were passed to the tool invoked.
@@ -164,6 +176,9 @@ class BidiAfterToolCallEvent(BidiHookEvent):
             or an Exception if the tool execution failed.
         exception: Exception if the tool execution failed, None if successful.
         cancel_message: The cancellation message if the user cancelled the tool call.
+        retry: Whether to retry the tool invocation. Can be set by hook callbacks
+            to trigger a retry. When True, the current result is discarded and the
+            tool is called again. Defaults to False.
     """
 
     selected_tool: AgentTool | None
@@ -172,9 +187,10 @@ class BidiAfterToolCallEvent(BidiHookEvent):
     result: ToolResult
     exception: Exception | None = None
     cancel_message: str | None = None
+    retry: bool = False
 
     def _can_write(self, name: str) -> bool:
-        return name == "result"
+        return name in ["result", "retry"]
 
     @property
     def should_reverse_callbacks(self) -> bool:
diff --git a/src/strands/hooks/events.py b/src/strands/hooks/events.py
index 1faa8a917..bb8a45526 100644
--- a/src/strands/hooks/events.py
+++ b/src/strands/hooks/events.py
@@ -158,6 +158,18 @@ class AfterToolCallEvent(HookEvent):
     Note: This event uses reverse callback ordering, meaning callbacks registered
     later will be invoked first during cleanup.
 
+    Tool Retrying:
+        When ``retry`` is set to True by a hook callback, the tool executor will
+        discard the current tool result and invoke the tool again. This has important
+        implications for streaming consumers:
+
+        - Streaming events from the discarded tool execution will have already been emitted
+          to callers before the retry occurs. Agent invokers consuming streamed events
+          should be prepared to handle this scenario, potentially by tracking retry state
+          or implementing idempotent event processing
+        - The original tool result is thrown away internally and not added to the
+          conversation history
+
     Attributes:
         selected_tool: The tool that was invoked. It may be None if tool lookup failed.
         tool_use: The tool parameters that were passed to the tool invoked.
@@ -165,6 +177,9 @@ class AfterToolCallEvent(HookEvent):
         result: The result of the tool invocation. Either a ToolResult on success
             or an Exception if the tool execution failed.
         cancel_message: The cancellation message if the user cancelled the tool call.
+        retry: Whether to retry the tool invocation. Can be set by hook callbacks
+            to trigger a retry. When True, the current result is discarded and the
+            tool is called again. Defaults to False.
     """
 
     selected_tool: AgentTool | None
@@ -173,9 +188,10 @@ class AfterToolCallEvent(HookEvent):
     result: ToolResult
     exception: Exception | None = None
     cancel_message: str | None = None
+    retry: bool = False
 
     def _can_write(self, name: str) -> bool:
-        return name == "result"
+        return name in ["result", "retry"]
 
     @property
     def should_reverse_callbacks(self) -> bool:
diff --git a/src/strands/tools/executors/_executor.py b/src/strands/tools/executors/_executor.py
index 6d58c5c75..971684097 100644
--- a/src/strands/tools/executors/_executor.py
+++ b/src/strands/tools/executors/_executor.py
@@ -148,109 +148,126 @@ async def _stream(
             }
         )
 
-        before_event, interrupts = await ToolExecutor._invoke_before_tool_call_hook(
-            agent, tool_func, tool_use, invocation_state
-        )
-
-        if interrupts:
-            yield ToolInterruptEvent(tool_use, interrupts)
-            return
-
-        if before_event.cancel_tool:
-            cancel_message = (
-                before_event.cancel_tool if isinstance(before_event.cancel_tool, str) else "tool cancelled by user"
+        # Retry loop for tool execution - hooks can set after_event.retry = True to retry
+        while True:
+            before_event, interrupts = await ToolExecutor._invoke_before_tool_call_hook(
+                agent, tool_func, tool_use, invocation_state
             )
-            yield ToolCancelEvent(tool_use, cancel_message)
 
-            cancel_result: ToolResult = {
-                "toolUseId": str(tool_use.get("toolUseId")),
-                "status": "error",
-                "content": [{"text": cancel_message}],
-            }
+            if interrupts:
+                yield ToolInterruptEvent(tool_use, interrupts)
+                return
 
-            after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
-                agent, None, tool_use, invocation_state, cancel_result, cancel_message=cancel_message
-            )
-            yield ToolResultEvent(after_event.result)
-            tool_results.append(after_event.result)
-            return
-
-        try:
-            selected_tool = before_event.selected_tool
-            tool_use = before_event.tool_use
-            invocation_state = before_event.invocation_state
-
-            if not selected_tool:
-                if tool_func == selected_tool:
-                    logger.error(
-                        "tool_name=<%s>, available_tools=<%s> | tool not found in registry",
-                        tool_name,
-                        list(agent.tool_registry.registry.keys()),
-                    )
-                else:
-                    logger.debug(
-                        "tool_name=<%s>, tool_use_id=<%s> | a hook resulted in a non-existing tool call",
-                        tool_name,
-                        str(tool_use.get("toolUseId")),
-                    )
+            if before_event.cancel_tool:
+                cancel_message = (
+                    before_event.cancel_tool if isinstance(before_event.cancel_tool, str) else "tool cancelled by user"
+                )
+                yield ToolCancelEvent(tool_use, cancel_message)
 
-                result: ToolResult = {
+                cancel_result: ToolResult = {
                     "toolUseId": str(tool_use.get("toolUseId")),
                     "status": "error",
-                    "content": [{"text": f"Unknown tool: {tool_name}"}],
+                    "content": [{"text": cancel_message}],
                 }
 
                 after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
-                    agent, selected_tool, tool_use, invocation_state, result
+                    agent, None, tool_use, invocation_state, cancel_result, cancel_message=cancel_message
                 )
                 yield ToolResultEvent(after_event.result)
                 tool_results.append(after_event.result)
                 return
-            if structured_output_context.is_enabled:
-                kwargs["structured_output_context"] = structured_output_context
-            async for event in selected_tool.stream(tool_use, invocation_state, **kwargs):
-                # Internal optimization; for built-in AgentTools, we yield TypedEvents out of .stream()
-                # so that we don't needlessly yield ToolStreamEvents for non-generator callbacks.
-                # In which case, as soon as we get a ToolResultEvent we're done and for ToolStreamEvent
-                # we yield it directly; all other cases (non-sdk AgentTools), we wrap events in
-                # ToolStreamEvent and the last event is just the result.
-
-                if isinstance(event, ToolInterruptEvent):
-                    yield event
-                    return
-
-                if isinstance(event, ToolResultEvent):
-                    # below the last "event" must point to the tool_result
-                    event = event.tool_result
-                    break
 
-                if isinstance(event, ToolStreamEvent):
-                    yield event
-                else:
-                    yield ToolStreamEvent(tool_use, event)
+            try:
+                selected_tool = before_event.selected_tool
+                tool_use = before_event.tool_use
+                invocation_state = before_event.invocation_state
+
+                if not selected_tool:
+                    if tool_func == selected_tool:
+                        logger.error(
+                            "tool_name=<%s>, available_tools=<%s> | tool not found in registry",
+                            tool_name,
+                            list(agent.tool_registry.registry.keys()),
+                        )
+                    else:
+                        logger.debug(
+                            "tool_name=<%s>, tool_use_id=<%s> | a hook resulted in a non-existing tool call",
+                            tool_name,
+                            str(tool_use.get("toolUseId")),
+                        )
+
+                    result: ToolResult = {
+                        "toolUseId": str(tool_use.get("toolUseId")),
+                        "status": "error",
+                        "content": [{"text": f"Unknown tool: {tool_name}"}],
+                    }
+
+                    after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
+                        agent, selected_tool, tool_use, invocation_state, result
+                    )
+                    # Check if retry requested for unknown tool error
+                    if after_event.retry:
+                        logger.debug("tool_name=<%s> | retry requested, retrying tool call", tool_name)
+                        continue
+                    yield ToolResultEvent(after_event.result)
+                    tool_results.append(after_event.result)
+                    return
+                if structured_output_context.is_enabled:
+                    kwargs["structured_output_context"] = structured_output_context
+                async for event in selected_tool.stream(tool_use, invocation_state, **kwargs):
+                    # Internal optimization; for built-in AgentTools, we yield TypedEvents out of .stream()
+                    # so that we don't needlessly yield ToolStreamEvents for non-generator callbacks.
+                    # In which case, as soon as we get a ToolResultEvent we're done and for ToolStreamEvent
+                    # we yield it directly; all other cases (non-sdk AgentTools), we wrap events in
+                    # ToolStreamEvent and the last event is just the result.
+
+                    if isinstance(event, ToolInterruptEvent):
+                        yield event
+                        return
+
+                    if isinstance(event, ToolResultEvent):
+                        # below the last "event" must point to the tool_result
+                        event = event.tool_result
+                        break
+
+                    if isinstance(event, ToolStreamEvent):
+                        yield event
+                    else:
+                        yield ToolStreamEvent(tool_use, event)
+
+                result = cast(ToolResult, event)
 
-            result = cast(ToolResult, event)
+                after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
+                    agent, selected_tool, tool_use, invocation_state, result
+                )
 
-            after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
-                agent, selected_tool, tool_use, invocation_state, result
-            )
+                # Check if retry requested
+                if after_event.retry:
+                    logger.debug("tool_name=<%s> | retry requested, retrying tool call", tool_name)
+                    continue
 
-            yield ToolResultEvent(after_event.result)
-            tool_results.append(after_event.result)
+                yield ToolResultEvent(after_event.result)
+                tool_results.append(after_event.result)
+                return
 
-        except Exception as e:
-            logger.exception("tool_name=<%s> | failed to process tool", tool_name)
-            error_result: ToolResult = {
-                "toolUseId": str(tool_use.get("toolUseId")),
-                "status": "error",
-                "content": [{"text": f"Error: {str(e)}"}],
-            }
+            except Exception as e:
+                logger.exception("tool_name=<%s> | failed to process tool", tool_name)
+                error_result: ToolResult = {
+                    "toolUseId": str(tool_use.get("toolUseId")),
+                    "status": "error",
+                    "content": [{"text": f"Error: {str(e)}"}],
+                }
 
-            after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
-                agent, selected_tool, tool_use, invocation_state, error_result, exception=e
-            )
-            yield ToolResultEvent(after_event.result)
-            tool_results.append(after_event.result)
+                after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
+                    agent, selected_tool, tool_use, invocation_state, error_result, exception=e
+                )
+                # Check if retry requested for exception
+                if after_event.retry:
+                    logger.debug("tool_name=<%s> | retry requested after exception, retrying tool call", tool_name)
+                    continue
+                yield ToolResultEvent(after_event.result)
+                tool_results.append(after_event.result)
+                return
 
     @staticmethod
     async def _stream_with_trace(

From d6cc22fa738cc00ed66a1b6dcb268e92201b79f2 Mon Sep 17 00:00:00 2001
From: Dean Schmigelski <dbschmigelski+github@gmail.com>
Date: Fri, 23 Jan 2026 15:01:43 -0500
Subject: [PATCH 2/7] test: add unit and integration tests for tool retry hook
 mechanism

Add tests verifying:
- AfterToolCallEvent.retry field is writable
- Setting retry=True causes tool re-execution
- Retry works with error results from @tool decorator
- Retry works with exceptions from tool.stream
- Max retries can be controlled by hook callback
- Integration tests with real Agent using HookProvider
---
 .../strands/tools/executors/test_executor.py  | 166 ++++++++++++++++++
 tests_integ/test_tool_retry_hook.py           | 131 ++++++++++++++
 2 files changed, 297 insertions(+)
 create mode 100644 tests_integ/test_tool_retry_hook.py

diff --git a/tests/strands/tools/executors/test_executor.py b/tests/strands/tools/executors/test_executor.py
index 8139fbf66..f3312a060 100644
--- a/tests/strands/tools/executors/test_executor.py
+++ b/tests/strands/tools/executors/test_executor.py
@@ -479,3 +479,169 @@ async def test_executor_stream_updates_invocation_state_with_agent(
     # Verify that the invocation_state was updated with the agent
     assert "agent" in empty_invocation_state
     assert empty_invocation_state["agent"] is agent
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_retry_on_success(executor, agent, tool_results, invocation_state, alist):
+    """Test that setting retry=True on AfterToolCallEvent causes tool re-execution."""
+    call_count = {"count": 0}
+
+    @strands.tool(name="retry_tool")
+    def retry_tool():
+        call_count["count"] += 1
+        return f"attempt_{call_count['count']}"
+
+    agent.tool_registry.register_tool(retry_tool)
+
+    # Retry once, then succeed
+    def retry_callback(event):
+        if isinstance(event, AfterToolCallEvent) and call_count["count"] < 2:
+            event.retry = True
+        return event
+
+    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
+
+    tool_use: ToolUse = {"name": "retry_tool", "toolUseId": "1", "input": {}}
+    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+
+    tru_events = await alist(stream)
+
+    # Should have been called twice due to retry
+    assert call_count["count"] == 2
+
+    # Final result should be from second attempt
+    exp_events = [
+        ToolResultEvent({"toolUseId": "1", "status": "success", "content": [{"text": "attempt_2"}]}),
+    ]
+    assert tru_events == exp_events
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_retry_on_error(executor, agent, tool_results, invocation_state, alist):
+    """Test that retry works when tool raises an exception."""
+    call_count = {"count": 0}
+
+    @strands.tool(name="error_retry_tool")
+    def error_retry_tool():
+        call_count["count"] += 1
+        if call_count["count"] < 2:
+            raise RuntimeError("Simulated failure")
+        return "success"
+
+    agent.tool_registry.register_tool(error_retry_tool)
+
+    # Retry on error - check result status since @tool decorator catches exceptions
+    def retry_callback(event):
+        if isinstance(event, AfterToolCallEvent) and event.result.get("status") == "error":
+            event.retry = True
+        return event
+
+    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
+
+    tool_use: ToolUse = {"name": "error_retry_tool", "toolUseId": "1", "input": {}}
+    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+
+    tru_events = await alist(stream)
+
+    # Should have been called twice - first failed, second succeeded
+    assert call_count["count"] == 2
+
+    # Final result should be success
+    exp_events = [
+        ToolResultEvent({"toolUseId": "1", "status": "success", "content": [{"text": "success"}]}),
+    ]
+    assert tru_events == exp_events
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_retry_respects_max_retries(executor, agent, tool_results, invocation_state, alist):
+    """Test that retry can be limited by the hook callback."""
+    call_count = {"count": 0}
+    max_retries = 3
+
+    @strands.tool(name="limited_retry_tool")
+    def limited_retry_tool():
+        call_count["count"] += 1
+        raise RuntimeError(f"Failure {call_count['count']}")
+
+    agent.tool_registry.register_tool(limited_retry_tool)
+
+    # Retry up to max_retries times - check result status since @tool decorator catches exceptions
+    def retry_callback(event):
+        if isinstance(event, AfterToolCallEvent) and event.result.get("status") == "error":
+            if call_count["count"] < max_retries:
+                event.retry = True
+        return event
+
+    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
+
+    tool_use: ToolUse = {"name": "limited_retry_tool", "toolUseId": "1", "input": {}}
+    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+
+    tru_events = await alist(stream)
+
+    # Should have been called max_retries times
+    assert call_count["count"] == max_retries
+
+    # Final result should be error from last attempt
+    assert len(tru_events) == 1
+    assert tru_events[0].tool_result["status"] == "error"
+    assert "Failure 3" in tru_events[0].tool_result["content"][0]["text"]
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_retry_on_exception(executor, agent, tool_results, invocation_state, alist):
+    """Test that retry works when tool.stream raises an exception directly."""
+    call_count = {"count": 0}
+
+    @strands.tool(name="stream_exception_tool")
+    def stream_exception_tool():
+        return "success"
+
+    # Replace stream to raise exception directly (like real exception_tool fixture)
+    async def mock_stream(_tool_use, _invocation_state, **kwargs):
+        call_count["count"] += 1
+        if call_count["count"] < 2:
+            raise RuntimeError("Stream exception")
+        yield {"toolUseId": "1", "status": "success", "content": [{"text": "recovered"}]}
+
+    stream_exception_tool.stream = mock_stream
+    agent.tool_registry.register_tool(stream_exception_tool)
+
+    # Retry on exception - event.exception is set when executor catches exception
+    def retry_callback(event):
+        if isinstance(event, AfterToolCallEvent) and event.exception is not None:
+            event.retry = True
+        return event
+
+    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
+
+    tool_use: ToolUse = {"name": "stream_exception_tool", "toolUseId": "1", "input": {}}
+    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+
+    tru_events = await alist(stream)
+
+    # Should have been called twice - first raised exception, second succeeded
+    assert call_count["count"] == 2
+
+    # Final result should be success (last event is ToolResultEvent)
+    assert tru_events[-1].tool_result["status"] == "success"
+    assert "recovered" in tru_events[-1].tool_result["content"][0]["text"]
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_retry_field_is_writable(executor, agent, tool_results, invocation_state, alist):
+    """Test that the retry field on AfterToolCallEvent can be written."""
+    tool_use: ToolUse = {"name": "weather_tool", "toolUseId": "1", "input": {}}
+
+    # Verify retry field can be set
+    def check_retry_writable(event):
+        if isinstance(event, AfterToolCallEvent):
+            assert event._can_write("retry") is True
+            # Don't actually retry, just verify it's writable
+        return event
+
+    agent.hooks.add_callback(AfterToolCallEvent, check_retry_writable)
+
+    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+    await alist(stream)
diff --git a/tests_integ/test_tool_retry_hook.py b/tests_integ/test_tool_retry_hook.py
new file mode 100644
index 000000000..b3f8f64f8
--- /dev/null
+++ b/tests_integ/test_tool_retry_hook.py
@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""Integration tests for tool retry hook mechanism.
+
+Tests that the AfterToolCallEvent.retry field works correctly with real agents,
+allowing hooks to trigger tool re-execution on failures.
+"""
+
+from strands import Agent, tool
+from strands.hooks import AfterToolCallEvent
+from strands.hooks.registry import HookProvider, HookRegistry
+from tests_integ.conftest import retry_on_flaky
+
+
+def make_failing_tool(fail_times: int = 1):
+    """Create a tool that fails a configurable number of times before succeeding."""
+    state = {"call_count": 0, "fail_times": fail_times}
+
+    @tool(name="flaky_tool")
+    def _flaky_tool(message: str) -> str:
+        """A tool that fails a few times before succeeding.
+
+        Args:
+            message: A message to include in the response.
+        """
+        state["call_count"] += 1
+        if state["call_count"] <= state["fail_times"]:
+            raise RuntimeError(f"Simulated failure {state['call_count']}")
+        return f"Success on attempt {state['call_count']}: {message}"
+
+    _flaky_tool.state = state
+    return _flaky_tool
+
+
+class SimpleRetryHook(HookProvider):
+    """A simple hook that retries failed tool calls up to max_attempts times."""
+
+    def __init__(self, max_attempts: int = 3):
+        self.max_attempts = max_attempts
+        self._attempts: dict[str, int] = {}
+
+    def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
+        registry.add_callback(AfterToolCallEvent, self._handle_after_tool_call)
+
+    def _handle_after_tool_call(self, event: AfterToolCallEvent) -> None:
+        tool_use_id = str(event.tool_use.get("toolUseId", ""))
+
+        # Track attempts per tool_use_id
+        current_attempt = self._attempts.get(tool_use_id, 0) + 1
+        self._attempts[tool_use_id] = current_attempt
+
+        # Check for error status - @tool decorator catches exceptions and returns error results
+        is_error = event.result.get("status") == "error"
+
+        # If there was an error and we haven't exceeded max attempts, retry
+        if is_error and current_attempt < self.max_attempts:
+            event.retry = True
+
+
+@retry_on_flaky("LLM responses may vary in tool calling behavior")
+def test_tool_retry_hook_retries_on_failure():
+    """Test that a hook can trigger tool retry on failure."""
+    flaky_tool = make_failing_tool(fail_times=1)
+    retry_hook = SimpleRetryHook(max_attempts=3)
+
+    agent = Agent(tools=[flaky_tool], hooks=[retry_hook])
+
+    # Ask the agent to use the tool
+    result = agent("Use the flaky_tool with message 'hello'")
+
+    # Tool should have been called twice (1 failure + 1 success)
+    assert flaky_tool.state["call_count"] == 2
+
+    # The result should contain the success message
+    assert "Success on attempt 2" in str(result)
+
+
+@retry_on_flaky("LLM responses may vary in tool calling behavior")
+def test_tool_retry_hook_respects_max_attempts():
+    """Test that retry hook respects max_attempts limit."""
+    # Tool that always fails
+    flaky_tool = make_failing_tool(fail_times=100)
+    retry_hook = SimpleRetryHook(max_attempts=3)
+
+    agent = Agent(tools=[flaky_tool], hooks=[retry_hook])
+
+    # Ask the agent to use the tool - it will fail but be retried up to max_attempts
+    agent("Use the flaky_tool with message 'test'")
+
+    # Tool should have been called exactly max_attempts times
+    assert flaky_tool.state["call_count"] == 3
+
+
+def test_tool_retry_hook_direct_tool_invocation():
+    """Test retry hook works with direct tool invocation."""
+    flaky_tool = make_failing_tool(fail_times=2)
+    retry_hook = SimpleRetryHook(max_attempts=5)
+
+    agent = Agent(tools=[flaky_tool], hooks=[retry_hook])
+
+    # Call tool directly
+    result = agent.tool.flaky_tool(message="direct call")
+
+    # Tool should have been called 3 times (2 failures + 1 success)
+    assert flaky_tool.state["call_count"] == 3
+    assert result["status"] == "success"
+    assert "Success on attempt 3" in result["content"][0]["text"]
+
+
+def test_tool_retry_hook_no_retry_on_success():
+    """Test that successful tool calls are not retried."""
+    call_count = {"count": 0}
+
+    @tool(name="success_tool")
+    def success_tool(value: str) -> str:
+        """A tool that always succeeds.
+
+        Args:
+            value: A value to return.
+        """
+        call_count["count"] += 1
+        return f"Result: {value}"
+
+    retry_hook = SimpleRetryHook(max_attempts=3)
+    agent = Agent(tools=[success_tool], hooks=[retry_hook])
+
+    # Call tool directly
+    result = agent.tool.success_tool(value="test")
+
+    # Tool should have been called only once
+    assert call_count["count"] == 1
+    assert result["status"] == "success"

From 87466596ef7cf3dc21218bbcb1b54a79e9881f86 Mon Sep 17 00:00:00 2001
From: Dean Schmigelski <dbschmigelski+github@gmail.com>
Date: Fri, 23 Jan 2026 15:06:29 -0500
Subject: [PATCH 3/7] refactor: remove BidiAfterToolCallEvent retry field from
 this PR

Keep the first PR focused on standard AfterToolCallEvent only.
BidiAfterToolCallEvent.retry can be added in a follow-up if needed.
---
 src/strands/experimental/hooks/events.py      |  18 +-
 src/strands/tools/executors/_executor.py      |   6 +-
 .../strands/tools/executors/test_executor.py  | 193 +++++++++---------
 tests_integ/test_tool_retry_hook.py           | 146 ++++---------
 4 files changed, 138 insertions(+), 225 deletions(-)

diff --git a/src/strands/experimental/hooks/events.py b/src/strands/experimental/hooks/events.py
index 9310b1729..081190af3 100644
--- a/src/strands/experimental/hooks/events.py
+++ b/src/strands/experimental/hooks/events.py
@@ -156,18 +156,6 @@ class BidiAfterToolCallEvent(BidiHookEvent):
     Note: This event uses reverse callback ordering, meaning callbacks registered
     later will be invoked first during cleanup.
 
-    Tool Retrying:
-        When ``retry`` is set to True by a hook callback, the tool executor will
-        discard the current tool result and invoke the tool again. This has important
-        implications for streaming consumers:
-
-        - Streaming events from the discarded tool execution will have already been emitted
-          to callers before the retry occurs. Agent invokers consuming streamed events
-          should be prepared to handle this scenario, potentially by tracking retry state
-          or implementing idempotent event processing
-        - The original tool result is thrown away internally and not added to the
-          conversation history
-
     Attributes:
         selected_tool: The tool that was invoked. It may be None if tool lookup failed.
         tool_use: The tool parameters that were passed to the tool invoked.
@@ -176,9 +164,6 @@ class BidiAfterToolCallEvent(BidiHookEvent):
             or an Exception if the tool execution failed.
         exception: Exception if the tool execution failed, None if successful.
         cancel_message: The cancellation message if the user cancelled the tool call.
-        retry: Whether to retry the tool invocation. Can be set by hook callbacks
-            to trigger a retry. When True, the current result is discarded and the
-            tool is called again. Defaults to False.
     """
 
     selected_tool: AgentTool | None
@@ -187,10 +172,9 @@ class BidiAfterToolCallEvent(BidiHookEvent):
     result: ToolResult
     exception: Exception | None = None
     cancel_message: str | None = None
-    retry: bool = False
 
     def _can_write(self, name: str) -> bool:
-        return name in ["result", "retry"]
+        return name == "result"
 
     @property
     def should_reverse_callbacks(self) -> bool:
diff --git a/src/strands/tools/executors/_executor.py b/src/strands/tools/executors/_executor.py
index 971684097..dbb72ccdf 100644
--- a/src/strands/tools/executors/_executor.py
+++ b/src/strands/tools/executors/_executor.py
@@ -206,7 +206,7 @@ async def _stream(
                         agent, selected_tool, tool_use, invocation_state, result
                     )
                     # Check if retry requested for unknown tool error
-                    if after_event.retry:
+                    if getattr(after_event, "retry", False):
                         logger.debug("tool_name=<%s> | retry requested, retrying tool call", tool_name)
                         continue
                     yield ToolResultEvent(after_event.result)
@@ -242,7 +242,7 @@ async def _stream(
                 )
 
                 # Check if retry requested
-                if after_event.retry:
+                if getattr(after_event, "retry", False):
                     logger.debug("tool_name=<%s> | retry requested, retrying tool call", tool_name)
                     continue
 
@@ -262,7 +262,7 @@ async def _stream(
                     agent, selected_tool, tool_use, invocation_state, error_result, exception=e
                 )
                 # Check if retry requested for exception
-                if after_event.retry:
+                if getattr(after_event, "retry", False):
                     logger.debug("tool_name=<%s> | retry requested after exception, retrying tool call", tool_name)
                     continue
                 yield ToolResultEvent(after_event.result)
diff --git a/tests/strands/tools/executors/test_executor.py b/tests/strands/tools/executors/test_executor.py
index f3312a060..2b3e4e56e 100644
--- a/tests/strands/tools/executors/test_executor.py
+++ b/tests/strands/tools/executors/test_executor.py
@@ -482,166 +482,157 @@ async def test_executor_stream_updates_invocation_state_with_agent(
 
 
 @pytest.mark.asyncio
-async def test_executor_stream_retry_on_success(executor, agent, tool_results, invocation_state, alist):
-    """Test that setting retry=True on AfterToolCallEvent causes tool re-execution."""
+async def test_executor_stream_no_retry_set(executor, agent, tool_results, invocation_state, alist):
+    """Test default behavior when retry is not set - tool executes once."""
     call_count = {"count": 0}
 
-    @strands.tool(name="retry_tool")
-    def retry_tool():
+    @strands.tool(name="counting_tool")
+    def counting_tool():
         call_count["count"] += 1
         return f"attempt_{call_count['count']}"
 
-    agent.tool_registry.register_tool(retry_tool)
+    agent.tool_registry.register_tool(counting_tool)
 
-    # Retry once, then succeed
-    def retry_callback(event):
-        if isinstance(event, AfterToolCallEvent) and call_count["count"] < 2:
-            event.retry = True
-        return event
-
-    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
-
-    tool_use: ToolUse = {"name": "retry_tool", "toolUseId": "1", "input": {}}
+    tool_use: ToolUse = {"name": "counting_tool", "toolUseId": "1", "input": {}}
     stream = executor._stream(agent, tool_use, tool_results, invocation_state)
 
     tru_events = await alist(stream)
 
-    # Should have been called twice due to retry
-    assert call_count["count"] == 2
+    # Tool should be called exactly once
+    assert call_count["count"] == 1
 
-    # Final result should be from second attempt
-    exp_events = [
-        ToolResultEvent({"toolUseId": "1", "status": "success", "content": [{"text": "attempt_2"}]}),
-    ]
-    assert tru_events == exp_events
+    # Single result event with first attempt's content
+    assert len(tru_events) == 1
+    assert tru_events[0].tool_result == {"toolUseId": "1", "status": "success", "content": [{"text": "attempt_1"}]}
+
+    # tool_results should contain the result
+    assert len(tool_results) == 1
+    assert tool_results[0] == {"toolUseId": "1", "status": "success", "content": [{"text": "attempt_1"}]}
 
 
 @pytest.mark.asyncio
-async def test_executor_stream_retry_on_error(executor, agent, tool_results, invocation_state, alist):
-    """Test that retry works when tool raises an exception."""
+async def test_executor_stream_retry_true(executor, agent, tool_results, invocation_state, alist):
+    """Test that retry=True causes tool re-execution."""
     call_count = {"count": 0}
 
-    @strands.tool(name="error_retry_tool")
-    def error_retry_tool():
+    @strands.tool(name="counting_tool")
+    def counting_tool():
         call_count["count"] += 1
-        if call_count["count"] < 2:
-            raise RuntimeError("Simulated failure")
-        return "success"
+        return f"attempt_{call_count['count']}"
 
-    agent.tool_registry.register_tool(error_retry_tool)
+    agent.tool_registry.register_tool(counting_tool)
 
-    # Retry on error - check result status since @tool decorator catches exceptions
-    def retry_callback(event):
-        if isinstance(event, AfterToolCallEvent) and event.result.get("status") == "error":
+    # Set retry=True on first call only
+    def retry_once(event):
+        if isinstance(event, AfterToolCallEvent) and call_count["count"] == 1:
             event.retry = True
         return event
 
-    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
+    agent.hooks.add_callback(AfterToolCallEvent, retry_once)
 
-    tool_use: ToolUse = {"name": "error_retry_tool", "toolUseId": "1", "input": {}}
+    tool_use: ToolUse = {"name": "counting_tool", "toolUseId": "1", "input": {}}
     stream = executor._stream(agent, tool_use, tool_results, invocation_state)
 
     tru_events = await alist(stream)
 
-    # Should have been called twice - first failed, second succeeded
+    # Tool should be called twice due to retry
     assert call_count["count"] == 2
 
-    # Final result should be success
-    exp_events = [
-        ToolResultEvent({"toolUseId": "1", "status": "success", "content": [{"text": "success"}]}),
-    ]
-    assert tru_events == exp_events
+    # Only final result is yielded (first attempt's result was discarded)
+    assert len(tru_events) == 1
+    assert tru_events[0].tool_result == {"toolUseId": "1", "status": "success", "content": [{"text": "attempt_2"}]}
+
+    # tool_results only contains the final result
+    assert len(tool_results) == 1
+    assert tool_results[0] == {"toolUseId": "1", "status": "success", "content": [{"text": "attempt_2"}]}
 
 
 @pytest.mark.asyncio
-async def test_executor_stream_retry_respects_max_retries(executor, agent, tool_results, invocation_state, alist):
-    """Test that retry can be limited by the hook callback."""
+async def test_executor_stream_retry_true_emits_events_from_both_attempts(
+    executor, agent, tool_results, invocation_state, alist
+):
+    """Test that streaming events from discarded attempt ARE emitted before retry.
+
+    This validates the documented behavior: 'Streaming events from the discarded
+    tool execution will have already been emitted to callers before the retry occurs.'
+    """
     call_count = {"count": 0}
-    max_retries = 3
 
-    @strands.tool(name="limited_retry_tool")
-    def limited_retry_tool():
+    @strands.tool(name="streaming_tool")
+    def streaming_tool():
+        return "unused"
+
+    # Provide streaming implementation (same pattern as exception_tool fixture)
+    async def tool_stream(_tool_use, _invocation_state, **kwargs):
         call_count["count"] += 1
-        raise RuntimeError(f"Failure {call_count['count']}")
+        yield f"streaming_from_attempt_{call_count['count']}"
+        yield ToolResultEvent(
+            {"toolUseId": "1", "status": "success", "content": [{"text": f"result_{call_count['count']}"}]}
+        )
 
-    agent.tool_registry.register_tool(limited_retry_tool)
+    streaming_tool.stream = tool_stream
+    agent.tool_registry.register_tool(streaming_tool)
 
-    # Retry up to max_retries times - check result status since @tool decorator catches exceptions
-    def retry_callback(event):
-        if isinstance(event, AfterToolCallEvent) and event.result.get("status") == "error":
-            if call_count["count"] < max_retries:
-                event.retry = True
+    # Set retry=True on first call
+    def retry_once(event):
+        if isinstance(event, AfterToolCallEvent) and call_count["count"] == 1:
+            event.retry = True
         return event
 
-    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
+    agent.hooks.add_callback(AfterToolCallEvent, retry_once)
 
-    tool_use: ToolUse = {"name": "limited_retry_tool", "toolUseId": "1", "input": {}}
+    tool_use: ToolUse = {"name": "streaming_tool", "toolUseId": "1", "input": {}}
     stream = executor._stream(agent, tool_use, tool_results, invocation_state)
 
     tru_events = await alist(stream)
 
-    # Should have been called max_retries times
-    assert call_count["count"] == max_retries
+    # Tool called twice
+    assert call_count["count"] == 2
 
-    # Final result should be error from last attempt
-    assert len(tru_events) == 1
-    assert tru_events[0].tool_result["status"] == "error"
-    assert "Failure 3" in tru_events[0].tool_result["content"][0]["text"]
+    # Streaming events from BOTH attempts are emitted (documented behavior)
+    stream_events = [e for e in tru_events if isinstance(e, ToolStreamEvent)]
+    assert len(stream_events) == 2
+    assert stream_events[0] == ToolStreamEvent(tool_use, "streaming_from_attempt_1")
+    assert stream_events[1] == ToolStreamEvent(tool_use, "streaming_from_attempt_2")
+
+    # Only final ToolResultEvent is emitted
+    result_events = [e for e in tru_events if isinstance(e, ToolResultEvent)]
+    assert len(result_events) == 1
+    assert result_events[0].tool_result["content"][0]["text"] == "result_2"
 
 
 @pytest.mark.asyncio
-async def test_executor_stream_retry_on_exception(executor, agent, tool_results, invocation_state, alist):
-    """Test that retry works when tool.stream raises an exception directly."""
+async def test_executor_stream_retry_false(executor, agent, tool_results, invocation_state, alist):
+    """Test that explicitly setting retry=False does not retry."""
     call_count = {"count": 0}
 
-    @strands.tool(name="stream_exception_tool")
-    def stream_exception_tool():
-        return "success"
-
-    # Replace stream to raise exception directly (like real exception_tool fixture)
-    async def mock_stream(_tool_use, _invocation_state, **kwargs):
+    @strands.tool(name="counting_tool")
+    def counting_tool():
         call_count["count"] += 1
-        if call_count["count"] < 2:
-            raise RuntimeError("Stream exception")
-        yield {"toolUseId": "1", "status": "success", "content": [{"text": "recovered"}]}
+        return f"attempt_{call_count['count']}"
 
-    stream_exception_tool.stream = mock_stream
-    agent.tool_registry.register_tool(stream_exception_tool)
+    agent.tool_registry.register_tool(counting_tool)
 
-    # Retry on exception - event.exception is set when executor catches exception
-    def retry_callback(event):
-        if isinstance(event, AfterToolCallEvent) and event.exception is not None:
-            event.retry = True
+    # Explicitly set retry=False
+    def no_retry(event):
+        if isinstance(event, AfterToolCallEvent):
+            event.retry = False
         return event
 
-    agent.hooks.add_callback(AfterToolCallEvent, retry_callback)
+    agent.hooks.add_callback(AfterToolCallEvent, no_retry)
 
-    tool_use: ToolUse = {"name": "stream_exception_tool", "toolUseId": "1", "input": {}}
+    tool_use: ToolUse = {"name": "counting_tool", "toolUseId": "1", "input": {}}
     stream = executor._stream(agent, tool_use, tool_results, invocation_state)
 
     tru_events = await alist(stream)
 
-    # Should have been called twice - first raised exception, second succeeded
-    assert call_count["count"] == 2
-
-    # Final result should be success (last event is ToolResultEvent)
-    assert tru_events[-1].tool_result["status"] == "success"
-    assert "recovered" in tru_events[-1].tool_result["content"][0]["text"]
+    # Tool should be called exactly once
+    assert call_count["count"] == 1
 
+    # Single result event
+    assert len(tru_events) == 1
+    assert tru_events[0].tool_result == {"toolUseId": "1", "status": "success", "content": [{"text": "attempt_1"}]}
 
-@pytest.mark.asyncio
-async def test_executor_stream_retry_field_is_writable(executor, agent, tool_results, invocation_state, alist):
-    """Test that the retry field on AfterToolCallEvent can be written."""
-    tool_use: ToolUse = {"name": "weather_tool", "toolUseId": "1", "input": {}}
-
-    # Verify retry field can be set
-    def check_retry_writable(event):
-        if isinstance(event, AfterToolCallEvent):
-            assert event._can_write("retry") is True
-            # Don't actually retry, just verify it's writable
-        return event
-
-    agent.hooks.add_callback(AfterToolCallEvent, check_retry_writable)
-
-    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
-    await alist(stream)
+    # tool_results should contain the result
+    assert len(tool_results) == 1
+    assert tool_results[0] == {"toolUseId": "1", "status": "success", "content": [{"text": "attempt_1"}]}
diff --git a/tests_integ/test_tool_retry_hook.py b/tests_integ/test_tool_retry_hook.py
index b3f8f64f8..3e35ff5e6 100644
--- a/tests_integ/test_tool_retry_hook.py
+++ b/tests_integ/test_tool_retry_hook.py
@@ -1,131 +1,69 @@
 #!/usr/bin/env python3
 """Integration tests for tool retry hook mechanism.
 
-Tests that the AfterToolCallEvent.retry field works correctly with real agents,
-allowing hooks to trigger tool re-execution on failures.
+Tests that setting AfterToolCallEvent.retry=True causes tool re-execution.
+Uses direct tool invocation to test the executor-level retry, not model behavior.
 """
 
 from strands import Agent, tool
 from strands.hooks import AfterToolCallEvent
-from strands.hooks.registry import HookProvider, HookRegistry
-from tests_integ.conftest import retry_on_flaky
 
 
-def make_failing_tool(fail_times: int = 1):
-    """Create a tool that fails a configurable number of times before succeeding."""
-    state = {"call_count": 0, "fail_times": fail_times}
+def test_tool_retry_hook_causes_reexecution():
+    """Test that setting retry=True on AfterToolCallEvent causes tool re-execution.
+
+    Verifies:
+    1. Tool is called again when retry=True
+    2. Hook receives AfterToolCallEvent for BOTH attempts
+    3. Same tool_use_id is used (proves executor retry, not model re-calling)
+    """
+    state = {"call_count": 0}
 
     @tool(name="flaky_tool")
-    def _flaky_tool(message: str) -> str:
-        """A tool that fails a few times before succeeding.
+    def flaky_tool(message: str) -> str:
+        """A tool that fails once then succeeds.
 
         Args:
             message: A message to include in the response.
         """
         state["call_count"] += 1
-        if state["call_count"] <= state["fail_times"]:
-            raise RuntimeError(f"Simulated failure {state['call_count']}")
-        return f"Success on attempt {state['call_count']}: {message}"
-
-    _flaky_tool.state = state
-    return _flaky_tool
-
-
-class SimpleRetryHook(HookProvider):
-    """A simple hook that retries failed tool calls up to max_attempts times."""
+        if state["call_count"] == 1:
+            raise RuntimeError("First call fails")
+        return f"Success on attempt {state['call_count']}"
 
-    def __init__(self, max_attempts: int = 3):
-        self.max_attempts = max_attempts
-        self._attempts: dict[str, int] = {}
+    hook_calls: list[dict] = []
 
-    def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
-        registry.add_callback(AfterToolCallEvent, self._handle_after_tool_call)
-
-    def _handle_after_tool_call(self, event: AfterToolCallEvent) -> None:
+    def retry_on_first_error(event: AfterToolCallEvent) -> None:
         tool_use_id = str(event.tool_use.get("toolUseId", ""))
-
-        # Track attempts per tool_use_id
-        current_attempt = self._attempts.get(tool_use_id, 0) + 1
-        self._attempts[tool_use_id] = current_attempt
-
-        # Check for error status - @tool decorator catches exceptions and returns error results
-        is_error = event.result.get("status") == "error"
-
-        # If there was an error and we haven't exceeded max attempts, retry
-        if is_error and current_attempt < self.max_attempts:
+        hook_calls.append(
+            {
+                "tool_use_id": tool_use_id,
+                "status": event.result.get("status"),
+                "attempt": state["call_count"],
+            }
+        )
+
+        # Retry once on error
+        if event.result.get("status") == "error" and state["call_count"] == 1:
             event.retry = True
 
+    agent = Agent(tools=[flaky_tool])
+    agent.hooks.add_callback(AfterToolCallEvent, retry_on_first_error)
 
-@retry_on_flaky("LLM responses may vary in tool calling behavior")
-def test_tool_retry_hook_retries_on_failure():
-    """Test that a hook can trigger tool retry on failure."""
-    flaky_tool = make_failing_tool(fail_times=1)
-    retry_hook = SimpleRetryHook(max_attempts=3)
-
-    agent = Agent(tools=[flaky_tool], hooks=[retry_hook])
-
-    # Ask the agent to use the tool
-    result = agent("Use the flaky_tool with message 'hello'")
-
-    # Tool should have been called twice (1 failure + 1 success)
-    assert flaky_tool.state["call_count"] == 2
-
-    # The result should contain the success message
-    assert "Success on attempt 2" in str(result)
-
-
-@retry_on_flaky("LLM responses may vary in tool calling behavior")
-def test_tool_retry_hook_respects_max_attempts():
-    """Test that retry hook respects max_attempts limit."""
-    # Tool that always fails
-    flaky_tool = make_failing_tool(fail_times=100)
-    retry_hook = SimpleRetryHook(max_attempts=3)
+    # Direct tool invocation bypasses model - tests executor retry mechanism
+    result = agent.tool.flaky_tool(message="test")
 
-    agent = Agent(tools=[flaky_tool], hooks=[retry_hook])
-
-    # Ask the agent to use the tool - it will fail but be retried up to max_attempts
-    agent("Use the flaky_tool with message 'test'")
-
-    # Tool should have been called exactly max_attempts times
-    assert flaky_tool.state["call_count"] == 3
-
-
-def test_tool_retry_hook_direct_tool_invocation():
-    """Test retry hook works with direct tool invocation."""
-    flaky_tool = make_failing_tool(fail_times=2)
-    retry_hook = SimpleRetryHook(max_attempts=5)
-
-    agent = Agent(tools=[flaky_tool], hooks=[retry_hook])
-
-    # Call tool directly
-    result = agent.tool.flaky_tool(message="direct call")
-
-    # Tool should have been called 3 times (2 failures + 1 success)
-    assert flaky_tool.state["call_count"] == 3
-    assert result["status"] == "success"
-    assert "Success on attempt 3" in result["content"][0]["text"]
-
-
-def test_tool_retry_hook_no_retry_on_success():
-    """Test that successful tool calls are not retried."""
-    call_count = {"count": 0}
-
-    @tool(name="success_tool")
-    def success_tool(value: str) -> str:
-        """A tool that always succeeds.
-
-        Args:
-            value: A value to return.
-        """
-        call_count["count"] += 1
-        return f"Result: {value}"
+    # Tool was called twice (1 failure + 1 success)
+    assert state["call_count"] == 2
 
-    retry_hook = SimpleRetryHook(max_attempts=3)
-    agent = Agent(tools=[success_tool], hooks=[retry_hook])
+    # Hook received AfterToolCallEvent for BOTH attempts
+    assert len(hook_calls) == 2
+    assert hook_calls[0]["status"] == "error"
+    assert hook_calls[0]["attempt"] == 1
+    assert hook_calls[1]["status"] == "success"
+    assert hook_calls[1]["attempt"] == 2
 
-    # Call tool directly
-    result = agent.tool.success_tool(value="test")
+    # Both calls used the same tool_use_id (executor retry, not new model call)
+    assert hook_calls[0]["tool_use_id"] == hook_calls[1]["tool_use_id"]
 
-    # Tool should have been called only once
-    assert call_count["count"] == 1
     assert result["status"] == "success"

From 2cb90fc919c818d7ab74bab0faa08b2c20626951 Mon Sep 17 00:00:00 2001
From: Dean Schmigelski <dbschmigelski+github@gmail.com>
Date: Fri, 23 Jan 2026 16:33:52 -0500
Subject: [PATCH 4/7] docs: clarify ToolStreamEvent vs ToolResultEvent behavior
 on retry

- ToolStreamEvents ARE emitted from discarded attempts
- ToolResultEvent is NOT emitted for discarded attempts
---
 src/strands/hooks/events.py                    | 12 ++++++------
 tests/strands/tools/executors/test_executor.py |  6 +++++-
 2 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/src/strands/hooks/events.py b/src/strands/hooks/events.py
index bb8a45526..ad40dfd7f 100644
--- a/src/strands/hooks/events.py
+++ b/src/strands/hooks/events.py
@@ -163,12 +163,12 @@ class AfterToolCallEvent(HookEvent):
         discard the current tool result and invoke the tool again. This has important
         implications for streaming consumers:
 
-        - Streaming events from the discarded tool execution will have already been emitted
-          to callers before the retry occurs. Agent invokers consuming streamed events
-          should be prepared to handle this scenario, potentially by tracking retry state
-          or implementing idempotent event processing
-        - The original tool result is thrown away internally and not added to the
-          conversation history
+        - ToolStreamEvents (intermediate streaming events) from the discarded tool execution
+          will have already been emitted to callers before the retry occurs. Agent invokers
+          consuming streamed events should be prepared to handle this scenario, potentially
+          by tracking retry state or implementing idempotent event processing
+        - ToolResultEvent is NOT emitted for discarded attempts - only the final attempt's
+          result is emitted and added to the conversation history
 
     Attributes:
         selected_tool: The tool that was invoked. It may be None if tool lookup failed.
diff --git a/tests/strands/tools/executors/test_executor.py b/tests/strands/tools/executors/test_executor.py
index 2b3e4e56e..bbc4aa71f 100644
--- a/tests/strands/tools/executors/test_executor.py
+++ b/tests/strands/tools/executors/test_executor.py
@@ -551,10 +551,14 @@ def retry_once(event):
 async def test_executor_stream_retry_true_emits_events_from_both_attempts(
     executor, agent, tool_results, invocation_state, alist
 ):
-    """Test that streaming events from discarded attempt ARE emitted before retry.
+    """Test that ToolStreamEvents from discarded attempt ARE emitted, but ToolResultEvent is NOT.
 
     This validates the documented behavior: 'Streaming events from the discarded
     tool execution will have already been emitted to callers before the retry occurs.'
+
+    Key distinction:
+    - ToolStreamEvent (intermediate): Yielded immediately, visible from BOTH attempts
+    - ToolResultEvent (final): Only yielded for the final attempt, discarded on retry
     """
     call_count = {"count": 0}
 

From 9b26019d7bc5752b90a3b5cf28cf96ee2e9e039c Mon Sep 17 00:00:00 2001
From: Dean Schmigelski <dbschmigelski+github@gmail.com>
Date: Mon, 26 Jan 2026 10:32:18 -0500
Subject: [PATCH 5/7] add bidi test case

---
 .../strands/tools/executors/test_executor.py  | 47 +++++++++++++++++++
 1 file changed, 47 insertions(+)

diff --git a/tests/strands/tools/executors/test_executor.py b/tests/strands/tools/executors/test_executor.py
index bbc4aa71f..f9459dc04 100644
--- a/tests/strands/tools/executors/test_executor.py
+++ b/tests/strands/tools/executors/test_executor.py
@@ -4,6 +4,7 @@
 import pytest
 
 import strands
+from strands.experimental.hooks.events import BidiAfterToolCallEvent
 from strands.hooks import AfterToolCallEvent, BeforeToolCallEvent
 from strands.interrupt import Interrupt
 from strands.telemetry.metrics import Trace
@@ -640,3 +641,49 @@ def no_retry(event):
     # tool_results should contain the result
     assert len(tool_results) == 1
     assert tool_results[0] == {"toolUseId": "1", "status": "success", "content": [{"text": "attempt_1"}]}
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_bidi_event_no_retry_attribute(executor, agent, tool_results, invocation_state, alist):
+    """Test that BidiAfterToolCallEvent (which lacks retry attribute) doesn't cause retry.
+
+    This tests the getattr(after_event, "retry", False) fallback for events without retry.
+    """
+    call_count = {"count": 0}
+
+    @strands.tool(name="counting_tool")
+    def counting_tool():
+        call_count["count"] += 1
+        return f"attempt_{call_count['count']}"
+
+    agent.tool_registry.register_tool(counting_tool)
+
+    tool_use: ToolUse = {"name": "counting_tool", "toolUseId": "1", "input": {}}
+    result: strands.types.tools.ToolResult = {
+        "toolUseId": "1",
+        "status": "success",
+        "content": [{"text": "attempt_1"}],
+    }
+
+    # Create a BidiAfterToolCallEvent (which has no retry attribute)
+    bidi_event = BidiAfterToolCallEvent(
+        agent=agent,
+        selected_tool=counting_tool,
+        tool_use=tool_use,
+        invocation_state=invocation_state,
+        result=result,
+    )
+
+    # Patch _invoke_after_tool_call_hook to return BidiAfterToolCallEvent
+    async def mock_after_hook(*args, **kwargs):
+        return bidi_event, []
+
+    with unittest.mock.patch.object(ToolExecutor, "_invoke_after_tool_call_hook", mock_after_hook):
+        stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+        tru_events = await alist(stream)
+
+    # Tool should be called once - no retry since BidiAfterToolCallEvent has no retry attr
+    assert call_count["count"] == 1
+
+    # Result should be returned
+    assert len(tru_events) == 1

From 4706d9b5317a34750bb2c0e2741e26f1ea6a7afb Mon Sep 17 00:00:00 2001
From: Dean Schmigelski <dbschmigelski+github@gmail.com>
Date: Tue, 27 Jan 2026 12:33:25 -0500
Subject: [PATCH 6/7] test: add coverage for exception and unknown tool retry
 paths

- test_executor_stream_retry_after_exception: covers retry after tool exception
- test_executor_stream_retry_after_unknown_tool: covers retry after unknown tool error
---
 .../strands/tools/executors/test_executor.py  | 71 +++++++++++++++++++
 1 file changed, 71 insertions(+)

diff --git a/tests/strands/tools/executors/test_executor.py b/tests/strands/tools/executors/test_executor.py
index f9459dc04..78e35c2aa 100644
--- a/tests/strands/tools/executors/test_executor.py
+++ b/tests/strands/tools/executors/test_executor.py
@@ -687,3 +687,74 @@ async def mock_after_hook(*args, **kwargs):
 
     # Result should be returned
     assert len(tru_events) == 1
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_retry_after_exception(executor, agent, tool_results, invocation_state, alist):
+    """Test that retry=True works when tool raises an exception.
+
+    Covers the exception path retry check.
+    """
+    call_count = {"count": 0}
+
+    @strands.tool(name="flaky_tool")
+    def flaky_tool():
+        call_count["count"] += 1
+        if call_count["count"] == 1:
+            raise RuntimeError("First call fails")
+        return "success"
+
+    agent.tool_registry.register_tool(flaky_tool)
+
+    # Retry once on error (check result status, not exception attribute)
+    def retry_on_error(event):
+        if isinstance(event, AfterToolCallEvent) and event.result.get("status") == "error" and call_count["count"] == 1:
+            event.retry = True
+        return event
+
+    agent.hooks.add_callback(AfterToolCallEvent, retry_on_error)
+
+    tool_use: ToolUse = {"name": "flaky_tool", "toolUseId": "1", "input": {}}
+    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+    tru_events = await alist(stream)
+
+    # Tool called twice (1 exception + 1 success)
+    assert call_count["count"] == 2
+
+    # Final result is success
+    assert len(tru_events) == 1
+    assert tru_events[0].tool_result["status"] == "success"
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_retry_after_unknown_tool(executor, agent, tool_results, invocation_state, alist):
+    """Test that retry=True triggers retry loop for unknown tool.
+
+    Covers the unknown tool path retry check. Tool lookup happens before retry loop,
+    so even after retry the tool remains unknown - this test verifies the retry
+    mechanism is triggered, not that it resolves the unknown tool.
+    """
+    hook_call_count = {"count": 0}
+
+    # Retry once on first unknown tool error
+    def retry_once_on_unknown(event):
+        if isinstance(event, AfterToolCallEvent):
+            hook_call_count["count"] += 1
+            # Retry only on first call
+            if hook_call_count["count"] == 1:
+                event.retry = True
+        return event
+
+    agent.hooks.add_callback(AfterToolCallEvent, retry_once_on_unknown)
+
+    tool_use: ToolUse = {"name": "nonexistent_tool", "toolUseId": "1", "input": {}}
+    stream = executor._stream(agent, tool_use, tool_results, invocation_state)
+    tru_events = await alist(stream)
+
+    # Hook called twice (retry was triggered)
+    assert hook_call_count["count"] == 2
+
+    # Final result is still error (tool remains unknown after retry)
+    assert len(tru_events) == 1
+    assert tru_events[0].tool_result["status"] == "error"
+    assert "Unknown tool" in tru_events[0].tool_result["content"][0]["text"]

From ae3dc3eb49998be3af12c0996cb4424929043f5a Mon Sep 17 00:00:00 2001
From: Dean Schmigelski <dbschmigelski+github@gmail.com>
Date: Tue, 27 Jan 2026 12:38:17 -0500
Subject: [PATCH 7/7] docs: clarify getattr usage for BidiAfterToolCallEvent
 compatibility

---
 src/strands/tools/executors/_executor.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/strands/tools/executors/_executor.py b/src/strands/tools/executors/_executor.py
index dbb72ccdf..ef000fbd6 100644
--- a/src/strands/tools/executors/_executor.py
+++ b/src/strands/tools/executors/_executor.py
@@ -206,6 +206,7 @@ async def _stream(
                         agent, selected_tool, tool_use, invocation_state, result
                     )
                     # Check if retry requested for unknown tool error
+                    # Use getattr because BidiAfterToolCallEvent doesn't have retry attribute
                     if getattr(after_event, "retry", False):
                         logger.debug("tool_name=<%s> | retry requested, retrying tool call", tool_name)
                         continue
@@ -241,7 +242,7 @@ async def _stream(
                     agent, selected_tool, tool_use, invocation_state, result
                 )
 
-                # Check if retry requested
+                # Check if retry requested (getattr for BidiAfterToolCallEvent compatibility)
                 if getattr(after_event, "retry", False):
                     logger.debug("tool_name=<%s> | retry requested, retrying tool call", tool_name)
                     continue
@@ -261,7 +262,7 @@ async def _stream(
                 after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
                     agent, selected_tool, tool_use, invocation_state, error_result, exception=e
                 )
-                # Check if retry requested for exception
+                # Check if retry requested (getattr for BidiAfterToolCallEvent compatibility)
                 if getattr(after_event, "retry", False):
                     logger.debug("tool_name=<%s> | retry requested after exception, retrying tool call", tool_name)
                     continue