feat(hooks): add retry mechanism for tool calls

src/strands/hooks/events.py

@@ -158,13 +158,28 @@ 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:
+
+        - 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.
         tool_use: The tool parameters that were passed to the tool invoked.
         invocation_state: Keyword arguments that were passed to the tool
         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:

src/strands/tools/executors/_executor.py

@@ -148,109 +148,127 @@ 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
+                    # 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
+                    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 (getattr for BidiAfterToolCallEvent compatibility)
+                if getattr(after_event, "retry", False):
+                    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 (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
+                yield ToolResultEvent(after_event.result)
+                tool_results.append(after_event.result)
+                return
 
     @staticmethod
     async def _stream_with_trace(