fix: Populate tool_args correctly for steering (#1531)

Author: clareliguori

src/strands/experimental/steering/context_providers/ledger_provider.py

@@ -47,7 +47,7 @@ def __call__(self, event: BeforeToolCallEvent, steering_context: SteeringContext
         tool_call_entry = {
             "timestamp": datetime.now().isoformat(),
             "tool_name": event.tool_use.get("name"),
-            "tool_args": event.tool_use.get("arguments", {}),
+            "tool_args": event.tool_use.get("input", {}),
             "status": "pending",
         }
         ledger["tool_calls"].append(tool_call_entry)

tests/strands/experimental/steering/context_providers/test_ledger_provider.py

@@ -30,7 +30,7 @@ def test_ledger_before_tool_call_new_ledger(mock_datetime):
     callback = LedgerBeforeToolCall()
     steering_context = SteeringContext()
 
-    tool_use = {"name": "test_tool", "arguments": {"param": "value"}}
+    tool_use = {"name": "test_tool", "input": {"param": "value"}}
     event = Mock(spec=BeforeToolCallEvent)
     event.tool_use = tool_use
 
@@ -65,7 +65,7 @@ def test_ledger_before_tool_call_existing_ledger(mock_datetime):
     }
     steering_context.data.set("ledger", existing_ledger)
 
-    tool_use = {"name": "new_tool", "arguments": {"param": "value"}}
+    tool_use = {"name": "new_tool", "input": {"param": "value"}}
     event = Mock(spec=BeforeToolCallEvent)
     event.tool_use = tool_use
 

tests/strands/experimental/steering/core/test_handler.py

@@ -241,7 +241,7 @@ def test_context_callbacks_receive_steering_context():
 
     # Create a mock event and call the callback
     event = Mock(spec=BeforeToolCallEvent)
-    event.tool_use = {"name": "test_tool", "arguments": {}}
+    event.tool_use = {"name": "test_tool", "input": {}}
 
     # The callback should execute without error and update the steering context
     before_callback(event)

tests_integ/steering/test_model_steering.py

@@ -1,6 +1,7 @@
 """Integration tests for model steering (steer_after_model)."""
 
 from strands import Agent, tool
+from strands.experimental.steering.context_providers.ledger_provider import LedgerProvider
 from strands.experimental.steering.core.action import Guide, ModelSteeringAction, Proceed
 from strands.experimental.steering.core.handler import SteeringHandler
 from strands.types.content import Message
@@ -154,7 +155,7 @@ class ForceToolUsageHandler(SteeringHandler):
         """Handler that forces a specific tool to be used before allowing termination."""
 
         def __init__(self, required_tool: str):
-            super().__init__()
+            super().__init__(context_providers=[LedgerProvider()])
             self.required_tool = required_tool
             self.tool_was_used = False
             self.guidance_given = False
@@ -171,6 +172,15 @@ async def steer_after_model(
             for block in content_blocks:
                 if "toolUse" in block and block["toolUse"].get("name") == self.required_tool:
                     self.tool_was_used = True
+
+                    # Verify tool is in the ledger
+                    ledger = self.steering_context.data.get("ledger")
+                    if ledger:
+                        tool_calls = ledger.get("tool_calls", [])
+                        assert any(tc.get("tool_name") == self.required_tool for tc in tool_calls), (
+                            f"{self.required_tool} should be in ledger when tool_was_used=True"
+                        )
+
                     return Proceed(reason="Required tool was used")
 
             # If tool wasn't used and we haven't guided yet, force its usage

tests_integ/steering/test_tool_steering.py

@@ -3,7 +3,9 @@
 import pytest
 
 from strands import Agent, tool
+from strands.experimental.steering.context_providers.ledger_provider import LedgerProvider
 from strands.experimental.steering.core.action import Guide, Interrupt, Proceed
+from strands.experimental.steering.core.handler import SteeringHandler
 from strands.experimental.steering.handlers.llm.llm_handler import LLMSteeringHandler
 
 
@@ -98,3 +100,45 @@ def test_agent_with_tool_steering_e2e():
     notification_metrics = tool_metrics["send_notification"]
     assert notification_metrics.call_count >= 1, "send_notification should have been called"
     assert notification_metrics.success_count >= 1, "send_notification should have succeeded"
+
+
+def test_ledger_captures_tool_calls():
+    """Test that ledger correctly captures tool call information."""
+
+    class LedgerCheckingHandler(SteeringHandler):
+        def __init__(self):
+            super().__init__(context_providers=[LedgerProvider()])
+
+        async def steer_before_tool(self, *, agent, tool_use, **kwargs):
+            ledger = self.steering_context.data.get("ledger")
+            assert ledger is not None, "Ledger should exist"
+            assert "tool_calls" in ledger, "Ledger should have tool_calls"
+
+            # Find the current tool call in the ledger
+            tool_calls = ledger["tool_calls"]
+            current_call = next((tc for tc in tool_calls if tc["tool_name"] == tool_use["name"]), None)
+            assert current_call is not None, f"{tool_use['name']} should be in ledger"
+            assert current_call["tool_args"] == tool_use["input"], "tool_args should match input"
+            assert current_call["status"] == "pending", "Status should be pending before execution"
+
+            return Proceed(reason="Ledger verified")
+
+    handler = LedgerCheckingHandler()
+    agent = Agent(tools=[send_notification], hooks=[handler])
+
+    agent("Send a notification to alice saying test message")
+
+    # Verify the ledger has the completed tool call
+    ledger = handler.steering_context.data.get("ledger")
+    assert ledger is not None
+    assert len(ledger["tool_calls"]) >= 1, "At least one tool call should be recorded"
+
+    # Check the tool call details
+    tool_call = ledger["tool_calls"][-1]
+    assert tool_call["tool_name"] == "send_notification"
+    assert "tool_args" in tool_call
+    assert tool_call["tool_args"]["recipient"] == "alice"
+    assert tool_call["tool_args"]["message"] == "test message"
+    assert tool_call["status"] == "success"
+    assert "completion_timestamp" in tool_call
+    assert tool_call["error"] is None