diff --git a/src/strands/agent/agent.py b/src/strands/agent/agent.py index ebead3b7d..23e01e603 100644 --- a/src/strands/agent/agent.py +++ b/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() diff --git a/tests/strands/agent/test_agent.py b/tests/strands/agent/test_agent.py index 967a0dafb..03e7b992a 100644 --- a/tests/strands/agent/test_agent.py +++ b/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):