linting

Author: dbschmigelski

src/strands/agent/agent.py

@@ -967,7 +967,6 @@ def _filter_tool_parameters_for_recording(self, tool_name: str, input_params: di
         properties = tool_spec["inputSchema"]["json"]["properties"]
         return {k: v for k, v in input_params.items() if k in properties}
 
-        
     def _initialize_system_prompt(
         self, system_prompt: str | list[SystemContentBlock] | None
     ) -> tuple[str | None, list[SystemContentBlock] | None]:

src/strands/event_loop/event_loop.py

@@ -335,12 +335,12 @@ async def _handle_model_execution(
                 tool_specs = agent.tool_registry.get_all_tool_specs()
             try:
                 async for event in stream_messages(
-                    agent.model, 
+                    agent.model,
                     agent.system_prompt,
-                    agent.messages, 
+                    agent.messages,
                     tool_specs,
                     system_prompt_content=agent._system_prompt_content,
-                    tool_choice=structured_output_context.tool_choice
+                    tool_choice=structured_output_context.tool_choice,
                 ):
                     yield event
 

src/strands/event_loop/streaming.py

@@ -421,7 +421,7 @@ async def stream_messages(
     *,
     tool_choice: Optional[Any] = None,
     system_prompt_content: Optional[list[SystemContentBlock]] = None,
-    **kwargs: Any
+    **kwargs: Any,
 ) -> AsyncGenerator[TypedEvent, None]:
     """Streams messages to the model and processes the response.
 
@@ -431,19 +431,22 @@ async def stream_messages(
         messages: List of messages to send.
         tool_specs: The list of tool specs.
         tool_choice: Optional tool choice constraint for forcing specific tool usage.
-        system_prompt_content: The authoritative system prompt content blocks that always contains the system prompt data.
+        system_prompt_content: The authoritative system prompt content blocks that always contains the
+            system prompt data.
+        **kwargs: Additional keyword arguments for future extensibility.
+
     Yields:
         The reason for stopping, the final message, and the usage metrics
     """
     logger.debug("model=<%s> | streaming messages", model)
 
     messages = _normalize_messages(messages)
     start_time = time.time()
-    
+
     chunks = model.stream(
-        messages, 
-        tool_specs if tool_specs else None, 
-        system_prompt, 
+        messages,
+        tool_specs if tool_specs else None,
+        system_prompt,
         tool_choice=tool_choice,
         system_prompt_content=system_prompt_content,
     )

src/strands/models/bedrock.py

@@ -218,9 +218,7 @@ def _format_request(
         # Add cache point if configured (backwards compatibility)
         if self.config.get("cache_prompt"):
             warnings.warn(
-                "cache_prompt is deprecated. Use SystemContentBlock with cachePoint instead.",
-                UserWarning,
-                stacklevel=3
+                "cache_prompt is deprecated. Use SystemContentBlock with cachePoint instead.", UserWarning, stacklevel=3
             )
             system_blocks.append({"cachePoint": {"type": self.config["cache_prompt"]}})
 
@@ -613,6 +611,7 @@ async def stream(
             tool_specs: List of tool specifications to make available to the model.
             system_prompt: System prompt to provide context to the model.
             tool_choice: Selection strategy for tool invocation.
+            system_prompt_content: System prompt content blocks to provide context to the model.
             **kwargs: Additional keyword arguments for future extensibility.
 
         Yields:

tests/fixtures/mocked_model_provider.py

@@ -60,7 +60,7 @@ async def stream(
         tool_choice: Optional[Any] = None,
         *,
         system_prompt_content=None,
-        **kwargs: Any
+        **kwargs: Any,
     ) -> AsyncGenerator[Any, None]:
         events = self.map_agent_message_to_events(self.agent_responses[self.index])
         for event in events:

tests/strands/agent/test_agent.py

@@ -2164,11 +2164,13 @@ def shell(command: str):
 
     # And that it continued to the LLM call
     assert agent.messages[-1] == {"content": [{"text": "I invoked a tool!"}], "role": "assistant"}
+
+
 def test_agent_string_system_prompt():
     """Test initialization with string system prompt."""
     system_prompt = "You are a helpful assistant."
     agent = Agent(system_prompt=system_prompt)
-    
+
     assert agent.system_prompt == system_prompt
     assert agent._system_prompt_content == [{"text": system_prompt}]
 
@@ -2178,7 +2180,7 @@ def test_agent_single_text_block_system_prompt():
     text = "You are a helpful assistant."
     system_prompt_content = [{"text": text}]
     agent = Agent(system_prompt=system_prompt_content)
-    
+
     assert agent.system_prompt == text
     assert agent._system_prompt_content == system_prompt_content
 
@@ -2188,10 +2190,10 @@ def test_agent_multiple_blocks_system_prompt():
     system_prompt_content = [
         {"text": "You are a helpful assistant."},
         {"cachePoint": {"type": "default"}},
-        {"text": "Additional instructions."}
+        {"text": "Additional instructions."},
     ]
     agent = Agent(system_prompt=system_prompt_content)
-    
+
     assert agent.system_prompt == "You are a helpful assistant.\nAdditional instructions."
     assert agent._system_prompt_content == system_prompt_content
 
@@ -2200,23 +2202,23 @@ def test_agent_single_non_text_block_system_prompt():
     """Test initialization with single non-text SystemContentBlock."""
     system_prompt_content = [{"cachePoint": {"type": "default"}}]
     agent = Agent(system_prompt=system_prompt_content)
-    
+
     assert agent.system_prompt is None
     assert agent._system_prompt_content == system_prompt_content
 
 
 def test_agent_none_system_prompt():
     """Test initialization with None system prompt."""
     agent = Agent(system_prompt=None)
-    
+
     assert agent.system_prompt is None
     assert agent._system_prompt_content is None
 
 
 def test_agent_empty_list_system_prompt():
     """Test initialization with empty list system prompt."""
     agent = Agent(system_prompt=[])
-    
+
     assert agent.system_prompt is None
     assert agent._system_prompt_content == []
 
@@ -2225,7 +2227,7 @@ def test_agent_backwards_compatibility_string_access():
     """Test that string system prompts maintain backwards compatibility."""
     system_prompt = "You are a helpful assistant."
     agent = Agent(system_prompt=system_prompt)
-    
+
     # Should be able to access as string for backwards compatibility
     assert agent.system_prompt == system_prompt
 
@@ -2235,9 +2237,6 @@ def test_agent_backwards_compatibility_single_text_block():
     text = "You are a helpful assistant."
     system_prompt_content = [{"text": text}]
     agent = Agent(system_prompt=system_prompt_content)
-    
+
     # Should extract text for backwards compatibility
     assert agent.system_prompt == text
-
-
-

tests/strands/event_loop/test_streaming.py

@@ -861,10 +861,7 @@ async def test_stream_messages_with_system_prompt_content(agenerator, alist):
         ]
     )
 
-    system_prompt_content = [
-        {"text": "You are a helpful assistant."},
-        {"cachePoint": {"type": "default"}}
-    ]
+    system_prompt_content = [{"text": "You are a helpful assistant."}, {"cachePoint": {"type": "default"}}]
 
     stream = strands.event_loop.streaming.stream_messages(
         mock_model,

tests/strands/models/test_bedrock.py

@@ -409,11 +409,8 @@ def test_format_request_system_prompt(model, messages, model_id, system_prompt):
 
 def test_format_request_system_prompt_content(model, messages, model_id):
     """Test _format_request with SystemContentBlock input."""
-    system_prompt_content = [
-        {"text": "You are a helpful assistant."},
-        {"cachePoint": {"type": "default"}}
-    ]
-    
+    system_prompt_content = [{"text": "You are a helpful assistant."}, {"cachePoint": {"type": "default"}}]
+
     tru_request = model._format_request(messages, system_prompt_content=system_prompt_content)
     exp_request = {
         "inferenceConfig": {},
@@ -429,18 +426,15 @@ def test_format_request_system_prompt_content_with_cache_prompt_config(model, me
     """Test _format_request with SystemContentBlock and cache_prompt config (backwards compatibility)."""
     system_prompt_content = [{"text": "You are a helpful assistant."}]
     model.update_config(cache_prompt="default")
-    
+
     with pytest.warns(UserWarning, match="cache_prompt is deprecated"):
         tru_request = model._format_request(messages, system_prompt_content=system_prompt_content)
-    
+
     exp_request = {
         "inferenceConfig": {},
         "modelId": model_id,
         "messages": messages,
-        "system": [
-            {"text": "You are a helpful assistant."},
-            {"cachePoint": {"type": "default"}}
-        ],
+        "system": [{"text": "You are a helpful assistant."}, {"cachePoint": {"type": "default"}}],
     }
 
     assert tru_request == exp_request
@@ -528,10 +522,10 @@ def test_format_request_tool_choice_tool(model, messages, model_id, tool_spec):
 
 def test_format_request_cache(model, messages, model_id, tool_spec, cache_type):
     model.update_config(cache_prompt=cache_type, cache_tools=cache_type)
-    
+
     with pytest.warns(UserWarning, match="cache_prompt is deprecated"):
         tru_request = model._format_request(messages, tool_specs=[tool_spec])
-    
+
     exp_request = {
         "inferenceConfig": {},
         "modelId": model_id,
@@ -668,18 +662,15 @@ async def test_stream(bedrock_client, model, messages, tool_spec, model_id, addi
 async def test_stream_with_system_prompt_content(bedrock_client, model, messages, alist):
     """Test stream method with system_prompt_content parameter."""
     bedrock_client.converse_stream.return_value = {"stream": ["e1", "e2"]}
-    
-    system_prompt_content = [
-        {"text": "You are a helpful assistant."},
-        {"cachePoint": {"type": "default"}}
-    ]
+
+    system_prompt_content = [{"text": "You are a helpful assistant."}, {"cachePoint": {"type": "default"}}]
 
     response = model.stream(messages, system_prompt_content=system_prompt_content)
     tru_chunks = await alist(response)
     exp_chunks = ["e1", "e2"]
 
     assert tru_chunks == exp_chunks
-    
+
     # Verify the request was formatted with system_prompt_content
     expected_request = {
         "inferenceConfig": {},
@@ -694,12 +685,14 @@ async def test_stream_with_system_prompt_content(bedrock_client, model, messages
 async def test_stream_backwards_compatibility_single_text_block(bedrock_client, model, messages, alist):
     """Test that single text block in system_prompt_content works with legacy system_prompt."""
     bedrock_client.converse_stream.return_value = {"stream": ["e1", "e2"]}
-    
+
     system_prompt_content = [{"text": "You are a helpful assistant."}]
 
-    response = model.stream(messages, system_prompt="You are a helpful assistant.", system_prompt_content=system_prompt_content)
+    response = model.stream(
+        messages, system_prompt="You are a helpful assistant.", system_prompt_content=system_prompt_content
+    )
     await alist(response)
-    
+
     # Verify the request was formatted with system_prompt_content
     expected_request = {
         "inferenceConfig": {},

tests_integ/models/test_model_bedrock.py

@@ -274,12 +274,9 @@ def test_multi_prompt_system_content():
     system_prompt_content = [
         {"text": "You are a helpful assistant."},
         {"text": "Always be concise."},
-        {"text": "End responses with 'Done.'"}
+        {"text": "End responses with 'Done.'"},
     ]
-    
-    agent = Agent(
-        system_prompt=system_prompt_content,
-        load_tools_from_directory=False
-    )
-    result = agent("Hello!")
-    # just verifying there is no failure
+
+    agent = Agent(system_prompt=system_prompt_content, load_tools_from_directory=False)
+    # just verifying there is no failure
+    agent("Hello!")

tests_integ/models/test_model_openai.py

@@ -225,12 +225,9 @@ def test_rate_limit_throttling_integration_no_retries(model):
 
 def test_content_blocks_handling(model):
     """Test that content blocks are handled properly without failures."""
-    content = [
-        {"text": "What is 2+2?"},
-        {"text": "Please be brief."}
-    ]
-    
+    content = [{"text": "What is 2+2?"}, {"text": "Please be brief."}]
+
     agent = Agent(model=model, load_tools_from_directory=False)
     result = agent(content)
-    
+
     assert "4" in result.message["content"][0]["text"]