Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions src/strands/agent/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
58 changes: 43 additions & 15 deletions src/strands/vended_plugins/skills/agent_skills.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Issue: The event loop still accesses the private agent._system_prompt_content directly (lines 324 and 346 of event_loop.py), while the skills plugin now uses the public agent.system_prompt_content. This creates an inconsistency: internal SDK code bypasses the shallow copy protection of the public property and accesses the raw internal list.

Suggestion: Since this PR introduces the public property, consider migrating the event_loop references in a follow-up. This ensures a single access pattern and means the shallow copy behavior (if desired) is consistent across all consumers.

# 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.
Expand Down
24 changes: 24 additions & 0 deletions tests/strands/agent/test_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down
199 changes: 197 additions & 2 deletions tests/strands/vended_plugins/skills/test_agent_skills.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Issue: The _set_system_prompt mock helper simulates the setter but joins text parts with "\n".join(text_parts). The real _initialize_system_prompt also uses "\n".join(text_parts), so this is consistent. Good faithfulness to the actual implementation.

One gap: the mock doesn't reject invalid types (e.g., int), while the real implementation would fall through to return None, None. Consider adding an else: raise TypeError(...) to catch test bugs, or at minimum assert that only expected types flow through.

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:
Expand Down Expand Up @@ -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 "<available_skills>" 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 "<available_skills>" 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 "<available_skills>" 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 "<available_skills>" 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 "<available_skills>" 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 "<available_skills>" 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 "<available_skills>" 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 "<available_skills>" in agent.system_prompt