fix(models/gemini): handle non-JSON error messages from Gemini API

src/strands/models/gemini.py

@@ -408,7 +408,13 @@ async def stream(
             if not error.message:
                 raise
 
-            message = json.loads(error.message)
+            try:
+                message = json.loads(error.message) if error.message else {}
+            except json.JSONDecodeError as e:
+                logger.warning("error_message=<%s> | Gemini API returned non-JSON error", error.message)
+                # Re-raise the original ClientError (not JSONDecodeError) and make the JSON error the explicit cause
+                raise error from e
+
             match message["error"]["status"]:
                 case "RESOURCE_EXHAUSTED" | "UNAVAILABLE":
                     raise ModelThrottledException(error.message) from error

tests/strands/models/test_gemini.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import unittest.mock
 
 import pydantic
@@ -621,3 +622,18 @@ async def test_structured_output(gemini_client, model, messages, model_id, weath
         "model": model_id,
     }
     gemini_client.aio.models.generate_content.assert_called_with(**exp_request)
+
+
+@pytest.mark.asyncio
+async def test_stream_handles_non_json_error(gemini_client, model, messages, caplog, alist):
+    error_message = "Invalid API key"
+    gemini_client.aio.models.generate_content_stream.side_effect = genai.errors.ClientError(
+        error_message, {"message": error_message}
+    )
+
+    with caplog.at_level(logging.WARNING):
+        with pytest.raises(genai.errors.ClientError, match=error_message):
+            await alist(model.stream(messages))
+
+    assert "Gemini API returned non-JSON error" in caplog.text
+    assert f"error_message=<{error_message}>" in caplog.text