diff --git a/src/strands/agent/agent.py b/src/strands/agent/agent.py index e8ea3c9bc..5ad0dc3dd 100644 --- a/src/strands/agent/agent.py +++ b/src/strands/agent/agent.py @@ -428,6 +428,20 @@ def system_prompt(self, value: str | list[SystemContentBlock] | None) -> None: """ self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(value) + @property + def system_prompt_content(self) -> list[SystemContentBlock] | None: + """Get the system prompt as a list of content blocks. + + Returns the full content block representation of the system prompt, + preserving cache points and other non-text blocks. This is useful for + plugins and tools that need to manipulate the system prompt while + maintaining its structure. + + Returns: + The system prompt as a list of SystemContentBlock, or None if not set. + """ + return list(self._system_prompt_content) if self._system_prompt_content is not None else None + @property def tool(self) -> _ToolCaller: """Call tool as a function. diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py index 23217e81c..3329c166f 100644 --- a/src/strands/vended_plugins/skills/agent_skills.py +++ b/src/strands/vended_plugins/skills/agent_skills.py @@ -141,29 +141,57 @@ def _on_before_invocation(self, event: BeforeInvocationEvent) -> None: injected XML per-agent, so a single plugin instance can be shared across multiple agents safely. + Handles two system prompt formats: + - ``str | None``: String manipulation (append/replace) + - ``list[SystemContentBlock]``: Content block manipulation, preserving + cache points and other non-text blocks + Args: event: The before-invocation event containing the agent reference. """ agent = event.agent + skills_xml = self._generate_skills_xml() - current_prompt = agent.system_prompt or "" - - # Remove the previously injected XML block by exact match state_data = agent.state.get(self._state_key) last_injected_xml = state_data.get("last_injected_xml") if isinstance(state_data, dict) else None - if last_injected_xml is not None: - if last_injected_xml in current_prompt: - current_prompt = current_prompt.replace(last_injected_xml, "") - else: - logger.warning("unable to find previously injected skills XML in system prompt, re-appending") - skills_xml = self._generate_skills_xml() - injection = f"\n\n{skills_xml}" - new_prompt = f"{current_prompt}{injection}" if current_prompt else skills_xml - - new_injected_xml = injection if current_prompt else skills_xml - self._set_state_field(agent, "last_injected_xml", new_injected_xml) - agent.system_prompt = new_prompt + # Check if the system prompt uses content blocks with non-text elements (cache points etc.) + content_blocks = agent.system_prompt_content + # Design: We use the structured block path only when non-text blocks (e.g. cache points) are present. + # All-text block lists fall through to the string path for backward compatibility, since the + # string path handles them correctly and avoids unnecessary list manipulation. This means users + # who set a list of only text blocks will see them flattened, which is the existing behavior. + if content_blocks is not None and any("text" not in block for block in content_blocks): + # Content block path: filter out old skills block, append new one as a text block. + # This preserves cache points and other non-text blocks. + if last_injected_xml is not None: + filtered = [ + block for block in content_blocks if not ("text" in block and block["text"] == last_injected_xml) + ] + if len(filtered) == len(content_blocks): + logger.warning("unable to find previously injected skills XML in system prompt, re-appending") + else: + filtered = list(content_blocks) + + self._set_state_field(agent, "last_injected_xml", skills_xml) + filtered.append({"text": skills_xml}) + agent.system_prompt = filtered + else: + # String path: existing behavior for simple string prompts + current_prompt = agent.system_prompt or "" + + if last_injected_xml is not None: + if last_injected_xml in current_prompt: + current_prompt = current_prompt.replace(last_injected_xml, "") + else: + logger.warning("unable to find previously injected skills XML in system prompt, re-appending") + + injection = f"\n\n{skills_xml}" + new_prompt = f"{current_prompt}{injection}" if current_prompt else skills_xml + + new_injected_xml = injection if current_prompt else skills_xml + self._set_state_field(agent, "last_injected_xml", new_injected_xml) + agent.system_prompt = new_prompt def get_available_skills(self) -> list[Skill]: """Get the list of available skills. diff --git a/tests/strands/agent/test_agent.py b/tests/strands/agent/test_agent.py index 1e27274a1..0d1e68a4d 100644 --- a/tests/strands/agent/test_agent.py +++ b/tests/strands/agent/test_agent.py @@ -1166,6 +1166,30 @@ def test_system_prompt_setter_none(): assert agent._system_prompt_content is None +def test_system_prompt_content_property_string(): + """Test system_prompt_content property returns content blocks when set with string.""" + agent = Agent(system_prompt="hello world") + + assert agent.system_prompt_content == [{"text": "hello world"}] + + +def test_system_prompt_content_property_list(): + """Test system_prompt_content property returns content blocks when set with list.""" + blocks = [{"text": "Instructions"}, {"cachePoint": {"type": "default"}}] + agent = Agent(system_prompt=blocks) + + assert agent.system_prompt_content == blocks + # Verify the string getter still works + assert agent.system_prompt == "Instructions" + + +def test_system_prompt_content_property_none(): + """Test system_prompt_content property returns None when prompt is None.""" + agent = Agent(system_prompt=None) + + assert agent.system_prompt_content is None + + @pytest.mark.asyncio async def test_stream_async_passes_invocation_state(agent, mock_model, mock_event_loop_cycle, agenerator, alist): mock_model.mock_stream.side_effect = [ diff --git a/tests/strands/vended_plugins/skills/test_agent_skills.py b/tests/strands/vended_plugins/skills/test_agent_skills.py index db82355a9..59ee921b5 100644 --- a/tests/strands/vended_plugins/skills/test_agent_skills.py +++ b/tests/strands/vended_plugins/skills/test_agent_skills.py @@ -37,6 +37,9 @@ def _mock_agent(): lambda self: self._system_prompt, lambda self, value: _set_system_prompt(self, value), ) + type(agent).system_prompt_content = property( + lambda self: list(self._system_prompt_content) if self._system_prompt_content is not None else None, + ) agent.hooks = HookRegistry() agent.add_hook = MagicMock( @@ -59,14 +62,20 @@ def _mock_tool_context(agent: MagicMock) -> ToolContext: return ToolContext(tool_use=tool_use, agent=agent, invocation_state={"agent": agent}) -def _set_system_prompt(agent: MagicMock, value: str | None) -> None: - """Simulate the Agent.system_prompt setter.""" +def _set_system_prompt(agent, value) -> None: + """Simulate the Agent.system_prompt setter for str, list[SystemContentBlock], or None.""" if isinstance(value, str): agent._system_prompt = value agent._system_prompt_content = [{"text": value}] + elif isinstance(value, list): + text_parts = [block["text"] for block in value if "text" in block] + agent._system_prompt = "\n".join(text_parts) if text_parts else None + agent._system_prompt_content = value elif value is None: agent._system_prompt = None agent._system_prompt_content = None + else: + raise TypeError(f"system_prompt must be str, list[SystemContentBlock], or None, got {type(value).__name__}") class TestSkillsPluginInit: @@ -778,3 +787,189 @@ def test_skills_plugin_isinstance_check(self): plugin = AgentSkills(skills=[]) assert isinstance(plugin, Plugin) + + +class TestContentBlockSystemPrompt: + """Tests for system prompt injection with SystemContentBlock[] (content blocks with cache points).""" + + def test_preserves_cache_points(self): + """Test that injection preserves cache points in content block prompts.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + # Set system prompt as content blocks with a cache point + agent.system_prompt = [ + {"text": "You are an agent."}, + {"cachePoint": {"type": "default"}}, + {"text": "More instructions."}, + ] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # Content blocks should be preserved + blocks = agent.system_prompt_content + assert any("cachePoint" in block for block in blocks), "cache point was lost" + # Skills XML should be appended as a text block + skills_blocks = [block for block in blocks if "text" in block and "" in block["text"]] + assert len(skills_blocks) == 1 + # Original text blocks still present + assert any("text" in block and block["text"] == "You are an agent." for block in blocks) + assert any("text" in block and block["text"] == "More instructions." for block in blocks) + + def test_idempotent_with_cache_points(self): + """Test that repeated injections with content blocks don't accumulate.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + agent.system_prompt = [ + {"text": "You are an agent."}, + {"cachePoint": {"type": "default"}}, + ] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + first_blocks = list(agent.system_prompt_content) + + plugin._on_before_invocation(event) + second_blocks = list(agent.system_prompt_content) + + # Should have exactly the same number of blocks + assert len(first_blocks) == len(second_blocks) + # Exactly one skills XML block + skills_blocks = [b for b in second_blocks if "text" in b and "" in b["text"]] + assert len(skills_blocks) == 1 + + def test_cache_point_not_flattened_to_string(self): + """Test that content block prompts are NOT converted to a flat string.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + agent.system_prompt = [ + {"text": "Instructions"}, + {"cachePoint": {"type": "default"}}, + ] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # The result should still be a content block list, not a string + blocks = agent.system_prompt_content + assert isinstance(blocks, list) + assert len(blocks) >= 2 # At least cache point + skills text + assert any("cachePoint" in block for block in blocks) + + def test_content_blocks_with_skills_swap(self): + """Test that swapping skills updates the XML block in content block prompts.""" + skill_a = _make_skill(name="skill-a", description="A") + skill_b = _make_skill(name="skill-b", description="B") + plugin = AgentSkills(skills=[skill_a]) + agent = _mock_agent() + + agent.system_prompt = [ + {"text": "Base prompt."}, + {"cachePoint": {"type": "default"}}, + ] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # Verify skill-a is in the prompt + blocks = agent.system_prompt_content + skills_text = [b["text"] for b in blocks if "text" in b and "" in b["text"]] + assert len(skills_text) == 1 + assert "skill-a" in skills_text[0] + + # Swap to skill-b + plugin.set_available_skills([skill_b]) + plugin._on_before_invocation(event) + + blocks = agent.system_prompt_content + skills_text = [b["text"] for b in blocks if "text" in b and "" in b["text"]] + assert len(skills_text) == 1 + assert "skill-b" in skills_text[0] + assert "skill-a" not in skills_text[0] + # Cache point still there + assert any("cachePoint" in b for b in blocks) + + def test_string_prompt_still_works(self): + """Test that plain string prompts still use the string path (regression test).""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + agent.system_prompt = "Simple string prompt." + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # Should still work as a string + assert isinstance(agent.system_prompt, str) + assert "Simple string prompt." in agent.system_prompt + assert "" in agent.system_prompt + + def test_all_text_blocks_use_string_path(self): + """Test that content blocks with ONLY text blocks use the string path.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + # All-text content blocks — no cache points + agent.system_prompt = [ + {"text": "Part one."}, + {"text": "Part two."}, + ] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # String path should have been used (concatenated result) + prompt = agent.system_prompt + assert isinstance(prompt, str) + assert "" in prompt + + def test_warns_when_previous_xml_not_found_content_blocks(self, caplog): + """Test warning when previous XML is missing from content block prompt.""" + import logging + + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + agent.system_prompt = [ + {"text": "Original."}, + {"cachePoint": {"type": "default"}}, + ] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # Replace the prompt entirely (removes the skills block) + agent.system_prompt = [ + {"text": "New prompt."}, + {"cachePoint": {"type": "default"}}, + ] + + with caplog.at_level(logging.WARNING): + plugin._on_before_invocation(event) + + assert "unable to find previously injected skills XML in system prompt" in caplog.text + # Skills XML still appended + blocks = agent.system_prompt_content + assert any("text" in b and "" in b["text"] for b in blocks) + + def test_none_prompt_with_content_blocks_fallback(self): + """Test that None prompt works (falls through to string path).""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + agent.system_prompt = None + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + assert "" in agent.system_prompt