feat(sdk): add system_prompt override slot to memory, skills, and summarization (#3451)

Brings three middleware classes (`MemoryMiddleware`, `SkillsMiddleware`,
`SummarizationToolMiddleware`) to parity with `FilesystemMiddleware` and
`SubAgentMiddleware`, which already accept a `system_prompt: str | None`
keyword-only argument. Downstream apps that want to suppress or replace
the SDK's middleware-appended prompt fragments — rather than have them
stacked verbatim on top of a custom top-level prompt — can now do so
per-middleware. Defaults are unchanged, so existing callers see no
behavior difference.

libs/deepagents/deepagents/middleware/memory.py

@@ -167,13 +167,9 @@ MEMORY_SYSTEM_PROMPT = """<agent_memory>
 class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]):
     """Middleware for loading agent memory from `AGENTS.md` files.
 
-    Loads memory content from configured sources and injects into the system prompt.
-
-    Supports multiple sources that are combined together.
-
-    Args:
-        backend: Backend instance or factory function for file operations.
-        sources: List of `MemorySource` configurations specifying paths and names.
+    Loads memory content from configured sources and injects into the system
+    prompt. Supports multiple sources that are combined together. See
+    constructor for the full argument list.
     """
 
     state_schema = MemoryState
@@ -184,6 +180,7 @@ class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]):
         backend: BACKEND_TYPES,
         sources: list[str],
         add_cache_control: bool = False,
+        system_prompt: str | None = MEMORY_SYSTEM_PROMPT,
     ) -> None:
         """Initialize the memory middleware.
 
@@ -210,10 +207,27 @@ class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]):
 
                 No-ops on non-Anthropic models; Bedrock and Vertex wrappers do
                 not qualify.
+            system_prompt: System-prompt fragment template. Must contain a
+                `{agent_memory}` slot for runtime memory substitution. Pass
+                `None` to skip appending entirely (memory is still loaded
+                into `state["memory_contents"]`).
+
+        Raises:
+            TypeError: If `system_prompt` is not `str` or `None`.
+            ValueError: If `system_prompt` is a string missing the
+                `{agent_memory}` format slot.
         """
+        if system_prompt is not None:
+            if not isinstance(system_prompt, str):
+                msg = f"system_prompt must be str or None, got {type(system_prompt).__name__}"
+                raise TypeError(msg)
+            if "{agent_memory}" not in system_prompt:
+                msg = "system_prompt must contain the `{agent_memory}` format slot"
+                raise ValueError(msg)
         self._backend = backend
         self.sources = sources
         self._add_cache_control = add_cache_control
+        self.system_prompt = system_prompt
 
     def _get_backend(self, state: MemoryState, runtime: Runtime, config: RunnableConfig) -> BackendProtocol:
         """Resolve backend from instance or factory.
@@ -239,25 +253,30 @@ class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]):
             return self._backend(tool_runtime)  # ty: ignore[call-top-callable, invalid-argument-type]
         return self._backend
 
-    def _format_agent_memory(self, contents: dict[str, str]) -> str:
+    def _format_agent_memory(self, contents: dict[str, str], template: str = MEMORY_SYSTEM_PROMPT) -> str:
         """Format memory with locations and contents paired together.
 
+        Substitutes loaded memory into the `{agent_memory}` slot of the
+        supplied template.
+
         Args:
             contents: Dict mapping source paths to content.
+            template: Surrounding template; must contain `{agent_memory}`.
 
         Returns:
-            Formatted string with location+content pairs wrapped in <agent_memory> tags.
+            Formatted string with location+content pairs substituted into
+            the supplied template.
         """
         if not contents:
-            return MEMORY_SYSTEM_PROMPT.format(agent_memory="(No memory loaded)")
+            return template.format(agent_memory="(No memory loaded)")
 
         sections = [f"{path}\n\n{contents[path].rstrip()}" for path in self.sources if contents.get(path)]
 
         if not sections:
-            return MEMORY_SYSTEM_PROMPT.format(agent_memory="(No memory loaded)")
+            return template.format(agent_memory="(No memory loaded)")
 
         memory_body = "\n\n".join(sections)
-        return MEMORY_SYSTEM_PROMPT.format(agent_memory=memory_body)
+        return template.format(agent_memory=memory_body)
 
     def before_agent(self, state: MemoryState, runtime: Runtime, config: RunnableConfig) -> MemoryStateUpdate | None:  # ty: ignore[invalid-method-override]
         """Load memory content before agent execution (synchronous).
@@ -336,14 +355,23 @@ class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]):
         Returns:
             Modified request with memory injected into system message.
         """
-        contents = request.state.get("memory_contents", {})
-        agent_memory = self._format_agent_memory(contents)
-
-        new_system_message = append_to_system_message(request.system_message, agent_memory)
+        if self.system_prompt is None:
+            new_system_message = request.system_message
+        else:
+            contents = request.state.get("memory_contents", {})
+            agent_memory = self._format_agent_memory(contents, self.system_prompt)
+            new_system_message = append_to_system_message(request.system_message, agent_memory)
 
         # Runtime check uses `request.model` (not a flag captured at init) so
         # the breakpoint correctly follows middleware-level model overrides.
-        if self._add_cache_control and isinstance(request.model, ChatAnthropic) and new_system_message.content_blocks:
+        # Runs regardless of `system_prompt` so callers who suppress the
+        # fragment still get the prompt-cache breakpoint they asked for.
+        if (
+            self._add_cache_control
+            and isinstance(request.model, ChatAnthropic)
+            and new_system_message is not None
+            and new_system_message.content_blocks
+        ):
             blocks: list[ContentBlock] = list(new_system_message.content_blocks)
             last = blocks[-1]
             base = last if isinstance(last, dict) else {}
@@ -352,6 +380,8 @@ class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]):
             blocks[-1] = {**base, "cache_control": {"type": "ephemeral"}}  # ty: ignore[invalid-assignment]
             new_system_message = SystemMessage(content_blocks=blocks)
 
+        if new_system_message is request.system_message:
+            return request
         return request.override(system_message=new_system_message)
 
     def wrap_model_call(

libs/deepagents/deepagents/middleware/skills.py

@@ -848,13 +848,7 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]):
         )
         ```
 
-    Args:
-        backend: Backend instance for file operations.
-        sources: List of skill sources.
-
-            Each entry is either a bare path (backwards-compatible) or a
-            `(path, label)` tuple. Bare paths derive a label from the
-            final path component; tuples use the supplied label verbatim.
+    See constructor for the full argument list.
 
     Attributes:
         sources: Paths-only view of sources (`list[str]`). Preserves the
@@ -865,7 +859,13 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]):
 
     state_schema = SkillsState
 
-    def __init__(self, *, backend: BACKEND_TYPES, sources: Sequence[SkillSource]) -> None:
+    def __init__(
+        self,
+        *,
+        backend: BACKEND_TYPES,
+        sources: Sequence[SkillSource],
+        system_prompt: str | None = SKILLS_SYSTEM_PROMPT,
+    ) -> None:
         """Initialize the skills middleware.
 
         Args:
@@ -877,18 +877,35 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]):
                 (e.g. `('/home/me/.claude/skills', 'User Claude')`). Labels
                 are rendered as `**{label} Skills**` in the system prompt
                 (do not include the trailing `Skills` in your label).
+            system_prompt: System-prompt fragment template. Must contain
+                `{skills_locations}`, `{skills_load_warnings}`, and
+                `{skills_list}` slots for runtime substitution. Pass `None`
+                to skip appending entirely (skills are still loaded into
+                `state["skills_metadata"]`).
 
         Raises:
             TypeError: If a tuple entry in `sources` is not exactly a
-                `(str, str)` pair.
+                `(str, str)` pair, or if `system_prompt` is not `str` or
+                `None`.
+            ValueError: If `system_prompt` is a string missing any of the
+                required format slots.
         """
+        if system_prompt is not None:
+            if not isinstance(system_prompt, str):
+                msg = f"system_prompt must be str or None, got {type(system_prompt).__name__}"
+                raise TypeError(msg)
+            required = ("{skills_locations}", "{skills_load_warnings}", "{skills_list}")
+            missing = [slot for slot in required if slot not in system_prompt]
+            if missing:
+                msg = f"system_prompt missing required format slot(s): {', '.join(missing)}"
+                raise ValueError(msg)
         self._backend = backend
         # `self.sources` remains paths-only (`list[str]`) to preserve
         # backwards-compat for callers that inspect it directly; label
         # information is mirrored on `self.source_labels` at the same index.
         self.sources: list[str] = [_source_path(s) for s in sources]
         self.source_labels: list[str] = [_derive_source_label(s) for s in sources]
-        self.system_prompt_template = SKILLS_SYSTEM_PROMPT
+        self.system_prompt_template = system_prompt
 
     def _get_backend(self, state: SkillsState, runtime: Runtime, config: RunnableConfig) -> BackendProtocol:
         """Resolve backend from instance or factory.
@@ -978,6 +995,9 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]):
         Returns:
             New model request with skills documentation injected into system message
         """
+        if self.system_prompt_template is None:
+            return request
+
         skills_metadata = request.state.get("skills_metadata", [])
         skills_load_errors = request.state.get("skills_load_errors", [])
         skills_locations = self._format_skills_locations()
@@ -1033,6 +1053,10 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]):
         skills = list(all_skills.values())
         update = SkillsStateUpdate(skills_metadata=skills)
         if skills_load_errors:
+            # Log even when `system_prompt_template is None`, otherwise the
+            # warnings only reach the model via the prompt fragment and
+            # silently disappear when the fragment is suppressed.
+            logger.warning("Skills load errors: %s", skills_load_errors)
             update["skills_load_errors"] = skills_load_errors
         return update
 
@@ -1075,6 +1099,10 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]):
         skills = list(all_skills.values())
         update = SkillsStateUpdate(skills_metadata=skills)
         if skills_load_errors:
+            # Log even when `system_prompt_template is None`, otherwise the
+            # warnings only reach the model via the prompt fragment and
+            # silently disappear when the fragment is suppressed.
+            logger.warning("Skills load errors: %s", skills_load_errors)
             update["skills_load_errors"] = skills_load_errors
         return update
 

libs/deepagents/deepagents/middleware/summarization.py

@@ -1311,14 +1311,31 @@ class SummarizationToolMiddleware(AgentMiddleware):
 
     state_schema = SummarizationState
 
-    def __init__(self, summarization: _DeepAgentsSummarizationMiddleware) -> None:
+    def __init__(
+        self,
+        summarization: _DeepAgentsSummarizationMiddleware,
+        *,
+        system_prompt: str | None = SUMMARIZATION_SYSTEM_PROMPT,
+    ) -> None:
         """Initialize with a reference to the summarization middleware.
 
         Args:
             summarization: The `SummarizationMiddleware` instance whose
                 summarization engine this tool will delegate to.
+            system_prompt: System-prompt fragment nudging the model to call
+                `compact_conversation`. Pass `None` to skip appending the
+                nudge entirely (the tool remains registered and callable
+                but the model is unlikely to discover it without an
+                external mention).
+
+        Raises:
+            TypeError: If `system_prompt` is not `str` or `None`.
         """
+        if system_prompt is not None and not isinstance(system_prompt, str):
+            msg = f"system_prompt must be str or None, got {type(system_prompt).__name__}"
+            raise TypeError(msg)
         self._summarization = summarization
+        self.system_prompt = system_prompt
         self.tools: list[BaseTool] = [self._create_compact_tool()]
 
     def _resolve_backend(self, runtime: ToolRuntime) -> BackendProtocol:
@@ -1583,7 +1600,9 @@ class SummarizationToolMiddleware(AgentMiddleware):
         Returns:
             The model response from the handler.
         """
-        new_system_message = append_to_system_message(request.system_message, SUMMARIZATION_SYSTEM_PROMPT)
+        if self.system_prompt is None:
+            return handler(request)
+        new_system_message = append_to_system_message(request.system_message, self.system_prompt)
         return handler(request.override(system_message=new_system_message))
 
     async def awrap_model_call(
@@ -1604,5 +1623,7 @@ class SummarizationToolMiddleware(AgentMiddleware):
         Returns:
             The model response from the handler.
         """
-        new_system_message = append_to_system_message(request.system_message, SUMMARIZATION_SYSTEM_PROMPT)
+        if self.system_prompt is None:
+            return await handler(request)
+        new_system_message = append_to_system_message(request.system_message, self.system_prompt)
         return await handler(request.override(system_message=new_system_message))

libs/deepagents/tests/unit_tests/middleware/test_compact_tool.py

@@ -5,10 +5,13 @@ from __future__ import annotations
 from typing import Any
 from unittest.mock import MagicMock, NonCallableMagicMock, patch
 
-from langchain_core.messages import AIMessage, HumanMessage
+import pytest
+from langchain.agents.middleware.types import ModelRequest
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
 from langgraph.types import Command
 
 from deepagents.middleware.summarization import (
+    SUMMARIZATION_SYSTEM_PROMPT,
     SummarizationMiddleware,
     SummarizationToolMiddleware,
     create_summarization_tool_middleware,
@@ -635,3 +638,74 @@ def test_create_summarization_tool_middleware_returns_instance() -> None:
 
     assert isinstance(mw, SummarizationToolMiddleware)
     assert mw.tools[0].name == "compact_conversation"
+
+
+# --- system_prompt override / suppression --------------------------------
+
+
+class TestSystemPromptOverride:
+    """Verify the `system_prompt` ctor arg controls the nudge fragment."""
+
+    def test_init_rejects_non_str_system_prompt(self) -> None:
+        """`system_prompt` must be str or None."""
+        with pytest.raises(TypeError, match="must be str or None"):
+            SummarizationToolMiddleware(_make_summarization_middleware(), system_prompt=0)  # type: ignore[arg-type]
+
+    def test_wrap_model_call_appends_default_nudge(self) -> None:
+        """Baseline: default `system_prompt` appends the standard nudge text."""
+        mw = _make_middleware()
+        captured: dict[str, ModelRequest] = {}
+
+        def handler(req: ModelRequest) -> None:
+            captured["req"] = req
+
+        request = ModelRequest(
+            model=GenericFakeChatModel(messages=iter([])),
+            messages=[HumanMessage(content="hi")],
+            system_message=SystemMessage(content="base"),
+            state={"messages": []},
+        )
+        mw.wrap_model_call(request, handler)  # type: ignore[arg-type]
+        appended = list(captured["req"].system_message.content_blocks)[-1].get("text", "")  # type: ignore[union-attr]
+        assert SUMMARIZATION_SYSTEM_PROMPT in appended
+
+    def test_wrap_model_call_skips_appending_when_system_prompt_none(self) -> None:
+        """`system_prompt=None` passes the request through untouched."""
+        summ = _make_summarization_middleware()
+        mw = SummarizationToolMiddleware(summ, system_prompt=None)
+        captured: dict[str, ModelRequest] = {}
+
+        def handler(req: ModelRequest) -> None:
+            captured["req"] = req
+
+        base = SystemMessage(content="base")
+        request = ModelRequest(
+            model=GenericFakeChatModel(messages=iter([])),
+            messages=[HumanMessage(content="hi")],
+            system_message=base,
+            state={"messages": []},
+        )
+        mw.wrap_model_call(request, handler)  # type: ignore[arg-type]
+        # Untouched: same request and same system_message identity.
+        assert captured["req"] is request
+        assert captured["req"].system_message is base
+
+    async def test_awrap_model_call_skips_appending_when_system_prompt_none(self) -> None:
+        """`system_prompt=None` passes the async request through untouched."""
+        summ = _make_summarization_middleware()
+        mw = SummarizationToolMiddleware(summ, system_prompt=None)
+        captured: dict[str, ModelRequest] = {}
+
+        async def handler(req: ModelRequest) -> None:
+            captured["req"] = req
+
+        base = SystemMessage(content="base")
+        request = ModelRequest(
+            model=GenericFakeChatModel(messages=iter([])),
+            messages=[HumanMessage(content="hi")],
+            system_message=base,
+            state={"messages": []},
+        )
+        await mw.awrap_model_call(request, handler)  # type: ignore[arg-type]
+        assert captured["req"] is request
+        assert captured["req"].system_message is base

libs/deepagents/tests/unit_tests/middleware/test_memory_middleware.py

@@ -10,6 +10,7 @@ from pathlib import Path
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
 
+import pytest
 from langchain.agents import create_agent
 from langchain.agents.middleware.types import ModelRequest
 from langchain_anthropic import ChatAnthropic
@@ -1047,3 +1048,66 @@ def test_create_deep_agent_wires_cache_control_for_anthropic_memory(tmp_path: Pa
     assert captured_system_messages, "Model never received a SystemMessage"
     last_block = captured_system_messages[0].content_blocks[-1]
     assert last_block.get("cache_control") == {"type": "ephemeral"}
+
+
+# --- system_prompt override / suppression --------------------------------
+
+
+def test_init_rejects_non_str_system_prompt() -> None:
+    """`system_prompt` must be str or None."""
+    with pytest.raises(TypeError, match="must be str or None"):
+        MemoryMiddleware(backend=StateBackend(), sources=[], system_prompt=0)  # type: ignore[arg-type]
+
+
+def test_init_rejects_template_missing_agent_memory_slot() -> None:
+    """Custom template without `{agent_memory}` fails fast at construction."""
+    with pytest.raises(ValueError, match="agent_memory"):
+        MemoryMiddleware(backend=StateBackend(), sources=[], system_prompt="no slot here")
+
+
+def test_modify_request_returns_unchanged_when_system_prompt_none() -> None:
+    """`system_prompt=None` skips appending; system message identical to input."""
+    middleware = MemoryMiddleware(backend=StateBackend(), sources=[], system_prompt=None)
+    base = SystemMessage(content="base")
+    request = _build_model_request(_fake_anthropic(), base)
+
+    result = middleware.modify_request(request)
+
+    assert result is request
+    assert result.system_message is base
+
+
+def test_modify_request_uses_custom_template() -> None:
+    """Custom template flows through `modify_request` instead of the default constant."""
+    middleware = MemoryMiddleware(
+        backend=StateBackend(),
+        sources=[],
+        system_prompt="CUSTOM-START\n{agent_memory}\nCUSTOM-END",
+    )
+    request = _build_model_request(_fake_anthropic(), SystemMessage(content="base"))
+
+    result = middleware.modify_request(request)
+    blocks = _system_blocks(result.system_message)
+    appended = blocks[-1].get("text", "")
+
+    assert "CUSTOM-START" in appended
+    assert "CUSTOM-END" in appended
+    assert "<agent_memory>" not in appended  # default template marker absent
+
+
+def test_modify_request_cache_control_runs_with_system_prompt_none() -> None:
+    """`add_cache_control` still attaches the breakpoint when `system_prompt=None`."""
+    middleware = MemoryMiddleware(
+        backend=StateBackend(),
+        sources=[],
+        add_cache_control=True,
+        system_prompt=None,
+    )
+    request = _build_model_request(_fake_anthropic(), SystemMessage(content="base"))
+
+    result = middleware.modify_request(request)
+    blocks = _system_blocks(result.system_message)
+
+    assert blocks[-1].get("cache_control") == {"type": "ephemeral"}
+    # No memory fragment was appended on top of `base`.
+    assert "agent_memory" not in blocks[-1].get("text", "")

libs/deepagents/tests/unit_tests/middleware/test_skills_middleware.py

@@ -14,7 +14,8 @@ from unittest.mock import MagicMock
 
 import pytest
 from langchain.agents import create_agent
-from langchain_core.messages import AIMessage, HumanMessage
+from langchain.agents.middleware.types import ModelRequest
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
 from langchain_core.runnables import RunnableConfig
 from langchain_core.runnables.config import var_child_runnable_config
 from langgraph.checkpoint.memory import InMemorySaver
@@ -2122,3 +2123,63 @@ async def test_skills_middleware_with_store_backend_assistant_id_async() -> None
     assert len(result_4["skills_metadata"]) == 1
     assert result_4["skills_metadata"][0]["name"] == "async-skill-one"
     assert result_4["skills_metadata"][0]["description"] == "Async skill for assistant 1"
+
+
+# --- system_prompt override / suppression --------------------------------
+
+
+def test_init_rejects_non_str_system_prompt() -> None:
+    """`system_prompt` must be str or None."""
+    with pytest.raises(TypeError, match="must be str or None"):
+        SkillsMiddleware(backend=StateBackend(), sources=[], system_prompt=0)  # type: ignore[arg-type]
+
+
+def test_init_rejects_template_missing_slot() -> None:
+    """Custom template missing any required slot fails fast at construction."""
+    with pytest.raises(ValueError, match="missing required format slot"):
+        SkillsMiddleware(
+            backend=StateBackend(),
+            sources=[],
+            # Missing `{skills_list}`.
+            system_prompt="{skills_locations} {skills_load_warnings}",
+        )
+
+
+def test_modify_request_returns_unchanged_when_system_prompt_none() -> None:
+    """`system_prompt=None` skips appending; system message identical to input."""
+    middleware = SkillsMiddleware(backend=StateBackend(), sources=[], system_prompt=None)
+    base = SystemMessage(content="base")
+    request = ModelRequest(
+        model=GenericFakeChatModel(messages=iter([])),  # ty: ignore[unresolved-reference]
+        messages=[HumanMessage(content="hi")],
+        system_message=base,
+        state={"messages": [], "skills_metadata": []},  # type: ignore[typeddict-unknown-key]
+    )
+
+    result = middleware.modify_request(request)
+
+    assert result is request
+    assert result.system_message is base
+
+
+def test_modify_request_uses_custom_template() -> None:
+    """Custom template flows through `modify_request` instead of the default constant."""
+    middleware = SkillsMiddleware(
+        backend=StateBackend(),
+        sources=[],
+        system_prompt="LOC:{skills_locations}|WARN:{skills_load_warnings}|LIST:{skills_list}",
+    )
+    request = ModelRequest(
+        model=GenericFakeChatModel(messages=iter([])),  # ty: ignore[unresolved-reference]
+        messages=[HumanMessage(content="hi")],
+        system_message=SystemMessage(content="base"),
+        state={"messages": [], "skills_metadata": []},  # type: ignore[typeddict-unknown-key]
+    )
+
+    result = middleware.modify_request(request)
+    appended = list(result.system_message.content_blocks)[-1].get("text", "")  # type: ignore[union-attr]
+
+    assert "LOC:" in appended
+    assert "WARN:" in appended
+    assert "LIST:" in appended
+    assert "## Skills System" not in appended  # default template marker absent