fix(bedrock): fold streaming tool input deltas back into tool args (#6149)

lib/crewai/src/crewai/llms/providers/bedrock/completion.py

@@ -50,6 +50,26 @@
 STRUCTURED_OUTPUT_TOOL_NAME = "structured_output"
 
 
+def _parse_streaming_tool_input(
+    accumulated: str, fallback: dict[str, Any]
+) -> dict[str, Any]:
+    """Resolve the final tool-call arguments from a Bedrock Converse stream.
+
+    Converse streams deliver `toolUse.input` as a sequence of partial JSON
+    strings on `contentBlockDelta`. The caller accumulates those into
+    ``accumulated`` and we parse it here at ``contentBlockStop``. When no
+    deltas arrived (some providers send the full input on the start block)
+    we fall back to the dict already attached to the tool-use block.
+    """
+    if accumulated:
+        try:
+            parsed = json.loads(accumulated)
+        except json.JSONDecodeError:
+            return fallback or {}
+        return parsed if isinstance(parsed, dict) else fallback or {}
+    return fallback or {}
+
+
 def _preprocess_structured_data(
     data: dict[str, Any], response_model: type[BaseModel]
 ) -> dict[str, Any]:
@@ -1034,8 +1054,14 @@ def _handle_streaming_converse(
                         logging.debug("Content block stopped in stream")
                         if current_tool_use:
                             function_name = current_tool_use["name"]
-                            function_args = cast(
-                                dict[str, Any], current_tool_use.get("input", {})
+                            # Streaming Converse delivers tool input as JSON
+                            # deltas in `accumulated_tool_input`; the start
+                            # event's `input` field is empty, so a direct
+                            # read of `current_tool_use["input"]` returned
+                            # {} for the entire stream (#6149).
+                            function_args = _parse_streaming_tool_input(
+                                accumulated_tool_input,
+                                cast(dict[str, Any], current_tool_use.get("input", {})),
                             )
 
                             # Check if this is the structured_output tool
@@ -1632,8 +1658,14 @@ async def _ahandle_streaming_converse(
                         logging.debug("Content block stopped in stream")
                         if current_tool_use:
                             function_name = current_tool_use["name"]
-                            function_args = cast(
-                                dict[str, Any], current_tool_use.get("input", {})
+                            # Streaming Converse delivers tool input as JSON
+                            # deltas in `accumulated_tool_input`; the start
+                            # event's `input` field is empty, so a direct
+                            # read of `current_tool_use["input"]` returned
+                            # {} for the entire stream (#6149).
+                            function_args = _parse_streaming_tool_input(
+                                accumulated_tool_input,
+                                cast(dict[str, Any], current_tool_use.get("input", {})),
                             )
 
                             # Check if this is the structured_output tool

lib/crewai/tests/llms/bedrock/test_bedrock.py

@@ -1185,3 +1185,68 @@ def test_bedrock_no_cache_tokens_defaults_to_zero():
 
         llm.call("Hello")
         assert llm._token_usage['cached_prompt_tokens'] == 0
+
+
+def _make_streaming_tool_use_events(tool_use_id, tool_name, input_chunks):
+    yield {
+        "contentBlockStart": {
+            "start": {
+                "toolUse": {
+                    "toolUseId": tool_use_id,
+                    "name": tool_name,
+                }
+            }
+        }
+    }
+    for chunk in input_chunks:
+        yield {
+            "contentBlockDelta": {
+                "delta": {"toolUse": {"input": chunk}}
+            }
+        }
+    yield {"contentBlockStop": {}}
+    yield {"messageStop": {"stopReason": "tool_use"}}
+
+
+def test_streaming_tool_call_accumulates_input_deltas(bedrock_mocks):
+    """Regression for #6149: streaming Converse must fold tool input deltas
+    back into the tool-call arguments instead of returning {}."""
+    _, mock_client = bedrock_mocks
+    mock_client.converse_stream.return_value = {
+        "stream": _make_streaming_tool_use_events(
+            tool_use_id="tu_test",
+            tool_name="get_weather",
+            input_chunks=['{"city":', ' "Paris"}'],
+        )
+    }
+
+    captured_args = {}
+
+    def get_weather(city: str) -> str:
+        captured_args["city"] = city
+        return f"Sunny in {city}"
+
+    llm = LLM(
+        model="bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0",
+        stream=True,
+    )
+    llm.call(
+        "What's the weather in Paris?",
+        available_functions={"get_weather": get_weather},
+        tools=[
+            {
+                "type": "function",
+                "function": {
+                    "name": "get_weather",
+                    "description": "Get the weather for a city.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"city": {"type": "string"}},
+                        "required": ["city"],
+                    },
+                },
+            }
+        ],
+    )
+
+    assert captured_args == {"city": "Paris"}