Python: Preserve structured instructions when merging chat options

python/packages/anthropic/README.md

@@ -40,3 +40,7 @@ system_blocks: list[BetaTextBlockParam] = [
 
 response = await client.get_response("Hello", options={"instructions": system_blocks})
 ```
+
+Instructions contributed later in a run — by a context provider such as `SkillsProvider`, or by per-run
+`options` — are appended as an additional text block after the configured blocks. The blocks you supply keep
+their structure and their position, so a `cache_control` breakpoint stays valid.

python/packages/anthropic/agent_framework_anthropic/_chat_client.py

@@ -669,10 +669,25 @@ def _extract_structured_instructions(
         messages: Sequence[Message],
         instructions: Any,
     ) -> Sequence[BetaTextBlockParam] | Sequence[Mapping[str, Any]]:
+        """Normalize structured instructions into Anthropic system blocks.
+
+        Plain strings are wrapped as text blocks, which is how instructions contributed later in a run
+        (by a context provider or per-run options) arrive alongside caller-supplied blocks.
+
+        Raises:
+            ValueError: If a leading system message is present, since that is a second, ambiguous
+                source of system content.
+        """
         if messages and isinstance(messages[0], Message) and messages[0].role == "system":
             raise ValueError("structured Anthropic instructions cannot be combined with a leading system message.")
 
-        return cast(Sequence[BetaTextBlockParam] | Sequence[Mapping[str, Any]], instructions)
+        raw_blocks: Sequence[Any] = (
+            [instructions] if isinstance(instructions, Mapping) else cast(Sequence[Any], instructions)
+        )
+        blocks: list[Any] = [
+            {"type": "text", "text": block} if isinstance(block, str) else block for block in raw_blocks
+        ]
+        return cast(Sequence[BetaTextBlockParam] | Sequence[Mapping[str, Any]], blocks)
 
     def _prepare_text_instructions_for_anthropic(self, messages: Sequence[Message], instructions: Any) -> str:
         if isinstance(instructions, str):

python/packages/anthropic/tests/test_anthropic_client.py

@@ -14,7 +14,10 @@
     ChatResponseUpdate,
     Content,
     FunctionInvocationLayer,
+    InlineSkill,
     Message,
+    SkillFrontmatter,
+    SkillsProvider,
     SupportsChatGetResponse,
     tool,
 )
@@ -30,7 +33,7 @@
 )
 from pydantic import BaseModel, Field
 
-from agent_framework_anthropic import AnthropicClient, RawAnthropicClient
+from agent_framework_anthropic import AnthropicChatOptions, AnthropicClient, RawAnthropicClient
 from agent_framework_anthropic._chat_client import AnthropicSettings
 from agent_framework_anthropic._feature_usage import FeatureIndex
 
@@ -1077,6 +1080,109 @@ async def test_prepare_options_structured_system_blocks_reject_conflicts(
         client._prepare_options(messages, options)
 
 
+async def test_prepare_options_wraps_appended_text_instructions_as_system_blocks(
+    mock_anthropic_client: MagicMock,
+) -> None:
+    """Text appended to structured blocks should become an additional text block."""
+    client = create_test_anthropic_client(mock_anthropic_client)
+    messages = [Message(role="user", contents=["Hello"])]
+    cached_block = {
+        "type": "text",
+        "text": "Stable instructions",
+        "cache_control": {"type": "ephemeral", "ttl": "1h"},
+    }
+
+    run_options = client._prepare_options(messages, {"instructions": [cached_block, "Appended instructions"]})
+
+    assert run_options["system"] == [cached_block, {"type": "text", "text": "Appended instructions"}]
+
+
+async def test_prepare_options_wraps_a_single_structured_mapping_as_system_blocks(
+    mock_anthropic_client: MagicMock,
+) -> None:
+    """A lone system block mapping should be normalized into a one-element block list."""
+    client = create_test_anthropic_client(mock_anthropic_client)
+    messages = [Message(role="user", contents=["Hello"])]
+    block = {"type": "text", "text": "Stable instructions"}
+
+    run_options = client._prepare_options(messages, {"instructions": block})
+
+    assert run_options["system"] == [block]
+
+
+@pytest.mark.parametrize("with_skills", [False, True], ids=["without_skills_provider", "with_skills_provider"])
+async def test_agent_run_preserves_structured_system_blocks(with_skills: bool) -> None:
+    """Regression test for #7700.
+
+    Structured system blocks must survive the public ``Agent.run()`` path whether or not a context
+    provider contributes instructions. Contributed instructions are appended as an extra system block
+    instead of collapsing the blocks into a string, which would disable Anthropic prompt caching.
+    """
+    requests: list[dict[str, Any]] = []
+
+    async def create(**kwargs: Any) -> BetaMessage:
+        requests.append(kwargs)
+        return BetaMessage(
+            id="msg_test",
+            content=[BetaTextBlock(type="text", text="ok")],
+            model="claude-3-5-sonnet-20241022",
+            role="assistant",
+            stop_reason="end_turn",
+            type="message",
+            usage=BetaUsage(input_tokens=1, output_tokens=1),
+        )
+
+    transport = MagicMock()
+    transport.base_url = "https://example.invalid"
+    transport.beta.messages.create = create
+
+    system_blocks = [
+        {
+            "type": "text",
+            "text": "Stable instructions that should be cached.",
+            "cache_control": {"type": "ephemeral", "ttl": "1h"},
+        },
+        {"type": "text", "text": "Dynamic request context that should not be cached."},
+    ]
+    context_providers = []
+    if with_skills:
+        skill = InlineSkill(
+            frontmatter=SkillFrontmatter(name="example-skill", description="A generic standalone example skill."),
+            instructions="Use this generic skill when asked for an example.",
+        )
+        context_providers.append(
+            SkillsProvider(
+                [skill],
+                disable_load_skill_approval=True,
+                disable_read_skill_resource_approval=True,
+            )
+        )
+
+    agent = Agent(
+        client=AnthropicClient(anthropic_client=transport, model="claude-3-5-sonnet-20241022"),
+        default_options=cast(
+            AnthropicChatOptions,
+            {"model": "claude-3-5-sonnet-20241022", "max_tokens": 64, "instructions": system_blocks},
+        ),
+        context_providers=context_providers,
+    )
+
+    async with agent:
+        await agent.run("Hello")
+
+    system = requests[0]["system"]
+    # The cached prefix must stay byte-identical so the cache breakpoint keeps matching.
+    assert system[: len(system_blocks)] == system_blocks
+
+    if not with_skills:
+        assert system == system_blocks
+        return
+
+    assert len(system) == len(system_blocks) + 1
+    assert system[-1]["type"] == "text"
+    assert "example-skill" in system[-1]["text"]
+
+
 async def test_prepare_options_splits_assistant_embedded_tool_results(
     mock_anthropic_client: MagicMock,
 ) -> None:

python/packages/core/agent_framework/_agents.py

@@ -56,6 +56,7 @@
     ChatResponseUpdate,
     Message,
     ResponseStream,
+    _append_instructions,  # pyright: ignore[reportPrivateUsage]
     _build_agent_response_from_chat_response,  # pyright: ignore[reportPrivateUsage]
     map_chat_to_agent_update,
     normalize_messages,
@@ -159,8 +160,8 @@ def _merge_options(base: dict[str, Any], override: dict[str, Any]) -> dict[str,
             # Merge metadata dicts
             result["metadata"] = {**result["metadata"], **value}
         elif key == "instructions" and result.get("instructions"):
-            # Concatenate instructions
-            result["instructions"] = f"{result['instructions']}\n{value}"
+            # Concatenate instructions, preserving provider-native structured values
+            result["instructions"] = _append_instructions(result["instructions"], value)
         else:
             result[key] = value
     return {key: value for key, value in result.items() if value is not None}
@@ -1623,10 +1624,7 @@ async def _prepare_session_and_messages(
         # Merge provider-contributed instructions into chat_options
         if session_context.instructions:
             combined_instructions = "\n".join(session_context.instructions)
-            if "instructions" in chat_options:
-                chat_options["instructions"] = f"{chat_options['instructions']}\n{combined_instructions}"
-            else:
-                chat_options["instructions"] = combined_instructions
+            chat_options["instructions"] = _append_instructions(chat_options.get("instructions"), combined_instructions)
 
         return session_context, chat_options
 

python/packages/core/agent_framework/_types.py

@@ -3917,6 +3917,38 @@ def validate_tool_mode(
     return tool_choice
 
 
+def _append_instructions(
+    base: str | Mapping[str, Any] | Sequence[Any] | None,
+    addition: str | Mapping[str, Any] | Sequence[Any] | None,
+) -> str | Mapping[str, Any] | Sequence[Any] | None:
+    """Append instructions to existing instructions without discarding their structure.
+
+    ``instructions`` is declared as ``str`` on :class:`ChatOptions`, but chat clients may widen it to a
+    provider-native structured form, such as a sequence of typed instruction blocks. Combining such a
+    value with string formatting would coerce it to its ``repr``, silently turning structured metadata
+    into literal text, so a non-string base is extended element-wise instead.
+
+    The addition is always placed after the existing instructions, so the leading portion stays
+    unchanged for providers that treat it as a stable, structure-sensitive prefix.
+
+    Args:
+        base: The existing instructions, if any.
+        addition: The instructions to append, if any.
+
+    Returns:
+        The combined instructions, preserving the structure of ``base`` when it is not a string.
+    """
+    if not base:
+        return addition
+    if not addition:
+        return base
+    if isinstance(base, str) and isinstance(addition, str):
+        return f"{base}\n{addition}"
+    combined: list[Any] = [base] if isinstance(base, (str, Mapping)) else list(base)
+    combined.extend([addition] if isinstance(addition, (str, Mapping)) else addition)
+    return combined
+
+
 def merge_chat_options(
     base: dict[str, Any] | None,
     override: dict[str, Any] | None,
@@ -3966,12 +3998,8 @@ def merge_chat_options(
             continue
 
         if key == "instructions":
-            # Concatenate instructions
-            base_instructions = result.get("instructions")
-            if base_instructions:
-                result["instructions"] = f"{base_instructions}\n{value}"
-            else:
-                result["instructions"] = value
+            # Concatenate instructions, preserving provider-native structured values
+            result["instructions"] = _append_instructions(result.get("instructions"), value)
         elif key == "tools":
             # Merge tools lists
             base_tools = result.get("tools")

python/packages/core/agent_framework/observability.py

@@ -2338,16 +2338,29 @@ def _start_streaming_span(attributes: dict[str, Any], span_name_attribute: str)
 
 
 def _get_instructions_from_options(options: Any) -> str | list[str] | None:
-    """Extract instructions from options dict."""
-    if options is None:
-        return None
-    if isinstance(options, Mapping):
-        instructions = cast(Mapping[str, Any], options).get("instructions")
-        if isinstance(instructions, str):
-            return instructions
-        if isinstance(instructions, list) and all(isinstance(item, str) for item in instructions):  # type: ignore
-            return instructions  # type: ignore[reportUnknownVariableType]
+    """Extract instructions from options dict.
+
+    Chat clients may widen instructions to a provider-native structured form, so structured entries
+    contribute only their ``text`` value to keep provider metadata out of the span.
+    """
+    if not isinstance(options, Mapping):
         return None
+    instructions = cast(Mapping[str, Any], options).get("instructions")
+    if isinstance(instructions, str):
+        return instructions
+    if isinstance(instructions, Mapping):
+        text = cast(Mapping[str, Any], instructions).get("text")
+        return text if isinstance(text, str) else None
+    if isinstance(instructions, Sequence):
+        extracted: list[str] = []
+        for item in cast(Sequence[Any], instructions):
+            if isinstance(item, str):
+                extracted.append(item)
+            elif isinstance(item, Mapping):
+                text = cast(Mapping[str, Any], item).get("text")
+                if isinstance(text, str):
+                    extracted.append(text)
+        return extracted or None
     return None
 
 

python/packages/core/tests/core/test_agents.py

@@ -2806,6 +2806,50 @@ async def before_run(self, *, agent, session, context, state):
     assert options.get("instructions") == "Context-provided instructions"
 
 
+@pytest.mark.asyncio
+async def test_chat_agent_context_provider_appends_to_structured_instructions(
+    chat_client_base: SupportsChatGetResponse,
+):
+    """Context provider instructions must not stringify provider-native structured instructions.
+
+    Chat clients may widen ``instructions`` to a structured, provider-native form such as a sequence
+    of typed instruction blocks. Merging contributed instructions must append to that structure
+    rather than collapse it into text.
+    """
+
+    class InstructionContextProvider(ContextProvider):
+        def __init__(self):
+            super().__init__(source_id="instruction-context")
+
+        async def before_run(self, *, agent, session, context, state):
+            context.extend_instructions("instruction-context", "Context-provided instructions")
+
+    blocks = [
+        {"type": "text", "text": "Stable.", "block_options": {"pinned": True}},
+        {"type": "text", "text": "Dynamic."},
+    ]
+    agent = Agent(
+        client=chat_client_base,
+        default_options=cast(ChatOptions, {"instructions": blocks}),
+        context_providers=[InstructionContextProvider()],
+    )
+
+    _, options = await agent._prepare_session_and_messages(  # pyright: ignore[reportPrivateUsage]
+        session=None, input_messages=[Message(role="user", contents=["Hello"])]
+    )
+
+    assert options.get("instructions") == [*blocks, "Context-provided instructions"]
+
+
+def test_merge_options_preserves_structured_instructions() -> None:
+    """Run-level instructions append to structured default instructions without stringifying them."""
+    blocks = [{"type": "text", "text": "Stable.", "block_options": {"pinned": True}}]
+
+    merged = _merge_options({"instructions": blocks}, {"instructions": "Run-level instructions"})
+
+    assert merged["instructions"] == [*blocks, "Run-level instructions"]
+
+
 async def test_chat_agent_context_provider_adds_middleware_when_agent_has_none(
     chat_client_base: SupportsChatGetResponse,
 ) -> None:

python/packages/core/tests/core/test_observability.py

@@ -36,6 +36,7 @@
     MessageListTimestampFilter,
     OtelAttr,
     _capture_messages,
+    _get_instructions_from_options,
     get_function_span,
 )
 
@@ -4548,6 +4549,42 @@ def test_get_instructions_from_options_dict_with_instructions():
     assert _get_instructions_from_options({"other_key": "value"}) is None
 
 
+def test_get_instructions_from_options_list_of_strings():
+    """A list of plain string instructions is recorded as-is."""
+    assert _get_instructions_from_options({"instructions": ["do stuff", "be brief"]}) == ["do stuff", "be brief"]
+
+
+def test_get_instructions_from_options_structured_blocks():
+    """Structured instruction blocks contribute their text without provider metadata."""
+    blocks = [
+        {"type": "text", "text": "Stable.", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
+        {"type": "text", "text": "Dynamic."},
+    ]
+
+    assert _get_instructions_from_options({"instructions": blocks}) == ["Stable.", "Dynamic."]
+    assert _get_instructions_from_options({"instructions": blocks[0]}) == "Stable."
+
+
+def test_get_instructions_from_options_mixed_structured_and_text():
+    """Instructions appended to structured blocks during a run are still recorded."""
+    instructions = [
+        {"type": "text", "text": "Stable.", "cache_control": {"type": "ephemeral"}},
+        "Appended by a context provider",
+    ]
+
+    assert _get_instructions_from_options({"instructions": instructions}) == [
+        "Stable.",
+        "Appended by a context provider",
+    ]
+
+
+def test_get_instructions_from_options_without_extractable_text():
+    """Entries carrying no usable text yield None rather than provider metadata."""
+    assert _get_instructions_from_options({"instructions": [{"type": "image", "source": {"data": "..."}}]}) is None
+    assert _get_instructions_from_options({"instructions": []}) is None
+    assert _get_instructions_from_options({"instructions": 42}) is None
+
+
 def test_get_span_attributes_with_non_dict_options():
     """Test _get_span_attributes handles non-dict options gracefully."""
     from agent_framework.observability import _get_span_attributes