From 6dfa6cbfb39a5d4ceb49bb945d58a93d26a30260 Mon Sep 17 00:00:00 2001
From: agent-of-mkmeral <217235299+strands-agent@users.noreply.github.com>
Date: Tue, 14 Apr 2026 22:21:07 +0000
Subject: [PATCH 1/5] feat(skills): support SystemContentBlock[] in system
 prompt injection

The AgentSkills plugin's _on_before_invocation hook previously only handled
string system prompts. When the system prompt was set as a list of
SystemContentBlock (e.g. with cache points), the plugin would:

1. Read agent.system_prompt (which returns a concatenated string)
2. Do string manipulation to append/replace skills XML
3. Set agent.system_prompt = new_string, destroying the content block structure

This meant cache points and other non-text blocks were silently lost on
every invocation.

Changes:
- Add system_prompt_content property to Agent (public getter for
  _system_prompt_content) so plugins can read the content block
  representation
- Update _on_before_invocation with two codepaths:
  - Content block path: when non-text blocks (cache points) are present,
    filter out old skills text block and append new one, preserving
    the list structure
  - String path: existing behavior for simple string prompts

This matches the TypeScript SDK's implementation which already handles
SystemContentBlock[] via a two-branch design in _injectSkillsXml.

Testing:
- 3 new agent property tests (string, list, None)
- 8 new skills plugin tests covering cache point preservation,
  idempotency, skills swap, string regression, all-text-blocks,
  warning on missing XML, and None fallback
- 133/133 skills tests passing, 0 regressions
---
 src/strands/agent/agent.py                    |  14 ++
 .../vended_plugins/skills/agent_skills.py     |  54 +++--
 tests/strands/agent/test_agent.py             |  24 +++
 .../skills/test_agent_skills.py               | 197 +++++++++++++++++-
 4 files changed, 272 insertions(+), 17 deletions(-)

diff --git a/src/strands/agent/agent.py b/src/strands/agent/agent.py
index 37fa5fc00..51e8c87d0 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 self._system_prompt_content
+
     @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 5e42b9230..cd1ab8be0 100644
--- a/src/strands/vended_plugins/skills/agent_skills.py
+++ b/src/strands/vended_plugins/skills/agent_skills.py
@@ -140,29 +140,53 @@ 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 = getattr(agent, "system_prompt_content", None)
+        has_structured_blocks = content_blocks is not None and any("text" not in block for block in content_blocks)
+
+        if has_structured_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 0057c50a3..73077ea32 100644
--- a/tests/strands/agent/test_agent.py
+++ b/tests/strands/agent/test_agent.py
@@ -1164,6 +1164,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 52802a6c1..e6f83c44d 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: self._system_prompt_content,
+    )
 
     agent.hooks = HookRegistry()
     agent.add_hook = MagicMock(
@@ -59,11 +62,15 @@ 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
@@ -689,3 +696,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

From 988b195f629a568bd64a4726f9686a94c742a4c4 Mon Sep 17 00:00:00 2001
From: agent-of-mkmeral <agent-of-mkmeral@users.noreply.github.com>
Date: Wed, 15 Apr 2026 17:15:33 +0000
Subject: [PATCH 2/5] fix: revert unrelated changes, fix lint E501
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Removed all unrelated changes that were accidentally included:
- Reverted MessageMetadata/get_message_metadata removal from content.py
- Reverted Message.metadata field removal
- Reverted event_loop metadata attachment removal
- Reverted streaming message sanitization removal
- Reverted Message.__required_keys__ change
- Reverted skill.py from_url removal
- Restored deleted test files

Kept only the targeted fix:
- system_prompt_content property on Agent (new public API)
- Content block support in AgentSkills._on_before_invocation
- Fixed E501 lint error (line 165 list comprehension)
- 3 new agent tests + 8 new skills content block tests

All checks: lint ✅ griffe ✅ 122 agent tests ✅ 66 skills tests ✅
---
 src/strands/agent/agent.py                    |   2 +-
 .../_recover_message_on_max_tokens_reached.py |   5 +-
 src/strands/event_loop/event_loop.py          |   7 +
 src/strands/event_loop/streaming.py           |   3 +
 src/strands/telemetry/tracer.py               |   4 +-
 src/strands/types/content.py                  |  30 +++-
 .../vended_plugins/skills/agent_skills.py     |   6 +-
 src/strands/vended_plugins/skills/skill.py    |  53 ++++++-
 .../strands/agent/hooks/test_agent_events.py  |  13 +-
 tests/strands/agent/test_agent.py             |  16 +-
 .../strands/agent/test_agent_cancellation.py  |   3 +-
 tests/strands/agent/test_agent_hooks.py       |   6 +-
 tests/strands/event_loop/test_event_loop.py   |  17 ++-
 .../event_loop/test_event_loop_metadata.py    | 141 ++++++++++++++++++
 ...t_recover_message_on_max_tokens_reached.py |  28 ++++
 .../tools/mcp/test_mcp_client_tasks.py        |   8 +-
 tests/strands/types/test_message_metadata.py  |  37 +++++
 .../vended_plugins/skills/test_skill.py       |  90 ++++++++++-
 18 files changed, 432 insertions(+), 37 deletions(-)
 create mode 100644 tests/strands/event_loop/test_event_loop_metadata.py
 create mode 100644 tests/strands/types/test_message_metadata.py

diff --git a/src/strands/agent/agent.py b/src/strands/agent/agent.py
index 51e8c87d0..6e29f7f5e 100644
--- a/src/strands/agent/agent.py
+++ b/src/strands/agent/agent.py
@@ -1039,7 +1039,7 @@ async def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
                 # Check if all item in input list are dictionaries
                 elif all(isinstance(item, dict) for item in prompt):
                     # Check if all items are messages
-                    if all(all(key in item for key in Message.__annotations__.keys()) for item in prompt):
+                    if all(all(key in item for key in Message.__required_keys__) for item in prompt):
                         # Messages input - add all messages to conversation
                         messages = cast(Messages, prompt)
 
diff --git a/src/strands/event_loop/_recover_message_on_max_tokens_reached.py b/src/strands/event_loop/_recover_message_on_max_tokens_reached.py
index ab6fb4abe..dc073ba07 100644
--- a/src/strands/event_loop/_recover_message_on_max_tokens_reached.py
+++ b/src/strands/event_loop/_recover_message_on_max_tokens_reached.py
@@ -68,4 +68,7 @@ def recover_message_on_max_tokens_reached(message: Message) -> Message:
             }
         )
 
-    return {"content": valid_content, "role": message["role"]}
+    recovered: Message = {"content": valid_content, "role": message["role"]}
+    if "metadata" in message:
+        recovered["metadata"] = message["metadata"]
+    return recovered
diff --git a/src/strands/event_loop/event_loop.py b/src/strands/event_loop/event_loop.py
index b4af16058..bf1cc7a84 100644
--- a/src/strands/event_loop/event_loop.py
+++ b/src/strands/event_loop/event_loop.py
@@ -354,6 +354,13 @@ async def _handle_model_execution(
                 stop_reason, message, usage, metrics = event["stop"]
                 invocation_state.setdefault("request_state", {})
 
+                # Attach metadata to the assistant message immediately so it's
+                # available to all downstream consumers (hooks, events, state).
+                message["metadata"] = {
+                    "usage": usage,
+                    "metrics": metrics,
+                }
+
                 after_model_call_event = AfterModelCallEvent(
                     agent=agent,
                     invocation_state=invocation_state,
diff --git a/src/strands/event_loop/streaming.py b/src/strands/event_loop/streaming.py
index 0a1161135..76eda48bf 100644
--- a/src/strands/event_loop/streaming.py
+++ b/src/strands/event_loop/streaming.py
@@ -488,6 +488,9 @@ async def stream_messages(
     logger.debug("model=<%s> | streaming messages", model)
 
     messages = _normalize_messages(messages)
+    # Whitelist only role and content before sending to the model provider.
+    # This ensures metadata (and any future non-model fields) never leak to providers.
+    messages = [Message(role=msg["role"], content=msg["content"]) for msg in messages]
     start_time = time.time()
 
     chunks = model.stream(
diff --git a/src/strands/telemetry/tracer.py b/src/strands/telemetry/tracer.py
index d5d399f95..37c16d3ae 100644
--- a/src/strands/telemetry/tracer.py
+++ b/src/strands/telemetry/tracer.py
@@ -527,9 +527,7 @@ def start_event_loop_cycle_span(
         event_loop_cycle_id = str(invocation_state.get("event_loop_cycle_id"))
         parent_span = parent_span if parent_span else invocation_state.get("event_loop_parent_span")
 
-        attributes: dict[str, AttributeValue] = self._get_common_attributes(
-            operation_name="execute_event_loop_cycle"
-        )
+        attributes: dict[str, AttributeValue] = self._get_common_attributes(operation_name="execute_event_loop_cycle")
         attributes["event_loop.cycle_id"] = event_loop_cycle_id
 
         if custom_trace_attributes:
diff --git a/src/strands/types/content.py b/src/strands/types/content.py
index 2b0714bee..8db1d1d98 100644
--- a/src/strands/types/content.py
+++ b/src/strands/types/content.py
@@ -6,11 +6,12 @@
 - Bedrock docs: https://docs.aws.amazon.com/bedrock/latest/APIReference/API_Types_Amazon_Bedrock_Runtime.html
 """
 
-from typing import Literal
+from typing import Any, Literal
 
 from typing_extensions import NotRequired, TypedDict
 
 from .citations import CitationsContentBlock
+from .event_loop import Metrics, Usage
 from .media import DocumentContent, ImageContent, VideoContent
 from .tools import ToolResult, ToolUse
 
@@ -177,17 +178,44 @@ class ContentBlockStop(TypedDict):
 """
 
 
+class MessageMetadata(TypedDict, total=False):
+    """Optional metadata attached to a message.
+
+    Not sent to model providers — explicitly stripped before model calls.
+    Persisted alongside the message in session storage.
+
+    Attributes:
+        usage: Token usage information from the model response.
+        metrics: Performance metrics from the model response.
+        custom: Arbitrary user/framework metadata (e.g. compression provenance).
+    """
+
+    usage: Usage
+    metrics: Metrics
+    custom: dict[str, Any]
+
+
 class Message(TypedDict):
     """A message in a conversation with the agent.
 
     Attributes:
         content: The message content.
         role: The role of the message sender.
+        metadata: Optional metadata, stripped before model calls.
     """
 
     content: list[ContentBlock]
     role: Role
+    metadata: NotRequired[MessageMetadata]
 
 
 Messages = list[Message]
 """A list of messages representing a conversation."""
+
+
+def get_message_metadata(message: Message) -> MessageMetadata:
+    """Get metadata for a message, returning empty dict if not present.
+
+    Individual fields (usage, metrics, custom) may not be present. Use .get() to safely access them.
+    """
+    return message.get("metadata", {})
diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py
index cd1ab8be0..0fdce0713 100644
--- a/src/strands/vended_plugins/skills/agent_skills.py
+++ b/src/strands/vended_plugins/skills/agent_skills.py
@@ -162,7 +162,11 @@ def _on_before_invocation(self, event: BeforeInvocationEvent) -> None:
             # 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)]
+                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:
diff --git a/src/strands/vended_plugins/skills/skill.py b/src/strands/vended_plugins/skills/skill.py
index 3e1b6bba5..a60c1cd6c 100644
--- a/src/strands/vended_plugins/skills/skill.py
+++ b/src/strands/vended_plugins/skills/skill.py
@@ -1,15 +1,17 @@
 """Skill data model and loading utilities for AgentSkills.io skills.
 
 This module defines the Skill dataclass and provides classmethods for
-discovering, parsing, and loading skills from the filesystem or raw content.
-Skills are directories containing a SKILL.md file with YAML frontmatter
-metadata and markdown instructions.
+discovering, parsing, and loading skills from the filesystem, raw content,
+or HTTPS URLs. Skills are directories containing a SKILL.md file with YAML
+frontmatter metadata and markdown instructions.
 """
 
 from __future__ import annotations
 
 import logging
 import re
+import urllib.error
+import urllib.request
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -222,6 +224,9 @@ class Skill:
         # Load all skills from a parent directory
         skills = Skill.from_directory("./skills/")
 
+        # From an HTTPS URL
+        skill = Skill.from_url("https://example.com/SKILL.md")
+
     Attributes:
         name: Unique identifier for the skill (1-64 chars, lowercase alphanumeric + hyphens).
         description: Human-readable description of what the skill does.
@@ -333,6 +338,48 @@ def from_content(cls, content: str, *, strict: bool = False) -> Skill:
 
         return _build_skill_from_frontmatter(frontmatter, body)
 
+    @classmethod
+    def from_url(cls, url: str, *, strict: bool = False) -> Skill:
+        """Load a skill by fetching its SKILL.md content from an HTTPS URL.
+
+        Fetches the raw SKILL.md content over HTTPS and parses it using
+        :meth:`from_content`.  The URL must point directly to the raw
+        file content (not an HTML page).
+
+        Example::
+
+            skill = Skill.from_url(
+                "https://raw.githubusercontent.com/org/repo/main/SKILL.md"
+            )
+
+        Args:
+            url: An ``https://`` URL pointing directly to raw SKILL.md content.
+            strict: If True, raise on any validation issue. If False (default),
+                warn and load anyway.
+
+        Returns:
+            A Skill instance populated from the fetched SKILL.md content.
+
+        Raises:
+            ValueError: If ``url`` is not an ``https://`` URL.
+            RuntimeError: If the SKILL.md content cannot be fetched.
+        """
+        if not url.startswith("https://"):
+            raise ValueError(f"url=<{url}> | not a valid HTTPS URL")
+
+        logger.info("url=<%s> | fetching skill content", url)
+
+        try:
+            req = urllib.request.Request(url, headers={"User-Agent": "strands-agents-sdk"})  # noqa: S310
+            with urllib.request.urlopen(req, timeout=30) as response:  # noqa: S310
+                content: str = response.read().decode("utf-8")
+        except urllib.error.HTTPError as e:
+            raise RuntimeError(f"url=<{url}> | HTTP {e.code}: {e.reason}") from e
+        except urllib.error.URLError as e:
+            raise RuntimeError(f"url=<{url}> | failed to fetch skill: {e.reason}") from e
+
+        return cls.from_content(content, strict=strict)
+
     @classmethod
     def from_directory(cls, skills_dir: str | Path, *, strict: bool = False) -> list[Skill]:
         """Load all skills from a parent directory containing skill subdirectories.
diff --git a/tests/strands/agent/hooks/test_agent_events.py b/tests/strands/agent/hooks/test_agent_events.py
index 02c367ccc..1f09579b0 100644
--- a/tests/strands/agent/hooks/test_agent_events.py
+++ b/tests/strands/agent/hooks/test_agent_events.py
@@ -147,6 +147,7 @@ async def test_stream_e2e_success(alist):
                     {"toolUse": {"input": {}, "name": "normal_tool", "toolUseId": "123"}},
                 ],
                 "role": "assistant",
+                "metadata": ANY,
             }
         },
         {
@@ -205,6 +206,7 @@ async def test_stream_e2e_success(alist):
                     {"toolUse": {"input": {}, "name": "async_tool", "toolUseId": "1234"}},
                 ],
                 "role": "assistant",
+                "metadata": ANY,
             }
         },
         {
@@ -263,6 +265,7 @@ async def test_stream_e2e_success(alist):
                     {"toolUse": {"input": {}, "name": "streaming_tool", "toolUseId": "12345"}},
                 ],
                 "role": "assistant",
+                "metadata": ANY,
             }
         },
         {
@@ -307,11 +310,11 @@ async def test_stream_e2e_success(alist):
         },
         {"event": {"contentBlockStop": {}}},
         {"event": {"messageStop": {"stopReason": "end_turn"}}},
-        {"message": {"content": [{"text": "I invoked the tools!"}], "role": "assistant"}},
+        {"message": {"content": [{"text": "I invoked the tools!"}], "role": "assistant", "metadata": ANY}},
         {
             "result": AgentResult(
                 stop_reason="end_turn",
-                message={"content": [{"text": "I invoked the tools!"}], "role": "assistant"},
+                message={"content": [{"text": "I invoked the tools!"}], "role": "assistant", "metadata": ANY},
                 metrics=ANY,
                 state={},
             ),
@@ -371,11 +374,11 @@ async def test_stream_e2e_throttle_and_redact(alist, mock_sleep):
         },
         {"event": {"contentBlockStop": {}}},
         {"event": {"messageStop": {"stopReason": "guardrail_intervened"}}},
-        {"message": {"content": [{"text": "INPUT BLOCKED!"}], "role": "assistant"}},
+        {"message": {"content": [{"text": "INPUT BLOCKED!"}], "role": "assistant", "metadata": ANY}},
         {
             "result": AgentResult(
                 stop_reason="guardrail_intervened",
-                message={"content": [{"text": "INPUT BLOCKED!"}], "role": "assistant"},
+                message={"content": [{"text": "INPUT BLOCKED!"}], "role": "assistant", "metadata": ANY},
                 metrics=ANY,
                 state={},
             ),
@@ -442,6 +445,7 @@ async def test_stream_e2e_reasoning_redacted_content(alist):
                     {"text": "Response with redacted reasoning"},
                 ],
                 "role": "assistant",
+                "metadata": ANY,
             }
         },
         {
@@ -453,6 +457,7 @@ async def test_stream_e2e_reasoning_redacted_content(alist):
                         {"text": "Response with redacted reasoning"},
                     ],
                     "role": "assistant",
+                    "metadata": ANY,
                 },
                 metrics=ANY,
                 state={},
diff --git a/tests/strands/agent/test_agent.py b/tests/strands/agent/test_agent.py
index 73077ea32..0d1e68a4d 100644
--- a/tests/strands/agent/test_agent.py
+++ b/tests/strands/agent/test_agent.py
@@ -336,7 +336,7 @@ def test_agent__call__(
         "stop_reason": result.stop_reason,
     }
     exp_result = {
-        "message": {"content": [{"text": "test text"}], "role": "assistant"},
+        "message": {"content": [{"text": "test text"}], "role": "assistant", "metadata": unittest.mock.ANY},
         "state": {},
         "stop_reason": "end_turn",
     }
@@ -781,6 +781,7 @@ def test_agent__call__callback(mock_model, agent, callback_handler, agenerator):
                     {"reasoningContent": {"reasoningText": {"text": "value", "signature": "value"}}},
                     {"text": "value"},
                 ],
+                "metadata": unittest.mock.ANY,
             },
         ),
         unittest.mock.call(
@@ -793,6 +794,7 @@ def test_agent__call__callback(mock_model, agent, callback_handler, agenerator):
                         {"reasoningContent": {"reasoningText": {"text": "value", "signature": "value"}}},
                         {"text": "value"},
                     ],
+                    "metadata": unittest.mock.ANY,
                 },
                 metrics=unittest.mock.ANY,
                 state={},
@@ -817,7 +819,7 @@ async def test_agent__call__in_async_context(mock_model, agent, agenerator):
     result = agent("test")
 
     tru_message = result.message
-    exp_message = {"content": [{"text": "abc"}], "role": "assistant"}
+    exp_message = {"content": [{"text": "abc"}], "role": "assistant", "metadata": unittest.mock.ANY}
     assert tru_message == exp_message
 
 
@@ -837,7 +839,7 @@ async def test_agent_invoke_async(mock_model, agent, agenerator):
     result = await agent.invoke_async("test")
 
     tru_message = result.message
-    exp_message = {"content": [{"text": "abc"}], "role": "assistant"}
+    exp_message = {"content": [{"text": "abc"}], "role": "assistant", "metadata": unittest.mock.ANY}
     assert tru_message == exp_message
 
 
@@ -1128,7 +1130,7 @@ async def test_stream_async_multi_modal_input(mock_model, agent, agenerator, ali
     tru_message = agent.messages
     exp_message = [
         {"content": prompt, "role": "user"},
-        {"content": [{"text": "I see text and an image"}], "role": "assistant"},
+        {"content": [{"text": "I see text and an image"}], "role": "assistant", "metadata": unittest.mock.ANY},
     ]
     assert tru_message == exp_message
 
@@ -1990,7 +1992,11 @@ def shell(command: str):
     }
 
     # And that it continued to the LLM call
-    assert agent.messages[-1] == {"content": [{"text": "I invoked a tool!"}], "role": "assistant"}
+    assert agent.messages[-1] == {
+        "content": [{"text": "I invoked a tool!"}],
+        "role": "assistant",
+        "metadata": unittest.mock.ANY,
+    }
 
 
 def test_agent_string_system_prompt():
diff --git a/tests/strands/agent/test_agent_cancellation.py b/tests/strands/agent/test_agent_cancellation.py
index 6af153f4a..756e96485 100644
--- a/tests/strands/agent/test_agent_cancellation.py
+++ b/tests/strands/agent/test_agent_cancellation.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import threading
+from unittest.mock import ANY
 
 import pytest
 
@@ -31,7 +32,7 @@ async def test_agent_cancel_before_invocation():
     result = await agent.invoke_async("Hello")
 
     assert result.stop_reason == "cancelled"
-    assert result.message == {"role": "assistant", "content": [{"text": "Cancelled by user"}]}
+    assert result.message == {"role": "assistant", "content": [{"text": "Cancelled by user"}], "metadata": ANY}
 
 
 @pytest.mark.asyncio
diff --git a/tests/strands/agent/test_agent_hooks.py b/tests/strands/agent/test_agent_hooks.py
index 3a40d69a8..2c61ee966 100644
--- a/tests/strands/agent/test_agent_hooks.py
+++ b/tests/strands/agent/test_agent_hooks.py
@@ -173,6 +173,7 @@ def test_agent__call__hooks(agent, hook_provider, agent_tool, mock_model, tool_u
             message={
                 "content": [{"toolUse": tool_use}],
                 "role": "assistant",
+                "metadata": ANY,
             },
             stop_reason="tool_use",
         ),
@@ -199,7 +200,7 @@ def test_agent__call__hooks(agent, hook_provider, agent_tool, mock_model, tool_u
         agent=agent,
         invocation_state=ANY,
         stop_response=AfterModelCallEvent.ModelStopResponse(
-            message=mock_model.agent_responses[1],
+            message={"role": "assistant", "content": [{"text": "I invoked a tool!"}], "metadata": ANY},
             stop_reason="end_turn",
         ),
         exception=None,
@@ -246,6 +247,7 @@ async def test_agent_stream_async_hooks(agent, hook_provider, agent_tool, mock_m
             message={
                 "content": [{"toolUse": tool_use}],
                 "role": "assistant",
+                "metadata": ANY,
             },
             stop_reason="tool_use",
         ),
@@ -272,7 +274,7 @@ async def test_agent_stream_async_hooks(agent, hook_provider, agent_tool, mock_m
         agent=agent,
         invocation_state=ANY,
         stop_response=AfterModelCallEvent.ModelStopResponse(
-            message=mock_model.agent_responses[1],
+            message={"role": "assistant", "content": [{"text": "I invoked a tool!"}], "metadata": ANY},
             stop_reason="end_turn",
         ),
         exception=None,
diff --git a/tests/strands/event_loop/test_event_loop.py b/tests/strands/event_loop/test_event_loop.py
index f91f7c2af..871371f5f 100644
--- a/tests/strands/event_loop/test_event_loop.py
+++ b/tests/strands/event_loop/test_event_loop.py
@@ -193,7 +193,7 @@ async def test_event_loop_cycle_text_response(
     tru_stop_reason, tru_message, _, tru_request_state, _, _ = events[-1]["stop"]
 
     exp_stop_reason = "end_turn"
-    exp_message = {"role": "assistant", "content": [{"text": "test text"}]}
+    exp_message = {"role": "assistant", "content": [{"text": "test text"}], "metadata": ANY}
     exp_request_state = {}
 
     assert tru_stop_reason == exp_stop_reason and tru_message == exp_message and tru_request_state == exp_request_state
@@ -225,7 +225,7 @@ async def test_event_loop_cycle_text_response_throttling(
     tru_stop_reason, tru_message, _, tru_request_state, _, _ = events[-1]["stop"]
 
     exp_stop_reason = "end_turn"
-    exp_message = {"role": "assistant", "content": [{"text": "test text"}]}
+    exp_message = {"role": "assistant", "content": [{"text": "test text"}], "metadata": ANY}
     exp_request_state = {}
 
     assert tru_stop_reason == exp_stop_reason and tru_message == exp_message and tru_request_state == exp_request_state
@@ -264,7 +264,7 @@ async def test_event_loop_cycle_exponential_backoff(
 
     # Verify the final response
     assert tru_stop_reason == "end_turn"
-    assert tru_message == {"role": "assistant", "content": [{"text": "test text"}]}
+    assert tru_message == {"role": "assistant", "content": [{"text": "test text"}], "metadata": ANY}
     assert tru_request_state == {}
 
     # Verify that sleep was called with increasing delays
@@ -354,7 +354,7 @@ async def test_event_loop_cycle_tool_result(
     tru_stop_reason, tru_message, _, tru_request_state, _, _ = events[-1]["stop"]
 
     exp_stop_reason = "end_turn"
-    exp_message = {"role": "assistant", "content": [{"text": "test text"}]}
+    exp_message = {"role": "assistant", "content": [{"text": "test text"}], "metadata": ANY}
     exp_request_state = {}
 
     assert tru_stop_reason == exp_stop_reason and tru_message == exp_message and tru_request_state == exp_request_state
@@ -389,7 +389,6 @@ async def test_event_loop_cycle_tool_result(
                     },
                 ],
             },
-            {"role": "assistant", "content": [{"text": "test text"}]},
         ],
         tool_registry.get_all_tool_specs(),
         "p1",
@@ -484,6 +483,7 @@ async def test_event_loop_cycle_stop(
                 }
             }
         ],
+        "metadata": ANY,
     }
     exp_request_state = {"stop_event_loop": True}
 
@@ -946,14 +946,14 @@ async def test_event_loop_cycle_exception_model_hooks(mock_sleep, agent, model,
         agent=agent,
         invocation_state=ANY,
         stop_response=AfterModelCallEvent.ModelStopResponse(
-            message={"content": [{"text": "test text"}], "role": "assistant"}, stop_reason="end_turn"
+            message={"content": [{"text": "test text"}], "role": "assistant", "metadata": ANY}, stop_reason="end_turn"
         ),
         exception=None,
     )
 
     # Final message
     assert next(events) == MessageAddedEvent(
-        agent=agent, message={"content": [{"text": "test text"}], "role": "assistant"}
+        agent=agent, message={"content": [{"text": "test text"}], "role": "assistant", "metadata": ANY}
     )
 
 
@@ -997,6 +997,7 @@ def interrupt_callback(event):
                     },
                 ],
                 "role": "assistant",
+                "metadata": ANY,
             },
         },
         "interrupts": {
@@ -1131,7 +1132,7 @@ async def test_invalid_tool_names_adds_tool_uses(agent, model, alist):
     # ensure that we got end_turn and not tool_use
     assert events[-1] == EventLoopStopEvent(
         stop_reason="end_turn",
-        message={"content": [{"text": "I invoked a tool!"}], "role": "assistant"},
+        message={"content": [{"text": "I invoked a tool!"}], "role": "assistant", "metadata": ANY},
         metrics=ANY,
         request_state={},
     )
diff --git a/tests/strands/event_loop/test_event_loop_metadata.py b/tests/strands/event_loop/test_event_loop_metadata.py
new file mode 100644
index 000000000..e6fe97f39
--- /dev/null
+++ b/tests/strands/event_loop/test_event_loop_metadata.py
@@ -0,0 +1,141 @@
+"""Tests for metadata population on assistant messages in the event loop."""
+
+import threading
+import unittest.mock
+
+import pytest
+
+import strands
+import strands.event_loop.event_loop
+from strands import Agent
+from strands.event_loop._retry import ModelRetryStrategy
+from strands.hooks import HookRegistry
+from strands.interrupt import _InterruptState
+from strands.telemetry.metrics import EventLoopMetrics
+from strands.tools.executors import SequentialToolExecutor
+from strands.tools.registry import ToolRegistry
+
+
+@pytest.fixture
+def model():
+    return unittest.mock.Mock()
+
+
+@pytest.fixture
+def messages():
+    return [{"role": "user", "content": [{"text": "Hello"}]}]
+
+
+@pytest.fixture
+def hook_registry():
+    registry = HookRegistry()
+    retry_strategy = ModelRetryStrategy()
+    retry_strategy.register_hooks(registry)
+    return registry
+
+
+@pytest.fixture
+def tool_registry():
+    return ToolRegistry()
+
+
+@pytest.fixture
+def agent(model, messages, tool_registry, hook_registry):
+    mock = unittest.mock.Mock(name="agent")
+    mock.__class__ = Agent
+    mock.config.cache_points = []
+    mock.model = model
+    mock.system_prompt = "test"
+    mock.messages = messages
+    mock.tool_registry = tool_registry
+    mock.thread_pool = None
+    mock.event_loop_metrics = EventLoopMetrics()
+    mock.event_loop_metrics.reset_usage_metrics()
+    mock.hooks = hook_registry
+    mock.tool_executor = SequentialToolExecutor()
+    mock._interrupt_state = _InterruptState()
+    mock._cancel_signal = threading.Event()
+    mock.trace_attributes = {}
+    mock.retry_strategy = ModelRetryStrategy()
+    return mock
+
+
+@pytest.mark.asyncio
+async def test_metadata_populated_on_assistant_message(agent, model, agenerator, alist):
+    """After a model response, the assistant message should have metadata with usage and metrics."""
+    model.stream.return_value = agenerator(
+        [
+            {"contentBlockDelta": {"delta": {"text": "response"}}},
+            {"contentBlockStop": {}},
+            {
+                "metadata": {
+                    "usage": {"inputTokens": 42, "outputTokens": 10, "totalTokens": 52},
+                    "metrics": {"latencyMs": 200},
+                }
+            },
+        ]
+    )
+
+    stream = strands.event_loop.event_loop.event_loop_cycle(agent=agent, invocation_state={})
+    await alist(stream)
+
+    # The assistant message should be in agent.messages
+    assistant_msg = agent.messages[-1]
+    assert assistant_msg["role"] == "assistant"
+    assert "metadata" in assistant_msg
+
+    meta = assistant_msg["metadata"]
+    assert meta["usage"]["inputTokens"] == 42
+    assert meta["usage"]["outputTokens"] == 10
+    assert meta["usage"]["totalTokens"] == 52
+    assert meta["metrics"]["latencyMs"] == 200
+
+
+@pytest.mark.asyncio
+async def test_metadata_has_default_usage_when_no_metadata_event(agent, model, agenerator, alist):
+    """When no metadata event is in the stream, metadata should still be set with defaults."""
+    model.stream.return_value = agenerator(
+        [
+            {"contentBlockDelta": {"delta": {"text": "response"}}},
+            {"contentBlockStop": {}},
+        ]
+    )
+
+    stream = strands.event_loop.event_loop.event_loop_cycle(agent=agent, invocation_state={})
+    await alist(stream)
+
+    assistant_msg = agent.messages[-1]
+    assert "metadata" in assistant_msg
+    assert assistant_msg["metadata"]["usage"]["inputTokens"] == 0
+    assert assistant_msg["metadata"]["usage"]["outputTokens"] == 0
+    assert assistant_msg["metadata"]["metrics"]["latencyMs"] == 0
+
+
+@pytest.mark.asyncio
+async def test_metadata_stripped_before_model_call(agent, model, agenerator, alist):
+    """Metadata from previous messages should be stripped before sending to the model."""
+    # Pre-populate a message with metadata (simulating a previous turn)
+    agent.messages.append(
+        {
+            "role": "assistant",
+            "content": [{"text": "previous response"}],
+            "metadata": {"usage": {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15}},
+        }
+    )
+    agent.messages.append({"role": "user", "content": [{"text": "follow up"}]})
+
+    model.stream.return_value = agenerator(
+        [
+            {"contentBlockDelta": {"delta": {"text": "response"}}},
+            {"contentBlockStop": {}},
+        ]
+    )
+
+    stream = strands.event_loop.event_loop.event_loop_cycle(agent=agent, invocation_state={})
+    await alist(stream)
+
+    # Verify that messages passed to model.stream() have no metadata key
+    call_args = model.stream.call_args
+    messages_sent = call_args[0][0]
+    for msg in messages_sent:
+        assert "metadata" not in msg, f"metadata leaked to model: {msg}"
diff --git a/tests/strands/event_loop/test_recover_message_on_max_tokens_reached.py b/tests/strands/event_loop/test_recover_message_on_max_tokens_reached.py
index 402e90966..6dff0fc29 100644
--- a/tests/strands/event_loop/test_recover_message_on_max_tokens_reached.py
+++ b/tests/strands/event_loop/test_recover_message_on_max_tokens_reached.py
@@ -224,6 +224,34 @@ def test_recover_message_on_max_tokens_reached_multiple_incomplete_tools():
     assert "incomplete due to maximum token limits" in result["content"][2]["text"]
 
 
+def test_recover_message_on_max_tokens_reached_preserves_metadata():
+    """Test that metadata is preserved through recovery."""
+    message: Message = {
+        "role": "assistant",
+        "content": [
+            {"toolUse": {"name": "calculator", "input": {}, "toolUseId": "123"}},
+        ],
+        "metadata": {"usage": {"inputTokens": 42, "outputTokens": 10, "totalTokens": 52}},
+    }
+
+    result = recover_message_on_max_tokens_reached(message)
+
+    assert "metadata" in result
+    assert result["metadata"]["usage"]["inputTokens"] == 42
+
+
+def test_recover_message_on_max_tokens_reached_without_metadata():
+    """Test that recovery works fine when no metadata is present."""
+    message: Message = {
+        "role": "assistant",
+        "content": [{"text": "some text"}],
+    }
+
+    result = recover_message_on_max_tokens_reached(message)
+
+    assert "metadata" not in result
+
+
 def test_recover_message_on_max_tokens_reached_preserves_user_role():
     """Test that the function preserves the original message role."""
     incomplete_message: Message = {
diff --git a/tests/strands/tools/mcp/test_mcp_client_tasks.py b/tests/strands/tools/mcp/test_mcp_client_tasks.py
index c21db9e28..d566ac6f5 100644
--- a/tests/strands/tools/mcp/test_mcp_client_tasks.py
+++ b/tests/strands/tools/mcp/test_mcp_client_tasks.py
@@ -251,9 +251,7 @@ def test_call_tool_sync_forwards_meta_to_task(self, mock_transport, mock_session
 
         with MCPClient(mock_transport["transport_callable"], tasks_config=TasksConfig()) as client:
             client.list_tools_sync()
-            client.call_tool_sync(
-                tool_use_id="test-id", name="meta_tool", arguments={"param": "value"}, meta=meta
-            )
+            client.call_tool_sync(tool_use_id="test-id", name="meta_tool", arguments={"param": "value"}, meta=meta)
 
             experimental.call_tool_as_task.assert_called_once()
             call_kwargs = experimental.call_tool_as_task.call_args
@@ -281,9 +279,7 @@ def test_call_tool_sync_forwards_none_meta_to_task(self, mock_transport, mock_se
 
         with MCPClient(mock_transport["transport_callable"], tasks_config=TasksConfig()) as client:
             client.list_tools_sync()
-            client.call_tool_sync(
-                tool_use_id="test-id", name="no_meta_tool", arguments={"param": "value"}
-            )
+            client.call_tool_sync(tool_use_id="test-id", name="no_meta_tool", arguments={"param": "value"})
 
             experimental.call_tool_as_task.assert_called_once()
             call_kwargs = experimental.call_tool_as_task.call_args
diff --git a/tests/strands/types/test_message_metadata.py b/tests/strands/types/test_message_metadata.py
new file mode 100644
index 000000000..a7f93f690
--- /dev/null
+++ b/tests/strands/types/test_message_metadata.py
@@ -0,0 +1,37 @@
+"""Tests for MessageMetadata and get_message_metadata."""
+
+from strands.types.content import Message, MessageMetadata, get_message_metadata
+
+
+def test_message_without_metadata():
+    msg: Message = {"role": "assistant", "content": [{"text": "hello"}]}
+    assert get_message_metadata(msg) == {}
+
+
+def test_message_with_metadata():
+    meta: MessageMetadata = {
+        "usage": {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15},
+        "metrics": {"latencyMs": 100},
+    }
+    msg: Message = {"role": "assistant", "content": [{"text": "hello"}], "metadata": meta}
+    assert get_message_metadata(msg) == meta
+    assert get_message_metadata(msg)["usage"]["inputTokens"] == 10
+
+
+def test_message_with_custom_metadata():
+    meta: MessageMetadata = {
+        "custom": {"source": "summarization", "original_turns": [5, 6, 7]},
+    }
+    msg: Message = {"role": "assistant", "content": [{"text": "summary"}], "metadata": meta}
+    result = get_message_metadata(msg)
+    assert result["custom"]["source"] == "summarization"
+
+
+def test_metadata_does_not_affect_role_and_content():
+    msg: Message = {
+        "role": "assistant",
+        "content": [{"text": "hello"}],
+        "metadata": {"usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2}},
+    }
+    assert msg["role"] == "assistant"
+    assert msg["content"] == [{"text": "hello"}]
diff --git a/tests/strands/vended_plugins/skills/test_skill.py b/tests/strands/vended_plugins/skills/test_skill.py
index 53d6f3507..cb67d71a2 100644
--- a/tests/strands/vended_plugins/skills/test_skill.py
+++ b/tests/strands/vended_plugins/skills/test_skill.py
@@ -551,11 +551,99 @@ def test_strict_mode(self):
             Skill.from_content(content, strict=True)
 
 
+class TestSkillFromUrl:
+    """Tests for Skill.from_url."""
+
+    _SKILL_MODULE = "strands.vended_plugins.skills.skill"
+    _SAMPLE_CONTENT = "---\nname: my-skill\ndescription: A remote skill\n---\nRemote instructions.\n"
+
+    def _mock_urlopen(self, content):
+        """Create a mock urlopen context manager returning the given content."""
+        from unittest.mock import MagicMock
+
+        mock_response = MagicMock()
+        mock_response.read.return_value = content.encode("utf-8")
+        mock_response.__enter__ = MagicMock(return_value=mock_response)
+        mock_response.__exit__ = MagicMock(return_value=False)
+        return mock_response
+
+    def test_from_url_returns_skill(self):
+        """Test loading a skill from a URL returns a single Skill."""
+        from unittest.mock import patch
+
+        mock_response = self._mock_urlopen(self._SAMPLE_CONTENT)
+        with patch(f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=mock_response):
+            skill = Skill.from_url("https://raw.githubusercontent.com/org/repo/main/SKILL.md")
+
+        assert isinstance(skill, Skill)
+        assert skill.name == "my-skill"
+        assert skill.description == "A remote skill"
+        assert "Remote instructions." in skill.instructions
+        assert skill.path is None
+
+    def test_from_url_invalid_url_raises(self):
+        """Test that a non-HTTPS URL raises ValueError."""
+        with pytest.raises(ValueError, match="not a valid HTTPS URL"):
+            Skill.from_url("./local-path")
+
+    def test_from_url_http_rejected(self):
+        """Test that http:// URLs are rejected."""
+        with pytest.raises(ValueError, match="not a valid HTTPS URL"):
+            Skill.from_url("http://example.com/SKILL.md")
+
+    def test_from_url_http_error_raises(self):
+        """Test that HTTP errors propagate as RuntimeError."""
+        import urllib.error
+        from unittest.mock import patch
+
+        with patch(
+            f"{self._SKILL_MODULE}.urllib.request.urlopen",
+            side_effect=urllib.error.HTTPError(
+                url="https://example.com", code=404, msg="Not Found", hdrs=None, fp=None
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="HTTP 404"):
+                Skill.from_url("https://example.com/SKILL.md")
+
+    def test_from_url_network_error_raises(self):
+        """Test that network errors propagate as RuntimeError."""
+        import urllib.error
+        from unittest.mock import patch
+
+        with patch(
+            f"{self._SKILL_MODULE}.urllib.request.urlopen",
+            side_effect=urllib.error.URLError("Connection refused"),
+        ):
+            with pytest.raises(RuntimeError, match="failed to fetch"):
+                Skill.from_url("https://example.com/SKILL.md")
+
+    def test_from_url_strict_mode(self):
+        """Test that strict mode is forwarded to from_content."""
+        from unittest.mock import patch
+
+        bad_content = "---\nname: BAD_NAME\ndescription: Bad\n---\nBody."
+
+        with patch(f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=self._mock_urlopen(bad_content)):
+            with pytest.raises(ValueError):
+                Skill.from_url("https://example.com/SKILL.md", strict=True)
+
+    def test_from_url_invalid_content_raises(self):
+        """Test that non-SKILL.md content (e.g. HTML page) raises ValueError."""
+        from unittest.mock import patch
+
+        html_content = "<html><body>Not a SKILL.md</body></html>"
+
+        with patch(f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=self._mock_urlopen(html_content)):
+            with pytest.raises(ValueError, match="frontmatter"):
+                Skill.from_url("https://example.com/SKILL.md")
+
+
 class TestSkillClassmethods:
     """Tests for Skill classmethod existence."""
 
     def test_skill_classmethods_exist(self):
-        """Test that Skill has from_file, from_content, and from_directory classmethods."""
+        """Test that Skill has from_file, from_content, from_directory, and from_url classmethods."""
         assert callable(getattr(Skill, "from_file", None))
         assert callable(getattr(Skill, "from_content", None))
         assert callable(getattr(Skill, "from_directory", None))
+        assert callable(getattr(Skill, "from_url", None))

From dbcb8d51d12d62eec4fe44af77a4776dcb5d2c69 Mon Sep 17 00:00:00 2001
From: agent-of-mkmeral <agent-of-mkmeral@users.noreply.github.com>
Date: Wed, 15 Apr 2026 17:37:42 +0000
Subject: [PATCH 3/5] =?UTF-8?q?fix:=20address=20review=20comments=20?=
 =?UTF-8?q?=E2=80=94=20shallow=20copy,=20direct=20access,=20design=20docs,?=
 =?UTF-8?q?=20type=20validation?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses 4 review thread items:
1. Return shallow copy from system_prompt_content to prevent mutation (agent.py)
2. Replace getattr with direct agent.system_prompt_content access (agent_skills.py)
3. Add design comment explaining all-text fallback heuristic (agent_skills.py)
4. Add TypeError guard in test mock for invalid types (test_agent_skills.py)
---
 src/strands/agent/agent.py                               | 2 +-
 src/strands/vended_plugins/skills/agent_skills.py        | 6 +++++-
 tests/strands/vended_plugins/skills/test_agent_skills.py | 2 ++
 3 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/src/strands/agent/agent.py b/src/strands/agent/agent.py
index 6e29f7f5e..5ad0dc3dd 100644
--- a/src/strands/agent/agent.py
+++ b/src/strands/agent/agent.py
@@ -440,7 +440,7 @@ def system_prompt_content(self) -> list[SystemContentBlock] | None:
         Returns:
             The system prompt as a list of SystemContentBlock, or None if not set.
         """
-        return self._system_prompt_content
+        return list(self._system_prompt_content) if self._system_prompt_content is not None else None
 
     @property
     def tool(self) -> _ToolCaller:
diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py
index 0fdce0713..fa5cc9975 100644
--- a/src/strands/vended_plugins/skills/agent_skills.py
+++ b/src/strands/vended_plugins/skills/agent_skills.py
@@ -155,7 +155,11 @@ def _on_before_invocation(self, event: BeforeInvocationEvent) -> None:
         last_injected_xml = state_data.get("last_injected_xml") if isinstance(state_data, dict) else None
 
         # Check if the system prompt uses content blocks with non-text elements (cache points etc.)
-        content_blocks = getattr(agent, "system_prompt_content", None)
+        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.
         has_structured_blocks = content_blocks is not None and any("text" not in block for block in content_blocks)
 
         if has_structured_blocks:
diff --git a/tests/strands/vended_plugins/skills/test_agent_skills.py b/tests/strands/vended_plugins/skills/test_agent_skills.py
index e6f83c44d..a64ec15dd 100644
--- a/tests/strands/vended_plugins/skills/test_agent_skills.py
+++ b/tests/strands/vended_plugins/skills/test_agent_skills.py
@@ -74,6 +74,8 @@ def _set_system_prompt(agent, value) -> None:
     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:

From 0b530adec1c5afba6cf1466b55fc59e35e7642bf Mon Sep 17 00:00:00 2001
From: agent-of-mkmeral <217235299+strands-agent@users.noreply.github.com>
Date: Wed, 15 Apr 2026 18:04:21 +0000
Subject: [PATCH 4/5] fix: add explicit None guard for mypy type narrowing

mypy cannot narrow content_blocks type through boolean variable
has_structured_blocks. Adding redundant 'content_blocks is not None'
to the if-condition allows mypy to narrow the union type properly.

Verified: hatch run prepare passes (format, ruff, mypy, tests).
---
 src/strands/vended_plugins/skills/agent_skills.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py
index fa5cc9975..a048a5808 100644
--- a/src/strands/vended_plugins/skills/agent_skills.py
+++ b/src/strands/vended_plugins/skills/agent_skills.py
@@ -162,14 +162,12 @@ def _on_before_invocation(self, event: BeforeInvocationEvent) -> None:
         # who set a list of only text blocks will see them flattened, which is the existing behavior.
         has_structured_blocks = content_blocks is not None and any("text" not in block for block in content_blocks)
 
-        if has_structured_blocks:
+        if has_structured_blocks and content_blocks is not None:
             # 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)
+                    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")

From e29bb2b8006e14f17cc77b35fb73229125ccaa94 Mon Sep 17 00:00:00 2001
From: agent-of-mkmeral <217235299+strands-agent@users.noreply.github.com>
Date: Wed, 15 Apr 2026 18:50:22 +0000
Subject: [PATCH 5/5] =?UTF-8?q?fix:=20address=20review=20comments=20?=
 =?UTF-8?q?=E2=80=94=20inline=20check,=20fix=20mock=20shallow=20copy?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Removed intermediate has_structured_blocks variable and redundant
  content_blocks is not None guard. Inlined the check directly into
  the if condition for clarity (review thread #5).
- Fixed test mock system_prompt_content property to return a shallow
  copy matching the real Agent implementation, so mutation bugs are
  caught in tests (review thread #11).
- Merged from upstream/main to pick up latest changes.
---
 src/strands/vended_plugins/skills/agent_skills.py        | 4 +---
 tests/strands/vended_plugins/skills/test_agent_skills.py | 2 +-
 2 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py
index 214a21d9a..3329c166f 100644
--- a/src/strands/vended_plugins/skills/agent_skills.py
+++ b/src/strands/vended_plugins/skills/agent_skills.py
@@ -161,9 +161,7 @@ def _on_before_invocation(self, event: BeforeInvocationEvent) -> None:
         # 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.
-        has_structured_blocks = content_blocks is not None and any("text" not in block for block in content_blocks)
-
-        if has_structured_blocks and content_blocks is not None:
+        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:
diff --git a/tests/strands/vended_plugins/skills/test_agent_skills.py b/tests/strands/vended_plugins/skills/test_agent_skills.py
index 764dbaaf6..59ee921b5 100644
--- a/tests/strands/vended_plugins/skills/test_agent_skills.py
+++ b/tests/strands/vended_plugins/skills/test_agent_skills.py
@@ -38,7 +38,7 @@ def _mock_agent():
         lambda self, value: _set_system_prompt(self, value),
     )
     type(agent).system_prompt_content = property(
-        lambda self: self._system_prompt_content,
+        lambda self: list(self._system_prompt_content) if self._system_prompt_content is not None else None,
     )
 
     agent.hooks = HookRegistry()