diff --git a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py index 0f34b67231..030eb47c12 100644 --- a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py +++ b/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 diff --git a/lib/crewai/tests/llms/bedrock/test_bedrock.py b/lib/crewai/tests/llms/bedrock/test_bedrock.py index d7421e852d..c1315b5c8c 100644 --- a/lib/crewai/tests/llms/bedrock/test_bedrock.py +++ b/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"}