From 2a4ca0e872ad9af5d839dd625aba310946b71f20 Mon Sep 17 00:00:00 2001 From: kimnamu Date: Sat, 13 Jun 2026 22:26:19 +0900 Subject: [PATCH 1/2] fix(bedrock): preserve streaming tool call arguments at contentBlockStop Streaming Converse handlers accumulate tool input as JSON string deltas in accumulated_tool_input but never fold it back into current_tool_use["input"], so function_args reads an empty {} at contentBlockStop. Parse the accumulated input into the tool-use block (with a {} fallback) in both the sync and async streaming handlers. This is the streaming counterpart of the non-streaming fix in #5415 (issue #4972). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../llms/providers/bedrock/completion.py | 20 +++ .../test_bedrock_streaming_tool_args.py | 129 ++++++++++++++++++ 2 files changed, 149 insertions(+) create mode 100644 lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py diff --git a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py index 0f34b67231..98dccfbdfb 100644 --- a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py +++ b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py @@ -1034,6 +1034,16 @@ def _handle_streaming_converse( logging.debug("Content block stopped in stream") if current_tool_use: function_name = current_tool_use["name"] + # Streamed tool input arrives as JSON string deltas in + # accumulated_tool_input; fold it back into the tool-use + # block so function_args (and the message history below) + # carry the real arguments instead of an empty input. + try: + current_tool_use["input"] = json.loads( + accumulated_tool_input + ) + except (json.JSONDecodeError, ValueError): + current_tool_use["input"] = {} function_args = cast( dict[str, Any], current_tool_use.get("input", {}) ) @@ -1632,6 +1642,16 @@ async def _ahandle_streaming_converse( logging.debug("Content block stopped in stream") if current_tool_use: function_name = current_tool_use["name"] + # Streamed tool input arrives as JSON string deltas in + # accumulated_tool_input; fold it back into the tool-use + # block so function_args (and the message history below) + # carry the real arguments instead of an empty input. + try: + current_tool_use["input"] = json.loads( + accumulated_tool_input + ) + except (json.JSONDecodeError, ValueError): + current_tool_use["input"] = {} function_args = cast( dict[str, Any], current_tool_use.get("input", {}) ) diff --git a/lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py b/lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py new file mode 100644 index 0000000000..264ca327eb --- /dev/null +++ b/lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py @@ -0,0 +1,129 @@ +"""Regression tests for streaming Bedrock tool-call argument handling. + +The streaming Converse handlers deliver tool input as a sequence of JSON +string deltas (``contentBlockDelta`` -> ``toolUse.input``) that are +accumulated separately from the tool-use block. These tests assert that the +accumulated input is folded back into the tool call at ``contentBlockStop``, +so executed tools receive their real arguments instead of an empty ``{}``. + +This is the streaming counterpart of the non-streaming fix in #5415 +(issue #4972). +""" + +import os +from unittest.mock import MagicMock, patch + +import pytest + +from crewai.llm import LLM +from crewai.llms.providers.bedrock.completion import BedrockCompletion + + +def _make_tool_use_stream() -> list[dict]: + """Synthetic Converse stream: a single tool call with JSON-chunked input.""" + # Tool input is delivered as two partial JSON string fragments that only + # form valid JSON once concatenated: '{"city":' + ' "Paris"}'. + chunk1 = '{"city":' + chunk2 = ' "Paris"}' + return [ + {"messageStart": {"role": "assistant"}}, + { + "contentBlockStart": { + "start": {"toolUse": {"toolUseId": "tool-1", "name": "get_weather"}}, + "contentBlockIndex": 0, + } + }, + {"contentBlockDelta": {"delta": {"toolUse": {"input": chunk1}}}}, + {"contentBlockDelta": {"delta": {"toolUse": {"input": chunk2}}}}, + {"contentBlockStop": {}}, + {"messageStop": {"stopReason": "tool_use"}}, + ] + + +def _build_completion() -> BedrockCompletion: + """Build a BedrockCompletion with mocked AWS credentials/session.""" + with patch.dict( + os.environ, + { + "AWS_ACCESS_KEY_ID": "test-access-key", + "AWS_SECRET_ACCESS_KEY": "test-secret-key", + "AWS_DEFAULT_REGION": "us-east-1", + }, + ): + with patch("crewai.llms.providers.bedrock.completion.Session"): + llm = LLM(model="bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0") + assert isinstance(llm, BedrockCompletion) + return llm + + +def test_streaming_tool_call_preserves_arguments(): + """Sync streaming: function_args must carry the streamed tool input.""" + llm = _build_completion() + + captured: dict = {} + + def capture(function_args, **kwargs): + captured["args"] = function_args + return None # returning None stops the recursive _handle_converse call + + mock_client = MagicMock() + mock_client.converse_stream.return_value = {"stream": _make_tool_use_stream()} + + with ( + patch.object(llm, "_get_sync_client", return_value=mock_client), + patch.object(llm, "_handle_tool_execution", side_effect=capture), + ): + llm._handle_streaming_converse( + messages=[{"role": "user", "content": "weather in Paris?"}], + body={}, + available_functions={"get_weather": lambda **kw: "sunny"}, + ) + + assert captured["args"] == {"city": "Paris"} + + +@pytest.mark.asyncio +async def test_async_streaming_tool_call_preserves_arguments(): + """Async streaming: function_args must carry the streamed tool input.""" + llm = _build_completion() + + class _AsyncStream: + def __init__(self, events): + self._events = events + + def __aiter__(self): + self._it = iter(self._events) + return self + + async def __anext__(self): + try: + return next(self._it) + except StopIteration: + raise StopAsyncIteration + + async def _converse_stream(**kwargs): + return {"stream": _AsyncStream(_make_tool_use_stream())} + + mock_async_client = MagicMock() + mock_async_client.converse_stream = _converse_stream + + async def _ensure(*args, **kwargs): + return mock_async_client + + captured: dict = {} + + def capture(function_args, **kwargs): + captured["args"] = function_args + return None + + with ( + patch.object(llm, "_ensure_async_client", side_effect=_ensure), + patch.object(llm, "_handle_tool_execution", side_effect=capture), + ): + await llm._ahandle_streaming_converse( + messages=[{"role": "user", "content": "weather in Paris?"}], + body={}, + available_functions={"get_weather": lambda **kw: "sunny"}, + ) + + assert captured["args"] == {"city": "Paris"} From a3697dfcd53c84a3803a9e9ec2cfd8d8f97b3fe9 Mon Sep 17 00:00:00 2001 From: kimnamu Date: Sun, 14 Jun 2026 06:53:12 +0900 Subject: [PATCH 2/2] fix(bedrock): coerce non-dict streaming tool input to empty dict json.loads on the accumulated tool input can return a valid-but-non-object JSON value (e.g. a string or list), which would fail at fn(**function_args) with a TypeError. Enforce a dict shape before use in both the sync and async streaming handlers, and add a regression test for the non-dict case. Addresses CodeRabbit review feedback on #6150. --- .../llms/providers/bedrock/completion.py | 18 ++++--- .../test_bedrock_streaming_tool_args.py | 53 +++++++++++++++++++ 2 files changed, 65 insertions(+), 6 deletions(-) diff --git a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py index 98dccfbdfb..33f86ac3b7 100644 --- a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py +++ b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py @@ -1039,10 +1039,13 @@ def _handle_streaming_converse( # block so function_args (and the message history below) # carry the real arguments instead of an empty input. try: - current_tool_use["input"] = json.loads( - accumulated_tool_input + parsed_input = json.loads(accumulated_tool_input) + current_tool_use["input"] = ( + parsed_input + if isinstance(parsed_input, dict) + else {} ) - except (json.JSONDecodeError, ValueError): + except (json.JSONDecodeError, ValueError, TypeError): current_tool_use["input"] = {} function_args = cast( dict[str, Any], current_tool_use.get("input", {}) @@ -1647,10 +1650,13 @@ async def _ahandle_streaming_converse( # block so function_args (and the message history below) # carry the real arguments instead of an empty input. try: - current_tool_use["input"] = json.loads( - accumulated_tool_input + parsed_input = json.loads(accumulated_tool_input) + current_tool_use["input"] = ( + parsed_input + if isinstance(parsed_input, dict) + else {} ) - except (json.JSONDecodeError, ValueError): + except (json.JSONDecodeError, ValueError, TypeError): current_tool_use["input"] = {} function_args = cast( dict[str, Any], current_tool_use.get("input", {}) diff --git a/lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py b/lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py index 264ca327eb..d9a435a4b1 100644 --- a/lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py +++ b/lib/crewai/tests/llms/bedrock/test_bedrock_streaming_tool_args.py @@ -40,6 +40,26 @@ def _make_tool_use_stream() -> list[dict]: ] +def _make_non_dict_tool_use_stream() -> list[dict]: + """Synthetic Converse stream whose tool input is valid JSON but not an object. + + ``json.loads`` succeeds here (returns a string), so the parsed value must + still be coerced to a dict before it reaches ``fn(**function_args)``. + """ + return [ + {"messageStart": {"role": "assistant"}}, + { + "contentBlockStart": { + "start": {"toolUse": {"toolUseId": "tool-1", "name": "get_weather"}}, + "contentBlockIndex": 0, + } + }, + {"contentBlockDelta": {"delta": {"toolUse": {"input": '"oops"'}}}}, + {"contentBlockStop": {}}, + {"messageStop": {"stopReason": "tool_use"}}, + ] + + def _build_completion() -> BedrockCompletion: """Build a BedrockCompletion with mocked AWS credentials/session.""" with patch.dict( @@ -127,3 +147,36 @@ def capture(function_args, **kwargs): ) assert captured["args"] == {"city": "Paris"} + + +def test_streaming_non_dict_tool_input_coerced_to_empty_dict(): + """Valid-but-non-object JSON input must be coerced to ``{}``. + + ``json.loads('"oops"')`` returns a string; passing it on as + ``fn(**function_args)`` would raise ``TypeError``. The handler must + guard against this and fall back to an empty dict. + """ + llm = _build_completion() + + captured: dict = {} + + def capture(function_args, **kwargs): + captured["args"] = function_args + return None + + mock_client = MagicMock() + mock_client.converse_stream.return_value = { + "stream": _make_non_dict_tool_use_stream() + } + + with ( + patch.object(llm, "_get_sync_client", return_value=mock_client), + patch.object(llm, "_handle_tool_execution", side_effect=capture), + ): + llm._handle_streaming_converse( + messages=[{"role": "user", "content": "weather in Paris?"}], + body={}, + available_functions={"get_weather": lambda **kw: "sunny"}, + ) + + assert captured["args"] == {}