feat: skip model invocation when latest message contains ToolUse (#1068)

* feat: skip model invocation when latest message contains ToolUse

- Add _has_tool_use_in_latest_message() helper function to detect ToolUse in latest message
- Modify event_loop_cycle() to skip model execution when ToolUse is detected
- Set stop_reason='tool_use' and use latest message directly for tool execution
- Add comprehensive test coverage with 10 test scenarios
- Maintain backward compatibility and existing functionality
- No performance impact, minimal overhead for detection

Resolves the requirement to skip model calls when the agent should directly
execute tools based on existing ToolUse messages in the conversation.

🤖 Assisted by the code-assist agent script

* fix: Check messages array size

Author: Unshure

src/strands/event_loop/event_loop.py

@@ -33,7 +33,7 @@
     ToolResultMessageEvent,
     TypedEvent,
 )
-from ..types.content import Message
+from ..types.content import Message, Messages
 from ..types.exceptions import (
     ContextWindowOverflowException,
     EventLoopException,
@@ -56,6 +56,26 @@
 MAX_DELAY = 240  # 4 minutes
 
 
+def _has_tool_use_in_latest_message(messages: "Messages") -> bool:
+    """Check if the latest message contains any ToolUse content blocks.
+
+    Args:
+        messages: List of messages in the conversation.
+
+    Returns:
+        True if the latest message contains at least one ToolUse content block, False otherwise.
+    """
+    if len(messages) > 0:
+        latest_message = messages[-1]
+        content_blocks = latest_message.get("content", [])
+
+        for content_block in content_blocks:
+            if "toolUse" in content_block:
+                return True
+
+    return False
+
+
 async def event_loop_cycle(
     agent: "Agent",
     invocation_state: dict[str, Any],
@@ -121,7 +141,10 @@ async def event_loop_cycle(
     if agent._interrupt_state.activated:
         stop_reason: StopReason = "tool_use"
         message = agent._interrupt_state.context["tool_use_message"]
-
+    # Skip model invocation if the latest message contains ToolUse
+    elif _has_tool_use_in_latest_message(agent.messages):
+        stop_reason = "tool_use"
+        message = agent.messages[-1]
     else:
         model_events = _handle_model_execution(
             agent, cycle_span, cycle_trace, invocation_state, tracer, structured_output_context

tests/strands/agent/test_agent.py

@@ -2063,6 +2063,27 @@ def test_agent_tool_caller_interrupt(user):
         agent.tool.test_tool()
 
 
+def test_latest_message_tool_use_skips_model_invoke(tool_decorated):
+    mock_model = MockedModelProvider([{"role": "assistant", "content": [{"text": "I see the tool result"}]}])
+
+    messages: Messages = [
+        {
+            "role": "assistant",
+            "content": [
+                {"toolUse": {"toolUseId": "123", "name": "tool_decorated", "input": {"random_string": "Hello"}}}
+            ],
+        }
+    ]
+    agent = Agent(model=mock_model, tools=[tool_decorated], messages=messages)
+
+    agent()
+
+    assert mock_model.index == 1
+    assert len(agent.messages) == 3
+    assert agent.messages[1]["content"][0]["toolResult"]["content"][0]["text"] == "Hello"
+    assert agent.messages[2]["content"][0]["text"] == "I see the tool result"
+
+
 def test_agent_del_before_tool_registry_set():
     """Test that Agent.__del__ doesn't fail if called before tool_registry is set."""
     agent = Agent()