feat: Add tool retries and stop-on-failure logic to Python e2e tests

Author: clareliguori

e2e_tests/python/fail_on_tool_error.py

@@ -0,0 +1,28 @@
+import time
+from strands.hooks import HookProvider, HookRegistry, AfterToolCallEvent
+
+
+class FailOnToolError(HookProvider):
+    def __init__(self, max_retries: int = 3, retry_delay: float = 1.0):
+        self.max_retries = max_retries
+        self.retry_delay = retry_delay
+
+    def register_hooks(self, registry: HookRegistry) -> None:
+        registry.add_callback(AfterToolCallEvent, self.check_result)
+
+    def check_result(self, event: AfterToolCallEvent) -> None:
+        if event.result.get("isError", False):
+            for attempt in range(self.max_retries):
+                time.sleep(self.retry_delay)
+                try:
+                    # Retry the tool execution
+                    result = event.selected_tool(event.tool_use, event.invocation_state)
+                    if not result.get("isError", False):
+                        event.result = result
+                        return
+                except Exception:
+                    continue
+            
+            # All retries failed
+            error_content = event.result.get("content", [{}])[0].get("text", "Unknown error")
+            raise RuntimeError(f"Tool {event.tool_use['name']} failed after {self.max_retries} retries: {error_content}")

e2e_tests/python/main.py

@@ -12,6 +12,7 @@
     create_lambda_function_url_client,
     create_automated_oauth_client,
 )
+from fail_on_tool_error import FailOnToolError
 
 # Configure logging
 logging.basicConfig(
@@ -83,6 +84,7 @@ def main() -> None:
         model=bedrock_model,
         tools=mcp_clients,
         system_prompt="You are a helpful assistant.",
+        hooks=[FailOnToolError()],
     )
 
     # Run test utterances