fix(agent): restore temporary system_prompt override for invocations

src/strands/agent/agent.py

@@ -83,6 +83,10 @@ class _DefaultCallbackHandlerSentinel:
     pass
 
 
+# Sentinel to distinguish "not provided" from explicit None for system_prompt override
+_UNSET: Any = object()
+
+
 class _DefaultRetryStrategySentinel:
     """Sentinel class to distinguish between explicit None and default parameter value for retry_strategy."""
 
@@ -385,6 +389,7 @@ def __call__(
         self,
         prompt: AgentInput = None,
         *,
+        system_prompt: str | list[SystemContentBlock] | None = _UNSET,
         invocation_state: dict[str, Any] | None = None,
         structured_output_model: type[BaseModel] | None = None,
         structured_output_prompt: str | None = None,
@@ -404,6 +409,8 @@ def __call__(
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
+            system_prompt: Temporary system prompt override for this invocation only.
+                The agent's original system prompt is restored after the call completes.
             invocation_state: Additional parameters to pass through the event loop.
             structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
             structured_output_prompt: Custom prompt for forcing structured output (overrides agent default).
@@ -421,6 +428,7 @@ def __call__(
         return run_async(
             lambda: self.invoke_async(
                 prompt,
+                system_prompt=system_prompt,
                 invocation_state=invocation_state,
                 structured_output_model=structured_output_model,
                 structured_output_prompt=structured_output_prompt,
@@ -432,6 +440,7 @@ async def invoke_async(
         self,
         prompt: AgentInput = None,
         *,
+        system_prompt: str | list[SystemContentBlock] | None = _UNSET,
         invocation_state: dict[str, Any] | None = None,
         structured_output_model: type[BaseModel] | None = None,
         structured_output_prompt: str | None = None,
@@ -451,6 +460,8 @@ async def invoke_async(
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
+            system_prompt: Temporary system prompt override for this invocation only.
+                The agent's original system prompt is restored after the call completes.
             invocation_state: Additional parameters to pass through the event loop.
             structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
             structured_output_prompt: Custom prompt for forcing structured output (overrides agent default).
@@ -466,6 +477,7 @@ async def invoke_async(
         """
         events = self.stream_async(
             prompt,
+            system_prompt=system_prompt,
             invocation_state=invocation_state,
             structured_output_model=structured_output_model,
             structured_output_prompt=structured_output_prompt,
@@ -655,6 +667,7 @@ async def stream_async(
         self,
         prompt: AgentInput = None,
         *,
+        system_prompt: str | list[SystemContentBlock] | None = _UNSET,
         invocation_state: dict[str, Any] | None = None,
         structured_output_model: type[BaseModel] | None = None,
         structured_output_prompt: str | None = None,
@@ -674,6 +687,8 @@ async def stream_async(
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
+            system_prompt: Temporary system prompt override for this invocation only.
+                The agent's original system prompt is restored after the call completes.
             invocation_state: Additional parameters to pass through the event loop.
             structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
             structured_output_prompt: Custom prompt for forcing structured output (overrides agent default).
@@ -710,6 +725,13 @@ async def stream_async(
                 )
 
         try:
+            # Apply temporary system prompt override if provided
+            original_system_prompt: str | list[SystemContentBlock] | None = _UNSET
+            if system_prompt is not _UNSET:
+                original_system_prompt = self._system_prompt
+                original_system_prompt_content = self._system_prompt_content
+                self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(system_prompt)
+
             self._interrupt_state.resume(prompt)
 
             self.event_loop_metrics.reset_usage_metrics()
@@ -756,6 +778,11 @@ async def stream_async(
                     raise
 
         finally:
+            # Restore original system prompt if it was temporarily overridden
+            if original_system_prompt is not _UNSET:
+                self._system_prompt = original_system_prompt
+                self._system_prompt_content = original_system_prompt_content
+
             if self._invocation_lock.locked():
                 self._invocation_lock.release()
 

tests/strands/agent/test_agent.py

@@ -434,7 +434,6 @@ def test_agent__call__passes_invocation_state(mock_model, agent, tool, mock_even
     async def check_invocation_state(**kwargs):
         invocation_state = kwargs["invocation_state"]
         assert invocation_state["some_value"] == "a_value"
-        assert invocation_state["system_prompt"] == override_system_prompt
         assert invocation_state["model"] == override_model
         assert invocation_state["event_loop_metrics"] == override_event_loop_metrics
         assert invocation_state["callback_handler"] == override_callback_handler
@@ -443,6 +442,9 @@ async def check_invocation_state(**kwargs):
         assert invocation_state["tool_config"] == override_tool_config
         assert invocation_state["agent"] == agent
 
+        # system_prompt is now a proper parameter (temporary override), not part of invocation_state
+        assert kwargs["agent"].system_prompt == override_system_prompt
+
         # Return expected values from event_loop_cycle
         yield EventLoopStopEvent("stop", {"role": "assistant", "content": [{"text": "Response"}]}, {}, {})
 
@@ -461,6 +463,29 @@ async def check_invocation_state(**kwargs):
     )
 
     mock_event_loop_cycle.assert_called_once()
+    # Verify original system prompt is restored after invocation
+    assert agent.system_prompt != override_system_prompt
+
+
+def test_agent__call__temporary_system_prompt_override(mock_model, agent, mock_event_loop_cycle, agenerator):
+    """Verify system_prompt parameter provides temporary override restored after call.
+
+    See: https://github.com/strands-agents/sdk-python/issues/344
+    """
+    original_prompt = agent.system_prompt
+    temp_prompt = "Temporary instructions for this call only"
+
+    async def capture_system_prompt(**kwargs):
+        # During the call, the agent's system_prompt should be the override
+        assert kwargs["agent"].system_prompt == temp_prompt
+        yield EventLoopStopEvent("stop", {"role": "assistant", "content": [{"text": "ok"}]}, {}, {})
+
+    mock_event_loop_cycle.side_effect = capture_system_prompt
+
+    agent("test", system_prompt=temp_prompt)
+
+    # After the call, the original system prompt must be restored
+    assert agent.system_prompt == original_prompt
 
 
 def test_agent__call__retry_with_reduced_context(mock_model, agent, tool, agenerator):