From fd3eda8fbb61d6616c43a7d7603830736411ad09 Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 12:41:09 -0300 Subject: [PATCH 01/12] feat: preserve thoughtSignature for Gemini 3 Pro function calling - Add thoughtSignature field to ToolUse TypedDict as optional field - Add thoughtSignature to ContentBlockStartToolUse TypedDict - Preserve thoughtSignature during streaming event processing - Fixes compatibility with Gemini 3 Pro thinking mode This change enables proper multi-turn function calling with Gemini 3 Pro, which requires thought_signature to be passed back in subsequent requests. Resolves: Gemini 3 Pro 400 error for missing thought_signature See: https://ai.google.dev/gemini-api/docs/thought-signatures --- src/strands/event_loop/streaming.py | 8 ++++++++ src/strands/types/content.py | 4 +++- src/strands/types/tools.py | 7 ++++++- 3 files changed, 17 insertions(+), 2 deletions(-) diff --git a/src/strands/event_loop/streaming.py b/src/strands/event_loop/streaming.py index 43836fe34..ab1e57b65 100644 --- a/src/strands/event_loop/streaming.py +++ b/src/strands/event_loop/streaming.py @@ -185,6 +185,9 @@ def handle_content_block_start(event: ContentBlockStartEvent) -> dict[str, Any]: current_tool_use["toolUseId"] = tool_use_data["toolUseId"] current_tool_use["name"] = tool_use_data["name"] current_tool_use["input"] = "" + # Preserve thoughtSignature if present (required for Gemini 3 Pro) + if "thoughtSignature" in tool_use_data: + current_tool_use["thoughtSignature"] = tool_use_data["thoughtSignature"] return current_tool_use @@ -285,6 +288,11 @@ def handle_content_block_stop(state: dict[str, Any]) -> dict[str, Any]: name=tool_use_name, input=current_tool_use["input"], ) + + # Preserve thoughtSignature if present (required for Gemini 3 Pro) + if "thoughtSignature" in current_tool_use: + tool_use["thoughtSignature"] = current_tool_use["thoughtSignature"] + content.append({"toolUse": tool_use}) state["current_tool_use"] = {} diff --git a/src/strands/types/content.py b/src/strands/types/content.py index 4d0bbe412..e87407ca8 100644 --- a/src/strands/types/content.py +++ b/src/strands/types/content.py @@ -123,16 +123,18 @@ class DeltaContent(TypedDict, total=False): toolUse: Dict[Literal["input"], str] -class ContentBlockStartToolUse(TypedDict): +class ContentBlockStartToolUse(TypedDict, total=False): """The start of a tool use block. Attributes: name: The name of the tool that the model is requesting to use. toolUseId: The ID for the tool request. + thoughtSignature: Optional encrypted token from Gemini for multi-turn reasoning. """ name: str toolUseId: str + thoughtSignature: NotRequired[str] class ContentBlockStart(TypedDict, total=False): diff --git a/src/strands/types/tools.py b/src/strands/types/tools.py index 8343647b2..27073f95e 100644 --- a/src/strands/types/tools.py +++ b/src/strands/types/tools.py @@ -52,7 +52,7 @@ class Tool(TypedDict): toolSpec: ToolSpec -class ToolUse(TypedDict): +class ToolUse(TypedDict, total=False): """A request from the model to use a specific tool with the provided input. Attributes: @@ -60,11 +60,16 @@ class ToolUse(TypedDict): Can be any JSON-serializable type. name: The name of the tool to invoke. toolUseId: A unique identifier for this specific tool use request. + thoughtSignature: Optional encrypted token from Gemini that preserves + the model's internal reasoning process for multi-turn conversations. + Required for Gemini 3 Pro when using function calling. + See: https://ai.google.dev/gemini-api/docs/thought-signatures """ input: Any name: str toolUseId: str + thoughtSignature: NotRequired[str] class ToolResultContent(TypedDict, total=False): From 9730a99b4061669542c065418ac0eb292555412f Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 12:55:58 -0300 Subject: [PATCH 02/12] feat: add Gemini thoughtSignature capture and handling - Capture thought_signature from Gemini function call responses - Base64 encode thought_signature for storage in message history - Decode and pass thought_signature back to Gemini in subsequent requests - Configure thinking_config to disable thinking text but preserve signatures - Add NotRequired import to content.py for type safety This complements the framework changes by implementing Gemini-specific handling of thought signatures for proper multi-turn function calling with Gemini 3 Pro. See: https://ai.google.dev/gemini-api/docs/thought-signatures --- pyproject.toml | 8 ++--- src/strands/models/gemini.py | 67 ++++++++++++++++++++++++++++++++---- src/strands/types/content.py | 2 +- 3 files changed, 65 insertions(+), 12 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index b542c7481..0e1564ebc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [build-system] -requires = ["hatchling", "hatch-vcs"] +requires = ["hatchling"] # Removed hatch-vcs for manual version testing build-backend = "hatchling.build" [project] name = "strands-agents" -dynamic = ["version"] # Version determined by git tags +version = "1.18.0dev" # Temporary override for testing thought_signature fix description = "A model-driven approach to building AI agents in just a few lines of code" readme = "README.md" requires-python = ">=3.10" @@ -94,8 +94,8 @@ Documentation = "https://strandsagents.com" packages = ["src/strands"] -[tool.hatch.version] -source = "vcs" # Use git tags for versioning +# [tool.hatch.version] +# source = "vcs" # Temporarily disabled for testing - using manual version [tool.hatch.envs.hatch-static-analysis] diff --git a/src/strands/models/gemini.py b/src/strands/models/gemini.py index c24d91a0d..04318ed23 100644 --- a/src/strands/models/gemini.py +++ b/src/strands/models/gemini.py @@ -3,6 +3,7 @@ - Docs: https://ai.google.dev/api """ +import base64 import json import logging import mimetypes @@ -141,12 +142,26 @@ def _format_request_content_part(self, content: ContentBlock) -> genai.types.Par ) if "toolUse" in content: + thought_signature_b64 = cast(Optional[str], content["toolUse"].get("thoughtSignature")) + + thought_signature = None + if thought_signature_b64: + try: + thought_signature = base64.b64decode(thought_signature_b64) + except Exception as e: + logger.error("toolUseId=<%s> | failed to decode thoughtSignature: %s", content["toolUse"].get("toolUseId"), e) + else: + # thoughtSignature is now preserved by the Strands framework (as of v1.18+) + # If missing, it means the model didn't provide one (e.g., older Gemini versions) + logger.debug("toolUseId=<%s> | no thoughtSignature in toolUse (model may not require it)", content["toolUse"].get("toolUseId")) + return genai.types.Part( function_call=genai.types.FunctionCall( args=content["toolUse"]["input"], id=content["toolUse"]["toolUseId"], name=content["toolUse"]["name"], ), + thought_signature=thought_signature, ) raise TypeError(f"content_type=<{next(iter(content))}> | unsupported type") @@ -212,9 +227,19 @@ def _format_request_config( Returns: Gemini request config. """ + # Disable thinking text output when tools are present + # Note: Setting include_thoughts=False prevents thinking text in responses but + # Gemini still returns thought_signature for function calls. As of Strands v1.18+, + # the framework properly preserves this field through the message history. + # See: https://ai.google.dev/gemini-api/docs/thought-signatures + thinking_config = None + if tool_specs: + thinking_config = genai.types.ThinkingConfig(include_thoughts=False) + return genai.types.GenerateContentConfig( system_instruction=system_prompt, tools=self._format_request_tools(tool_specs), + thinking_config=thinking_config, **(params or {}), ) @@ -268,14 +293,24 @@ def _format_chunk(self, event: dict[str, Any]) -> StreamEvent: # that name be set in the equivalent FunctionResponse type. Consequently, we assign # function name to toolUseId in our tool use block. And another reason, function_call is # not guaranteed to have id populated. + tool_use: dict[str, Any] = { + "name": event["data"].function_call.name, + "toolUseId": event["data"].function_call.name, + } + + # Get thought_signature from the event dict (passed from stream method) + thought_sig = event.get("thought_signature") + + if thought_sig: + # Ensure it's bytes for encoding + if isinstance(thought_sig, str): + thought_sig = thought_sig.encode("utf-8") + # Use base64 encoding for storage + tool_use["thoughtSignature"] = base64.b64encode(thought_sig).decode("utf-8") + return { "contentBlockStart": { - "start": { - "toolUse": { - "name": event["data"].function_call.name, - "toolUseId": event["data"].function_call.name, - }, - }, + "start": {"toolUse": cast(Any, tool_use)}, }, } @@ -373,6 +408,10 @@ async def stream( yield self._format_chunk({"chunk_type": "content_start", "data_type": "text"}) tool_used = False + # Track thought_signature to associate with function calls + # According to Gemini docs, thought_signature can be on any part + last_thought_signature: Optional[bytes] = None + async for event in response: candidates = event.candidates candidate = candidates[0] if candidates else None @@ -380,8 +419,22 @@ async def stream( parts = content.parts if content and content.parts else [] for part in parts: + # Check ALL parts for thought_signature (Gemini may still include it even with thinking disabled) + if hasattr(part, "thought_signature") and part.thought_signature: + last_thought_signature = part.thought_signature + if part.function_call: - yield self._format_chunk({"chunk_type": "content_start", "data_type": "tool", "data": part}) + # Use the last thought_signature captured + effective_thought_signature = last_thought_signature + + yield self._format_chunk( + { + "chunk_type": "content_start", + "data_type": "tool", + "data": part, + "thought_signature": effective_thought_signature, + } + ) yield self._format_chunk({"chunk_type": "content_delta", "data_type": "tool", "data": part}) yield self._format_chunk({"chunk_type": "content_stop", "data_type": "tool", "data": part}) tool_used = True diff --git a/src/strands/types/content.py b/src/strands/types/content.py index e87407ca8..e4efc9208 100644 --- a/src/strands/types/content.py +++ b/src/strands/types/content.py @@ -8,7 +8,7 @@ from typing import Dict, List, Literal, Optional -from typing_extensions import TypedDict +from typing_extensions import NotRequired, TypedDict from .citations import CitationsContentBlock from .media import DocumentContent, ImageContent, VideoContent From 73f9bc38db3cc88f4ac623d4f60df57364c22220 Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 13:17:10 -0300 Subject: [PATCH 03/12] fix: resolve mypy type checking errors - Fix variable name conflict with thought_signature - Break long lines to comply with 120 character limit - Use explicit type annotations for thought signature variables --- src/strands/models/gemini.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/strands/models/gemini.py b/src/strands/models/gemini.py index 04318ed23..0dccfa12b 100644 --- a/src/strands/models/gemini.py +++ b/src/strands/models/gemini.py @@ -142,18 +142,20 @@ def _format_request_content_part(self, content: ContentBlock) -> genai.types.Par ) if "toolUse" in content: - thought_signature_b64 = cast(Optional[str], content["toolUse"].get("thoughtSignature")) + thought_signature_b64 = content["toolUse"].get("thoughtSignature") - thought_signature = None + tool_use_thought_signature: Optional[bytes] = None if thought_signature_b64: try: - thought_signature = base64.b64decode(thought_signature_b64) + tool_use_thought_signature = base64.b64decode(thought_signature_b64) except Exception as e: - logger.error("toolUseId=<%s> | failed to decode thoughtSignature: %s", content["toolUse"].get("toolUseId"), e) + tool_use_id = content["toolUse"].get("toolUseId") + logger.error("toolUseId=<%s> | failed to decode thoughtSignature: %s", tool_use_id, e) else: # thoughtSignature is now preserved by the Strands framework (as of v1.18+) # If missing, it means the model didn't provide one (e.g., older Gemini versions) - logger.debug("toolUseId=<%s> | no thoughtSignature in toolUse (model may not require it)", content["toolUse"].get("toolUseId")) + tool_use_id = content["toolUse"].get("toolUseId") + logger.debug("toolUseId=<%s> | no thoughtSignature in toolUse (model may not require it)", tool_use_id) return genai.types.Part( function_call=genai.types.FunctionCall( @@ -161,7 +163,7 @@ def _format_request_content_part(self, content: ContentBlock) -> genai.types.Par id=content["toolUse"]["toolUseId"], name=content["toolUse"]["name"], ), - thought_signature=thought_signature, + thought_signature=tool_use_thought_signature, ) raise TypeError(f"content_type=<{next(iter(content))}> | unsupported type") From 944b0e1cb7ca4363934bbb1d0eaf42d0c0860eda Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 13:17:50 -0300 Subject: [PATCH 04/12] style: apply code formatting (remove trailing whitespace) --- pyproject.toml | 1 + src/strands/models/gemini.py | 16 ++++++++-------- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 0e1564ebc..426e7b3bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ dependencies = [ "opentelemetry-api>=1.30.0,<2.0.0", "opentelemetry-sdk>=1.30.0,<2.0.0", "opentelemetry-instrumentation-threading>=0.51b0,<1.00b0", + "hatch>=1.15.1", ] diff --git a/src/strands/models/gemini.py b/src/strands/models/gemini.py index 0dccfa12b..5069335ef 100644 --- a/src/strands/models/gemini.py +++ b/src/strands/models/gemini.py @@ -143,7 +143,7 @@ def _format_request_content_part(self, content: ContentBlock) -> genai.types.Par if "toolUse" in content: thought_signature_b64 = content["toolUse"].get("thoughtSignature") - + tool_use_thought_signature: Optional[bytes] = None if thought_signature_b64: try: @@ -237,7 +237,7 @@ def _format_request_config( thinking_config = None if tool_specs: thinking_config = genai.types.ThinkingConfig(include_thoughts=False) - + return genai.types.GenerateContentConfig( system_instruction=system_prompt, tools=self._format_request_tools(tool_specs), @@ -299,17 +299,17 @@ def _format_chunk(self, event: dict[str, Any]) -> StreamEvent: "name": event["data"].function_call.name, "toolUseId": event["data"].function_call.name, } - + # Get thought_signature from the event dict (passed from stream method) thought_sig = event.get("thought_signature") - + if thought_sig: # Ensure it's bytes for encoding if isinstance(thought_sig, str): thought_sig = thought_sig.encode("utf-8") # Use base64 encoding for storage tool_use["thoughtSignature"] = base64.b64encode(thought_sig).decode("utf-8") - + return { "contentBlockStart": { "start": {"toolUse": cast(Any, tool_use)}, @@ -413,7 +413,7 @@ async def stream( # Track thought_signature to associate with function calls # According to Gemini docs, thought_signature can be on any part last_thought_signature: Optional[bytes] = None - + async for event in response: candidates = event.candidates candidate = candidates[0] if candidates else None @@ -424,11 +424,11 @@ async def stream( # Check ALL parts for thought_signature (Gemini may still include it even with thinking disabled) if hasattr(part, "thought_signature") and part.thought_signature: last_thought_signature = part.thought_signature - + if part.function_call: # Use the last thought_signature captured effective_thought_signature = last_thought_signature - + yield self._format_chunk( { "chunk_type": "content_start", From 542437201b788c2bc5ba78ce491949890ba9d192 Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 13:53:24 -0300 Subject: [PATCH 05/12] Unit test implementation --- src/strands/models/gemini.py | 16 +- tests/strands/event_loop/test_streaming.py | 220 +++++++++++++++++++++ tests/strands/models/test_gemini.py | 193 ++++++++++++++++++ tests/strands/types/test_content.py | 97 +++++++++ tests/strands/types/test_tools.py | 108 ++++++++++ 5 files changed, 626 insertions(+), 8 deletions(-) create mode 100644 tests/strands/types/test_content.py create mode 100644 tests/strands/types/test_tools.py diff --git a/src/strands/models/gemini.py b/src/strands/models/gemini.py index 5069335ef..0dccfa12b 100644 --- a/src/strands/models/gemini.py +++ b/src/strands/models/gemini.py @@ -143,7 +143,7 @@ def _format_request_content_part(self, content: ContentBlock) -> genai.types.Par if "toolUse" in content: thought_signature_b64 = content["toolUse"].get("thoughtSignature") - + tool_use_thought_signature: Optional[bytes] = None if thought_signature_b64: try: @@ -237,7 +237,7 @@ def _format_request_config( thinking_config = None if tool_specs: thinking_config = genai.types.ThinkingConfig(include_thoughts=False) - + return genai.types.GenerateContentConfig( system_instruction=system_prompt, tools=self._format_request_tools(tool_specs), @@ -299,17 +299,17 @@ def _format_chunk(self, event: dict[str, Any]) -> StreamEvent: "name": event["data"].function_call.name, "toolUseId": event["data"].function_call.name, } - + # Get thought_signature from the event dict (passed from stream method) thought_sig = event.get("thought_signature") - + if thought_sig: # Ensure it's bytes for encoding if isinstance(thought_sig, str): thought_sig = thought_sig.encode("utf-8") # Use base64 encoding for storage tool_use["thoughtSignature"] = base64.b64encode(thought_sig).decode("utf-8") - + return { "contentBlockStart": { "start": {"toolUse": cast(Any, tool_use)}, @@ -413,7 +413,7 @@ async def stream( # Track thought_signature to associate with function calls # According to Gemini docs, thought_signature can be on any part last_thought_signature: Optional[bytes] = None - + async for event in response: candidates = event.candidates candidate = candidates[0] if candidates else None @@ -424,11 +424,11 @@ async def stream( # Check ALL parts for thought_signature (Gemini may still include it even with thinking disabled) if hasattr(part, "thought_signature") and part.thought_signature: last_thought_signature = part.thought_signature - + if part.function_call: # Use the last thought_signature captured effective_thought_signature = last_thought_signature - + yield self._format_chunk( { "chunk_type": "content_start", diff --git a/tests/strands/event_loop/test_streaming.py b/tests/strands/event_loop/test_streaming.py index 3f5a6c998..c79a7b383 100644 --- a/tests/strands/event_loop/test_streaming.py +++ b/tests/strands/event_loop/test_streaming.py @@ -124,6 +124,10 @@ def test_handle_message_start(): {"start": {"toolUse": {"toolUseId": "test", "name": "test"}}}, {"toolUseId": "test", "name": "test", "input": ""}, ), + ( + {"start": {"toolUse": {"toolUseId": "test", "name": "test", "thoughtSignature": "YWJj"}}}, + {"toolUseId": "test", "name": "test", "input": "", "thoughtSignature": "YWJj"}, + ), ], ) def test_handle_content_block_start(chunk: ContentBlockStartEvent, exp_tool_use): @@ -132,6 +136,31 @@ def test_handle_content_block_start(chunk: ContentBlockStartEvent, exp_tool_use) assert tru_tool_use == exp_tool_use +def test_handle_content_block_start_with_thought_signature(): + """Test that thoughtSignature is preserved when starting tool use block.""" + chunk: ContentBlockStartEvent = { + "start": { + "toolUse": { + "toolUseId": "test-id", + "name": "test_tool", + "thoughtSignature": "dGVzdF9zaWduYXR1cmU=", + } + } + } + + tru_tool_use = strands.event_loop.streaming.handle_content_block_start(chunk) + exp_tool_use = { + "toolUseId": "test-id", + "name": "test_tool", + "input": "", + "thoughtSignature": "dGVzdF9zaWduYXR1cmU=", + } + + assert tru_tool_use == exp_tool_use + assert "thoughtSignature" in tru_tool_use + assert tru_tool_use["thoughtSignature"] == "dGVzdF9zaWduYXR1cmU=" + + @pytest.mark.parametrize( ("event", "state", "exp_updated_state", "callback_args"), [ @@ -245,6 +274,39 @@ def test_handle_content_block_delta(event: ContentBlockDeltaEvent, state, exp_up "redactedContent": b"", }, ), + # Tool Use - With thoughtSignature + ( + { + "content": [], + "current_tool_use": { + "toolUseId": "123", + "name": "test", + "input": '{"key": "value"}', + "thoughtSignature": "dGVzdF9zaWduYXR1cmU=", + }, + "text": "", + "reasoningText": "", + "citationsContent": [], + "redactedContent": b"", + }, + { + "content": [ + { + "toolUse": { + "toolUseId": "123", + "name": "test", + "input": {"key": "value"}, + "thoughtSignature": "dGVzdF9zaWduYXR1cmU=", + } + } + ], + "current_tool_use": {}, + "text": "", + "reasoningText": "", + "citationsContent": [], + "redactedContent": b"", + }, + ), # Tool Use - Missing input ( { @@ -1058,3 +1120,161 @@ async def test_stream_messages_normalizes_messages(agenerator, alist): {"content": [{"toolUse": {"name": "INVALID_TOOL_NAME"}}], "role": "assistant"}, {"content": [{"toolUse": {"name": "INVALID_TOOL_NAME"}}], "role": "assistant"}, ] + + +@pytest.mark.asyncio +async def test_process_stream_preserves_thought_signature(agenerator, alist): + """Test that thoughtSignature is preserved through the entire streaming pipeline.""" + response = [ + {"messageStart": {"role": "assistant"}}, + { + "contentBlockStart": { + "start": { + "toolUse": { + "toolUseId": "calculator-123", + "name": "calculator", + "thoughtSignature": "dGVzdF9zaWduYXR1cmVfYnl0ZXM=", + } + } + }, + }, + { + "contentBlockDelta": {"delta": {"toolUse": {"input": '{"expression": "2+2"}'}}}, + }, + {"contentBlockStop": {}}, + { + "messageStop": {"stopReason": "tool_use"}, + }, + { + "metadata": { + "usage": {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15}, + "metrics": {"latencyMs": 100}, + } + }, + ] + + stream = strands.event_loop.streaming.process_stream(agenerator(response)) + + last_event = cast(ModelStopReason, (await alist(stream))[-1]) + message = _get_message_from_event(last_event) + + # Verify the message has the tool use with thoughtSignature preserved + assert len(message["content"]) == 1 + assert "toolUse" in message["content"][0] + tool_use = message["content"][0]["toolUse"] + assert tool_use["toolUseId"] == "calculator-123" + assert tool_use["name"] == "calculator" + assert tool_use["input"] == {"expression": "2+2"} + assert "thoughtSignature" in tool_use + assert tool_use["thoughtSignature"] == "dGVzdF9zaWduYXR1cmVfYnl0ZXM=" + + +@pytest.mark.asyncio +async def test_process_stream_tool_use_without_thought_signature(agenerator, alist): + """Test that tool use works correctly when thoughtSignature is not present.""" + response = [ + {"messageStart": {"role": "assistant"}}, + { + "contentBlockStart": { + "start": { + "toolUse": { + "toolUseId": "calculator-123", + "name": "calculator", + # No thoughtSignature + } + } + }, + }, + { + "contentBlockDelta": {"delta": {"toolUse": {"input": '{"expression": "2+2"}'}}}, + }, + {"contentBlockStop": {}}, + { + "messageStop": {"stopReason": "tool_use"}, + }, + { + "metadata": { + "usage": {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15}, + "metrics": {"latencyMs": 100}, + } + }, + ] + + stream = strands.event_loop.streaming.process_stream(agenerator(response)) + + last_event = cast(ModelStopReason, (await alist(stream))[-1]) + message = _get_message_from_event(last_event) + + # Verify the message has the tool use without thoughtSignature + assert len(message["content"]) == 1 + assert "toolUse" in message["content"][0] + tool_use = message["content"][0]["toolUse"] + assert tool_use["toolUseId"] == "calculator-123" + assert tool_use["name"] == "calculator" + assert tool_use["input"] == {"expression": "2+2"} + assert "thoughtSignature" not in tool_use + + +@pytest.mark.asyncio +async def test_process_stream_multiple_tool_uses_with_thought_signatures(agenerator, alist): + """Test that multiple tool uses each preserve their thoughtSignature.""" + response = [ + {"messageStart": {"role": "assistant"}}, + { + "contentBlockStart": { + "start": { + "toolUse": { + "toolUseId": "tool1", + "name": "calculator", + "thoughtSignature": "c2lnbmF0dXJlMQ==", + } + } + }, + }, + { + "contentBlockDelta": {"delta": {"toolUse": {"input": '{"expression": "2+2"}'}}}, + }, + {"contentBlockStop": {}}, + { + "contentBlockStart": { + "start": { + "toolUse": { + "toolUseId": "tool2", + "name": "weather", + "thoughtSignature": "c2lnbmF0dXJlMg==", + } + } + }, + }, + { + "contentBlockDelta": {"delta": {"toolUse": {"input": '{"city": "SF"}'}}}, + }, + {"contentBlockStop": {}}, + { + "messageStop": {"stopReason": "tool_use"}, + }, + { + "metadata": { + "usage": {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15}, + "metrics": {"latencyMs": 100}, + } + }, + ] + + stream = strands.event_loop.streaming.process_stream(agenerator(response)) + + last_event = cast(ModelStopReason, (await alist(stream))[-1]) + message = _get_message_from_event(last_event) + + # Verify both tool uses have their respective thoughtSignatures + assert len(message["content"]) == 2 + + tool_use1 = message["content"][0]["toolUse"] + assert tool_use1["toolUseId"] == "tool1" + assert tool_use1["name"] == "calculator" + assert tool_use1["thoughtSignature"] == "c2lnbmF0dXJlMQ==" + + tool_use2 = message["content"][1]["toolUse"] + assert tool_use2["toolUseId"] == "tool2" + assert tool_use2["name"] == "weather" + assert tool_use2["thoughtSignature"] == "c2lnbmF0dXJlMg==" diff --git a/tests/strands/models/test_gemini.py b/tests/strands/models/test_gemini.py index a8f5351cc..37c2f4fc0 100644 --- a/tests/strands/models/test_gemini.py +++ b/tests/strands/models/test_gemini.py @@ -258,6 +258,29 @@ async def test_stream_request_with_tool_spec(gemini_client, model, model_id, too gemini_client.aio.models.generate_content_stream.assert_called_with(**exp_request) +@pytest.mark.asyncio +async def test_stream_request_with_tool_spec_sets_thinking_config(gemini_client, model, model_id, tool_spec): + """Test that thinking_config is set to disable thinking text when tools are present.""" + await anext(model.stream([], [tool_spec])) + + # Get the actual call arguments + call_args = gemini_client.aio.models.generate_content_stream.call_args + config = call_args.kwargs.get("config") + + # Verify thinking_config is set correctly + assert config is not None + # Config might be a dict when mocked + if isinstance(config, dict): + assert "thinking_config" in config + thinking_config = config["thinking_config"] + assert thinking_config["include_thoughts"] is False + else: + assert hasattr(config, "thinking_config") + thinking_config = config.thinking_config + assert thinking_config is not None + assert thinking_config.include_thoughts is False + + @pytest.mark.asyncio async def test_stream_request_with_tool_use(gemini_client, model, model_id): messages = [ @@ -299,6 +322,76 @@ async def test_stream_request_with_tool_use(gemini_client, model, model_id): gemini_client.aio.models.generate_content_stream.assert_called_with(**exp_request) +@pytest.mark.asyncio +async def test_stream_request_with_tool_use_and_thought_signature(gemini_client, model, model_id): + """Test that thoughtSignature is properly decoded from base64 and passed to Gemini API.""" + messages = [ + { + "role": "assistant", + "content": [ + { + "toolUse": { + "toolUseId": "c1", + "name": "calculator", + "input": {"expression": "2+2"}, + "thoughtSignature": "YWJjZGVmZ2g=", # base64 encoded "abcdefgh" + }, + }, + ], + }, + ] + await anext(model.stream(messages)) + + # Verify that the call was made - Gemini SDK handles the thought_signature serialization internally + call_args = gemini_client.aio.models.generate_content_stream.call_args + assert call_args is not None + + # Check that the content includes the thought_signature (SDK may serialize it differently) + contents = call_args.kwargs["contents"] + assert len(contents) == 1 + assert len(contents[0]["parts"]) == 1 + part = contents[0]["parts"][0] + assert "function_call" in part + assert part["function_call"]["name"] == "calculator" + # The SDK handles thought_signature internally, verify it's present + assert "thought_signature" in part + + +@pytest.mark.asyncio +async def test_stream_request_with_tool_use_missing_thought_signature(gemini_client, model, model_id): + """Test that missing thoughtSignature is handled gracefully.""" + messages = [ + { + "role": "assistant", + "content": [ + { + "toolUse": { + "toolUseId": "c1", + "name": "calculator", + "input": {"expression": "2+2"}, + # No thoughtSignature + }, + }, + ], + }, + ] + await anext(model.stream(messages)) + + # Verify the call was made without thought_signature when not present + call_args = gemini_client.aio.models.generate_content_stream.call_args + assert call_args is not None + + contents = call_args.kwargs["contents"] + assert len(contents) == 1 + assert len(contents[0]["parts"]) == 1 + part = contents[0]["parts"][0] + assert "function_call" in part + assert part["function_call"]["name"] == "calculator" + # When thoughtSignature is missing, the SDK may omit it entirely + # This is acceptable behavior + assert "thought_signature" not in part or part.get("thought_signature") is None + + @pytest.mark.asyncio async def test_stream_request_with_tool_results(gemini_client, model, model_id): messages = [ @@ -469,6 +562,106 @@ async def test_stream_response_tool_use(gemini_client, model, messages, agenerat assert tru_chunks == exp_chunks +@pytest.mark.asyncio +async def test_stream_response_tool_use_with_thought_signature(gemini_client, model, messages, agenerator, alist): + """Test that thoughtSignature from Gemini response is captured and base64-encoded.""" + gemini_client.aio.models.generate_content_stream.return_value = agenerator( + [ + genai.types.GenerateContentResponse( + candidates=[ + genai.types.Candidate( + content=genai.types.Content( + parts=[ + genai.types.Part( + function_call=genai.types.FunctionCall( + args={"expression": "2+2"}, + id="c1", + name="calculator", + ), + thought_signature=b"test_signature_bytes", # Raw bytes from Gemini + ), + ], + ), + finish_reason="STOP", + ), + ], + usage_metadata=genai.types.GenerateContentResponseUsageMetadata( + prompt_token_count=1, + total_token_count=3, + ), + ), + ] + ) + + tru_chunks = await alist(model.stream(messages)) + exp_chunks = [ + {"messageStart": {"role": "assistant"}}, + {"contentBlockStart": {"start": {}}}, + { + "contentBlockStart": { + "start": { + "toolUse": { + "name": "calculator", + "toolUseId": "calculator", + "thoughtSignature": "dGVzdF9zaWduYXR1cmVfYnl0ZXM=", # base64 encoded + } + } + } + }, + {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"expression": "2+2"}'}}}}, + {"contentBlockStop": {}}, + {"contentBlockStop": {}}, + {"messageStop": {"stopReason": "tool_use"}}, + {"metadata": {"usage": {"inputTokens": 1, "outputTokens": 2, "totalTokens": 3}, "metrics": {"latencyMs": 0}}}, + ] + assert tru_chunks == exp_chunks + + +@pytest.mark.asyncio +async def test_stream_response_tool_use_without_thought_signature(gemini_client, model, messages, agenerator, alist): + """Test that missing thoughtSignature in response is handled gracefully.""" + gemini_client.aio.models.generate_content_stream.return_value = agenerator( + [ + genai.types.GenerateContentResponse( + candidates=[ + genai.types.Candidate( + content=genai.types.Content( + parts=[ + genai.types.Part( + function_call=genai.types.FunctionCall( + args={"expression": "2+2"}, + id="c1", + name="calculator", + ), + # No thought_signature + ), + ], + ), + finish_reason="STOP", + ), + ], + usage_metadata=genai.types.GenerateContentResponseUsageMetadata( + prompt_token_count=1, + total_token_count=3, + ), + ), + ] + ) + + tru_chunks = await alist(model.stream(messages)) + exp_chunks = [ + {"messageStart": {"role": "assistant"}}, + {"contentBlockStart": {"start": {}}}, + {"contentBlockStart": {"start": {"toolUse": {"name": "calculator", "toolUseId": "calculator"}}}}, + {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"expression": "2+2"}'}}}}, + {"contentBlockStop": {}}, + {"contentBlockStop": {}}, + {"messageStop": {"stopReason": "tool_use"}}, + {"metadata": {"usage": {"inputTokens": 1, "outputTokens": 2, "totalTokens": 3}, "metrics": {"latencyMs": 0}}}, + ] + assert tru_chunks == exp_chunks + + @pytest.mark.asyncio async def test_stream_response_reasoning(gemini_client, model, messages, agenerator, alist): gemini_client.aio.models.generate_content_stream.return_value = agenerator( diff --git a/tests/strands/types/test_content.py b/tests/strands/types/test_content.py new file mode 100644 index 000000000..4478727dc --- /dev/null +++ b/tests/strands/types/test_content.py @@ -0,0 +1,97 @@ +"""Tests for strands.types.content module.""" + +import pytest + +from strands.types.content import ContentBlockStartToolUse + + +def test_content_block_start_tool_use_required_fields(): + """Test that ContentBlockStartToolUse can be created with only required fields.""" + content_block: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + } + + assert content_block["toolUseId"] == "test-id" + assert content_block["name"] == "test_tool" + assert "thoughtSignature" not in content_block + + +def test_content_block_start_tool_use_with_thought_signature(): + """Test that ContentBlockStartToolUse can include optional thoughtSignature field.""" + content_block: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "thoughtSignature": "YWJjZGVmZ2g=", + } + + assert content_block["toolUseId"] == "test-id" + assert content_block["name"] == "test_tool" + assert content_block["thoughtSignature"] == "YWJjZGVmZ2g=" + + +def test_content_block_start_tool_use_thought_signature_is_optional(): + """Test that thoughtSignature is truly optional.""" + # Create with thoughtSignature + with_sig: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "thoughtSignature": "test", + } + assert "thoughtSignature" in with_sig + + # Create without thoughtSignature + without_sig: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + } + assert "thoughtSignature" not in without_sig + + +def test_content_block_start_tool_use_base64_encoded(): + """Test that thoughtSignature should be base64 encoded string.""" + content_block: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "thoughtSignature": "dGVzdF9zaWduYXR1cmVfYnl0ZXM=", + } + + assert content_block["thoughtSignature"] == "dGVzdF9zaWduYXR1cmVfYnl0ZXM=" + + +def test_content_block_start_tool_use_empty_signature(): + """Test that empty thoughtSignature is valid.""" + content_block: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "thoughtSignature": "", + } + + assert content_block["thoughtSignature"] == "" + + +def test_content_block_start_tool_use_special_characters_in_name(): + """Test that tool names with special characters work correctly.""" + content_block: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "default_api:vehicleDetails", + "thoughtSignature": "c2lnbmF0dXJl", + } + + assert content_block["name"] == "default_api:vehicleDetails" + assert content_block["thoughtSignature"] == "c2lnbmF0dXJl" + + +def test_content_block_start_tool_use_long_signature(): + """Test that long base64 encoded signatures are supported.""" + # Simulate a long signature (typical of Gemini's encrypted tokens) + long_signature = "dGVzdF9zaWduYXR1cmVfYnl0ZXNfdGhhdF9pc192ZXJ5X2xvbmdf" * 5 + content_block: ContentBlockStartToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "thoughtSignature": long_signature, + } + + assert content_block["thoughtSignature"] == long_signature + assert len(content_block["thoughtSignature"]) > 100 + diff --git a/tests/strands/types/test_tools.py b/tests/strands/types/test_tools.py new file mode 100644 index 000000000..ae9842c15 --- /dev/null +++ b/tests/strands/types/test_tools.py @@ -0,0 +1,108 @@ +"""Tests for strands.types.tools module.""" + +import pytest + +from strands.types.tools import ToolUse + + +def test_tool_use_required_fields(): + """Test that ToolUse can be created with only required fields.""" + tool_use: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": {"key": "value"}, + } + + assert tool_use["toolUseId"] == "test-id" + assert tool_use["name"] == "test_tool" + assert tool_use["input"] == {"key": "value"} + assert "thoughtSignature" not in tool_use + + +def test_tool_use_with_thought_signature(): + """Test that ToolUse can include optional thoughtSignature field.""" + tool_use: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": {"key": "value"}, + "thoughtSignature": "YWJjZGVmZ2g=", + } + + assert tool_use["toolUseId"] == "test-id" + assert tool_use["name"] == "test_tool" + assert tool_use["input"] == {"key": "value"} + assert tool_use["thoughtSignature"] == "YWJjZGVmZ2g=" + + +def test_tool_use_thought_signature_is_optional(): + """Test that thoughtSignature is truly optional and doesn't require all fields.""" + # Create with thoughtSignature + tool_use_with_sig: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": {}, + "thoughtSignature": "test", + } + assert "thoughtSignature" in tool_use_with_sig + + # Create without thoughtSignature + tool_use_without_sig: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": {}, + } + assert "thoughtSignature" not in tool_use_without_sig + + +def test_tool_use_empty_input(): + """Test that ToolUse works with empty input.""" + tool_use: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": {}, + } + + assert tool_use["input"] == {} + assert "thoughtSignature" not in tool_use + + +def test_tool_use_complex_input(): + """Test that ToolUse works with complex nested input.""" + tool_use: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": { + "nested": {"key": "value"}, + "array": [1, 2, 3], + "string": "test", + }, + "thoughtSignature": "c2lnbmF0dXJl", + } + + assert tool_use["input"]["nested"]["key"] == "value" + assert tool_use["input"]["array"] == [1, 2, 3] + assert tool_use["thoughtSignature"] == "c2lnbmF0dXJl" + + +def test_tool_use_base64_encoded_signature(): + """Test that thoughtSignature should be base64 encoded string.""" + # Valid base64 encoded signature + tool_use: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": {}, + "thoughtSignature": "dGVzdF9zaWduYXR1cmVfYnl0ZXM=", + } + + assert tool_use["thoughtSignature"] == "dGVzdF9zaWduYXR1cmVfYnl0ZXM=" + + # Empty signature should also be valid + tool_use_empty: ToolUse = { + "toolUseId": "test-id", + "name": "test_tool", + "input": {}, + "thoughtSignature": "", + } + + assert tool_use_empty["thoughtSignature"] == "" + From a20b7ea9bc27ca4e6fce5ab03761fe4fba2032db Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 14:13:28 -0300 Subject: [PATCH 06/12] chore: revert pyproject.toml to use dynamic versioning with hatch-vcs --- pyproject.toml | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 426e7b3bc..b542c7481 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [build-system] -requires = ["hatchling"] # Removed hatch-vcs for manual version testing +requires = ["hatchling", "hatch-vcs"] build-backend = "hatchling.build" [project] name = "strands-agents" -version = "1.18.0dev" # Temporary override for testing thought_signature fix +dynamic = ["version"] # Version determined by git tags description = "A model-driven approach to building AI agents in just a few lines of code" readme = "README.md" requires-python = ">=3.10" @@ -38,7 +38,6 @@ dependencies = [ "opentelemetry-api>=1.30.0,<2.0.0", "opentelemetry-sdk>=1.30.0,<2.0.0", "opentelemetry-instrumentation-threading>=0.51b0,<1.00b0", - "hatch>=1.15.1", ] @@ -95,8 +94,8 @@ Documentation = "https://strandsagents.com" packages = ["src/strands"] -# [tool.hatch.version] -# source = "vcs" # Temporarily disabled for testing - using manual version +[tool.hatch.version] +source = "vcs" # Use git tags for versioning [tool.hatch.envs.hatch-static-analysis] From a999c77f2f0c5f02566ee166b626c24a9ba5b29e Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 14:25:06 -0300 Subject: [PATCH 07/12] test: update test_stream_request_with_tool_spec to expect thinking_config --- tests/strands/models/test_gemini.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/strands/models/test_gemini.py b/tests/strands/models/test_gemini.py index 37c2f4fc0..6dbd81929 100644 --- a/tests/strands/models/test_gemini.py +++ b/tests/strands/models/test_gemini.py @@ -251,6 +251,7 @@ async def test_stream_request_with_tool_spec(gemini_client, model, model_id, too ], }, ], + "thinking_config": {"include_thoughts": False}, }, "contents": [], "model": model_id, From 59283424839ec0f2f3b4934705acebcf55cfd0b9 Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 14:27:54 -0300 Subject: [PATCH 08/12] style: remove trailing whitespace and unused imports --- src/strands/models/gemini.py | 16 ++++++++-------- tests/strands/types/test_content.py | 2 -- tests/strands/types/test_tools.py | 2 -- 3 files changed, 8 insertions(+), 12 deletions(-) diff --git a/src/strands/models/gemini.py b/src/strands/models/gemini.py index 0dccfa12b..5069335ef 100644 --- a/src/strands/models/gemini.py +++ b/src/strands/models/gemini.py @@ -143,7 +143,7 @@ def _format_request_content_part(self, content: ContentBlock) -> genai.types.Par if "toolUse" in content: thought_signature_b64 = content["toolUse"].get("thoughtSignature") - + tool_use_thought_signature: Optional[bytes] = None if thought_signature_b64: try: @@ -237,7 +237,7 @@ def _format_request_config( thinking_config = None if tool_specs: thinking_config = genai.types.ThinkingConfig(include_thoughts=False) - + return genai.types.GenerateContentConfig( system_instruction=system_prompt, tools=self._format_request_tools(tool_specs), @@ -299,17 +299,17 @@ def _format_chunk(self, event: dict[str, Any]) -> StreamEvent: "name": event["data"].function_call.name, "toolUseId": event["data"].function_call.name, } - + # Get thought_signature from the event dict (passed from stream method) thought_sig = event.get("thought_signature") - + if thought_sig: # Ensure it's bytes for encoding if isinstance(thought_sig, str): thought_sig = thought_sig.encode("utf-8") # Use base64 encoding for storage tool_use["thoughtSignature"] = base64.b64encode(thought_sig).decode("utf-8") - + return { "contentBlockStart": { "start": {"toolUse": cast(Any, tool_use)}, @@ -413,7 +413,7 @@ async def stream( # Track thought_signature to associate with function calls # According to Gemini docs, thought_signature can be on any part last_thought_signature: Optional[bytes] = None - + async for event in response: candidates = event.candidates candidate = candidates[0] if candidates else None @@ -424,11 +424,11 @@ async def stream( # Check ALL parts for thought_signature (Gemini may still include it even with thinking disabled) if hasattr(part, "thought_signature") and part.thought_signature: last_thought_signature = part.thought_signature - + if part.function_call: # Use the last thought_signature captured effective_thought_signature = last_thought_signature - + yield self._format_chunk( { "chunk_type": "content_start", diff --git a/tests/strands/types/test_content.py b/tests/strands/types/test_content.py index 4478727dc..e6dca2af5 100644 --- a/tests/strands/types/test_content.py +++ b/tests/strands/types/test_content.py @@ -1,6 +1,5 @@ """Tests for strands.types.content module.""" -import pytest from strands.types.content import ContentBlockStartToolUse @@ -94,4 +93,3 @@ def test_content_block_start_tool_use_long_signature(): assert content_block["thoughtSignature"] == long_signature assert len(content_block["thoughtSignature"]) > 100 - diff --git a/tests/strands/types/test_tools.py b/tests/strands/types/test_tools.py index ae9842c15..f030ba5dd 100644 --- a/tests/strands/types/test_tools.py +++ b/tests/strands/types/test_tools.py @@ -1,6 +1,5 @@ """Tests for strands.types.tools module.""" -import pytest from strands.types.tools import ToolUse @@ -105,4 +104,3 @@ def test_tool_use_base64_encoded_signature(): } assert tool_use_empty["thoughtSignature"] == "" - From a80c60233a661e739fdd1b6087958835c8a75ede Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 15:18:03 -0300 Subject: [PATCH 09/12] refactor: rename type test files to be more specific - Rename test_tools.py -> test_tool_use.py (tests ToolUse TypedDict) - Rename test_content.py -> test_content_block_start_tool_use.py (tests ContentBlockStartToolUse TypedDict) This makes the test file names more descriptive and avoids confusion with tests/strands/tools/test_tools.py --- .../{test_content.py => test_content_block_start_tool_use.py} | 0 tests/strands/types/{test_tools.py => test_tool_use.py} | 0 2 files changed, 0 insertions(+), 0 deletions(-) rename tests/strands/types/{test_content.py => test_content_block_start_tool_use.py} (100%) rename tests/strands/types/{test_tools.py => test_tool_use.py} (100%) diff --git a/tests/strands/types/test_content.py b/tests/strands/types/test_content_block_start_tool_use.py similarity index 100% rename from tests/strands/types/test_content.py rename to tests/strands/types/test_content_block_start_tool_use.py diff --git a/tests/strands/types/test_tools.py b/tests/strands/types/test_tool_use.py similarity index 100% rename from tests/strands/types/test_tools.py rename to tests/strands/types/test_tool_use.py From 3e23297a2b5a6db42138eca2b5e14999063609aa Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 15:49:59 -0300 Subject: [PATCH 10/12] Remove unnecessary TypedDict unit tests - Removed test_content_block_start_tool_use.py (~96 lines) - Removed test_tool_use.py (~107 lines) These tests only verified basic Python dict behavior without testing any SDK logic. All meaningful coverage is maintained by integration tests in test_streaming.py which test actual thoughtSignature handling through the streaming pipeline. --- .../test_content_block_start_tool_use.py | 95 ---------------- tests/strands/types/test_tool_use.py | 106 ------------------ 2 files changed, 201 deletions(-) delete mode 100644 tests/strands/types/test_content_block_start_tool_use.py delete mode 100644 tests/strands/types/test_tool_use.py diff --git a/tests/strands/types/test_content_block_start_tool_use.py b/tests/strands/types/test_content_block_start_tool_use.py deleted file mode 100644 index e6dca2af5..000000000 --- a/tests/strands/types/test_content_block_start_tool_use.py +++ /dev/null @@ -1,95 +0,0 @@ -"""Tests for strands.types.content module.""" - - -from strands.types.content import ContentBlockStartToolUse - - -def test_content_block_start_tool_use_required_fields(): - """Test that ContentBlockStartToolUse can be created with only required fields.""" - content_block: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - } - - assert content_block["toolUseId"] == "test-id" - assert content_block["name"] == "test_tool" - assert "thoughtSignature" not in content_block - - -def test_content_block_start_tool_use_with_thought_signature(): - """Test that ContentBlockStartToolUse can include optional thoughtSignature field.""" - content_block: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "thoughtSignature": "YWJjZGVmZ2g=", - } - - assert content_block["toolUseId"] == "test-id" - assert content_block["name"] == "test_tool" - assert content_block["thoughtSignature"] == "YWJjZGVmZ2g=" - - -def test_content_block_start_tool_use_thought_signature_is_optional(): - """Test that thoughtSignature is truly optional.""" - # Create with thoughtSignature - with_sig: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "thoughtSignature": "test", - } - assert "thoughtSignature" in with_sig - - # Create without thoughtSignature - without_sig: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - } - assert "thoughtSignature" not in without_sig - - -def test_content_block_start_tool_use_base64_encoded(): - """Test that thoughtSignature should be base64 encoded string.""" - content_block: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "thoughtSignature": "dGVzdF9zaWduYXR1cmVfYnl0ZXM=", - } - - assert content_block["thoughtSignature"] == "dGVzdF9zaWduYXR1cmVfYnl0ZXM=" - - -def test_content_block_start_tool_use_empty_signature(): - """Test that empty thoughtSignature is valid.""" - content_block: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "thoughtSignature": "", - } - - assert content_block["thoughtSignature"] == "" - - -def test_content_block_start_tool_use_special_characters_in_name(): - """Test that tool names with special characters work correctly.""" - content_block: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "default_api:vehicleDetails", - "thoughtSignature": "c2lnbmF0dXJl", - } - - assert content_block["name"] == "default_api:vehicleDetails" - assert content_block["thoughtSignature"] == "c2lnbmF0dXJl" - - -def test_content_block_start_tool_use_long_signature(): - """Test that long base64 encoded signatures are supported.""" - # Simulate a long signature (typical of Gemini's encrypted tokens) - long_signature = "dGVzdF9zaWduYXR1cmVfYnl0ZXNfdGhhdF9pc192ZXJ5X2xvbmdf" * 5 - content_block: ContentBlockStartToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "thoughtSignature": long_signature, - } - - assert content_block["thoughtSignature"] == long_signature - assert len(content_block["thoughtSignature"]) > 100 diff --git a/tests/strands/types/test_tool_use.py b/tests/strands/types/test_tool_use.py deleted file mode 100644 index f030ba5dd..000000000 --- a/tests/strands/types/test_tool_use.py +++ /dev/null @@ -1,106 +0,0 @@ -"""Tests for strands.types.tools module.""" - - -from strands.types.tools import ToolUse - - -def test_tool_use_required_fields(): - """Test that ToolUse can be created with only required fields.""" - tool_use: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": {"key": "value"}, - } - - assert tool_use["toolUseId"] == "test-id" - assert tool_use["name"] == "test_tool" - assert tool_use["input"] == {"key": "value"} - assert "thoughtSignature" not in tool_use - - -def test_tool_use_with_thought_signature(): - """Test that ToolUse can include optional thoughtSignature field.""" - tool_use: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": {"key": "value"}, - "thoughtSignature": "YWJjZGVmZ2g=", - } - - assert tool_use["toolUseId"] == "test-id" - assert tool_use["name"] == "test_tool" - assert tool_use["input"] == {"key": "value"} - assert tool_use["thoughtSignature"] == "YWJjZGVmZ2g=" - - -def test_tool_use_thought_signature_is_optional(): - """Test that thoughtSignature is truly optional and doesn't require all fields.""" - # Create with thoughtSignature - tool_use_with_sig: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": {}, - "thoughtSignature": "test", - } - assert "thoughtSignature" in tool_use_with_sig - - # Create without thoughtSignature - tool_use_without_sig: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": {}, - } - assert "thoughtSignature" not in tool_use_without_sig - - -def test_tool_use_empty_input(): - """Test that ToolUse works with empty input.""" - tool_use: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": {}, - } - - assert tool_use["input"] == {} - assert "thoughtSignature" not in tool_use - - -def test_tool_use_complex_input(): - """Test that ToolUse works with complex nested input.""" - tool_use: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": { - "nested": {"key": "value"}, - "array": [1, 2, 3], - "string": "test", - }, - "thoughtSignature": "c2lnbmF0dXJl", - } - - assert tool_use["input"]["nested"]["key"] == "value" - assert tool_use["input"]["array"] == [1, 2, 3] - assert tool_use["thoughtSignature"] == "c2lnbmF0dXJl" - - -def test_tool_use_base64_encoded_signature(): - """Test that thoughtSignature should be base64 encoded string.""" - # Valid base64 encoded signature - tool_use: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": {}, - "thoughtSignature": "dGVzdF9zaWduYXR1cmVfYnl0ZXM=", - } - - assert tool_use["thoughtSignature"] == "dGVzdF9zaWduYXR1cmVfYnl0ZXM=" - - # Empty signature should also be valid - tool_use_empty: ToolUse = { - "toolUseId": "test-id", - "name": "test_tool", - "input": {}, - "thoughtSignature": "", - } - - assert tool_use_empty["thoughtSignature"] == "" From 82f87017ad5fb656a0f7b7f27adfccd9b864a80d Mon Sep 17 00:00:00 2001 From: danielp Date: Fri, 21 Nov 2025 15:54:38 -0300 Subject: [PATCH 11/12] Remove redundant thoughtSignature test function The test_handle_content_block_start_with_thought_signature() function was redundant because the parameterized test already covers the same functionality (line 127-130). This standalone test just repeated the same assertions without testing any additional edge cases or coverage. Removed ~23 lines of duplicate test code while maintaining full coverage. --- tests/strands/event_loop/test_streaming.py | 25 ---------------------- 1 file changed, 25 deletions(-) diff --git a/tests/strands/event_loop/test_streaming.py b/tests/strands/event_loop/test_streaming.py index c79a7b383..5c8483632 100644 --- a/tests/strands/event_loop/test_streaming.py +++ b/tests/strands/event_loop/test_streaming.py @@ -136,31 +136,6 @@ def test_handle_content_block_start(chunk: ContentBlockStartEvent, exp_tool_use) assert tru_tool_use == exp_tool_use -def test_handle_content_block_start_with_thought_signature(): - """Test that thoughtSignature is preserved when starting tool use block.""" - chunk: ContentBlockStartEvent = { - "start": { - "toolUse": { - "toolUseId": "test-id", - "name": "test_tool", - "thoughtSignature": "dGVzdF9zaWduYXR1cmU=", - } - } - } - - tru_tool_use = strands.event_loop.streaming.handle_content_block_start(chunk) - exp_tool_use = { - "toolUseId": "test-id", - "name": "test_tool", - "input": "", - "thoughtSignature": "dGVzdF9zaWduYXR1cmU=", - } - - assert tru_tool_use == exp_tool_use - assert "thoughtSignature" in tru_tool_use - assert tru_tool_use["thoughtSignature"] == "dGVzdF9zaWduYXR1cmU=" - - @pytest.mark.parametrize( ("event", "state", "exp_updated_state", "callback_args"), [ From fa9c74c9e1398f99b767c258cc338bdcf2577fa3 Mon Sep 17 00:00:00 2001 From: danielp Date: Mon, 24 Nov 2025 15:25:24 -0300 Subject: [PATCH 12/12] test: add coverage for thoughtSignature edge cases in gemini.py Added two new tests to cover previously missing code paths: 1. test_stream_request_with_invalid_base64_thought_signature() - Covers base64 decode error handling (lines 151-153) - Verifies graceful degradation when thoughtSignature contains invalid base64 data - Tests error logging without crashing the request 2. test_stream_response_tool_use_with_string_thought_signature() - Covers string-to-bytes conversion path (lines 308-309) - Tests edge case where thought_signature is returned as string instead of bytes from Gemini API - Verifies proper UTF-8 encoding and base64 conversion Coverage improvement: 3 previously uncovered code paths now tested. --- tests/strands/models/test_gemini.py | 101 ++++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) diff --git a/tests/strands/models/test_gemini.py b/tests/strands/models/test_gemini.py index 6dbd81929..b312ff532 100644 --- a/tests/strands/models/test_gemini.py +++ b/tests/strands/models/test_gemini.py @@ -831,3 +831,104 @@ async def test_stream_handles_non_json_error(gemini_client, model, messages, cap assert "Gemini API returned non-JSON error" in caplog.text assert f"error_message=<{error_message}>" in caplog.text + + +@pytest.mark.asyncio +async def test_stream_request_with_invalid_base64_thought_signature(gemini_client, model, model_id, caplog): + """Test that invalid base64 in thoughtSignature logs error but doesn't crash.""" + messages = [ + { + "role": "assistant", + "content": [ + { + "toolUse": { + "toolUseId": "c1", + "name": "calculator", + "input": {"expression": "2+2"}, + "thoughtSignature": "invalid-base64-data!!!", # Invalid base64 + }, + }, + ], + }, + ] + + with caplog.at_level(logging.ERROR): + await anext(model.stream(messages)) + + # Verify error was logged + assert "failed to decode thoughtSignature" in caplog.text + assert "toolUseId=" in caplog.text + + # Verify the request was still made (graceful degradation) + call_args = gemini_client.aio.models.generate_content_stream.call_args + assert call_args is not None + + # Verify content was formatted despite the error + contents = call_args.kwargs["contents"] + assert len(contents) == 1 + assert len(contents[0]["parts"]) == 1 + part = contents[0]["parts"][0] + assert "function_call" in part + assert part["function_call"]["name"] == "calculator" + # thought_signature should be None when decode fails + assert part.get("thought_signature") is None + + +@pytest.mark.asyncio +async def test_stream_response_tool_use_with_string_thought_signature( + gemini_client, model, messages, agenerator, alist +): + """Test that thoughtSignature as a string (not bytes) is properly converted and encoded.""" + # Create a mock Part with thought_signature as a string instead of bytes + mock_part = genai.types.Part( + function_call=genai.types.FunctionCall( + args={"expression": "3+3"}, + id="c2", + name="calculator", + ) + ) + # Manually set thought_signature as a string (edge case) + mock_part.thought_signature = "string_signature" # String instead of bytes + + gemini_client.aio.models.generate_content_stream.return_value = agenerator( + [ + genai.types.GenerateContentResponse( + candidates=[ + genai.types.Candidate( + content=genai.types.Content( + parts=[mock_part], + ), + finish_reason="STOP", + ), + ], + usage_metadata=genai.types.GenerateContentResponseUsageMetadata( + prompt_token_count=1, + total_token_count=3, + ), + ), + ] + ) + + tru_chunks = await alist(model.stream(messages)) + exp_chunks = [ + {"messageStart": {"role": "assistant"}}, + {"contentBlockStart": {"start": {}}}, + { + "contentBlockStart": { + "start": { + "toolUse": { + "name": "calculator", + "toolUseId": "calculator", + # String should be encoded to bytes, then base64 encoded + "thoughtSignature": "c3RyaW5nX3NpZ25hdHVyZQ==", # base64("string_signature") + } + } + } + }, + {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"expression": "3+3"}'}}}}, + {"contentBlockStop": {}}, + {"contentBlockStop": {}}, + {"messageStop": {"stopReason": "tool_use"}}, + {"metadata": {"usage": {"inputTokens": 1, "outputTokens": 2, "totalTokens": 3}, "metrics": {"latencyMs": 0}}}, + ] + assert tru_chunks == exp_chunks