feat(context): add context_manager="auto" facade on Agent

strands-py/src/strands/agent/agent.py

@@ -17,6 +17,7 @@
 from typing import (
     TYPE_CHECKING,
     Any,
+    Literal,
     TypeVar,
     Union,
     cast,
@@ -139,6 +140,7 @@ def __init__(
         name: str | None = None,
         description: str | None = None,
         state: AgentState | dict | None = None,
+        context_manager: Literal["auto"] | None = None,
         plugins: list[Plugin] | None = None,
         hooks: list[HookProvider | HookCallback] | None = None,
         session_manager: SessionManager | None = None,
@@ -191,6 +193,11 @@ def __init__(
                 Defaults to None.
             state: stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
                 Defaults to an empty AgentState object.
+            context_manager: Context management strategy. When set to ``"auto"``, composes
+                a ContextOffloader plugin (max_result_tokens=1500, preview_tokens=750) with a
+                SummarizingConversationManager (summary_ratio=0.3) using benchmark-validated defaults.
+                If ``conversation_manager`` is also provided, the user's conversation manager is used
+                instead. Defaults to None (no context management).
             plugins: List of Plugin instances to extend agent functionality.
                 Plugins are initialized with the agent instance after construction and can register hooks,
                 modify agent attributes, or perform other setup tasks.
@@ -239,7 +246,11 @@ def __init__(
         else:
             self.callback_handler = callback_handler
 
-        if self.model.stateful and conversation_manager is not None:
+        resolved_cm, resolved_plugins = self._resolve_context_manager(
+            context_manager, conversation_manager, plugins
+        )
+
+        if self.model.stateful and (conversation_manager is not None or context_manager is not None):
             raise ValueError(
                 "conversation_manager cannot be used with a stateful model. "
                 "The model manages conversation state server-side."
@@ -248,6 +259,8 @@ def __init__(
         self.conversation_manager: ConversationManager
         if self.model.stateful:
             self.conversation_manager = NullConversationManager()
+        elif resolved_cm:
+            self.conversation_manager = resolved_cm
         elif conversation_manager:
             self.conversation_manager = conversation_manager
         else:
@@ -362,12 +375,49 @@ def __init__(
         # Register built-in plugins
         self._plugin_registry.add_and_init(_ModelPlugin())
 
-        if plugins:
-            for plugin in plugins:
+        plugins_to_register = resolved_plugins if resolved_plugins is not None else plugins
+        if plugins_to_register:
+            for plugin in plugins_to_register:
                 self._plugin_registry.add_and_init(plugin)
 
         self.hooks.invoke_callbacks(AgentInitializedEvent(agent=self))
 
+    @staticmethod
+    def _resolve_context_manager(
+        context_manager: "Literal['auto'] | None",
+        conversation_manager: ConversationManager | None,
+        plugins: list[Plugin] | None,
+    ) -> tuple[ConversationManager | None, list[Plugin] | None]:
+        """Resolve context_manager facade into concrete conversation_manager and plugins."""
+        if context_manager is None:
+            return None, None
+
+        from ..vended_plugins.context_offloader import ContextOffloader, InMemoryStorage
+        from .conversation_manager import SummarizingConversationManager
+
+        resolved_plugins = list(plugins) if plugins else []
+
+        has_offloader = any(
+            isinstance(p, ContextOffloader) for p in resolved_plugins
+        )
+        if not has_offloader:
+            offloader = ContextOffloader(
+                storage=InMemoryStorage(),
+                max_result_tokens=1_500,
+                preview_tokens=750,
+            )
+            resolved_plugins.insert(0, offloader)
+
+        if conversation_manager is not None:
+            resolved_cm = conversation_manager
+        else:
+            resolved_cm = SummarizingConversationManager(
+                summary_ratio=0.3,
+                proactive_compression={"compression_threshold": 0.85},
+            )
+
+        return resolved_cm, resolved_plugins
+
     def cancel(self) -> None:
         """Cancel the currently running agent invocation.
 

strands-py/tests/strands/agent/test_agent_context_manager.py

@@ -0,0 +1,103 @@
+"""Tests for the context_manager parameter on Agent."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from strands import Agent
+from strands.agent.conversation_manager import SlidingWindowConversationManager, SummarizingConversationManager
+from strands.vended_plugins.context_offloader import ContextOffloader, InMemoryStorage
+
+
+@pytest.fixture
+def mock_model():
+    model = MagicMock()
+    model.stateful = False
+    model.context_window_limit = 200_000
+    return model
+
+
+class TestContextManagerNone:
+    def test_default_preserves_sliding_window(self, mock_model):
+        agent = Agent(model=mock_model)
+        assert isinstance(agent.conversation_manager, SlidingWindowConversationManager)
+
+    def test_explicit_none_preserves_sliding_window(self, mock_model):
+        agent = Agent(model=mock_model, context_manager=None)
+        assert isinstance(agent.conversation_manager, SlidingWindowConversationManager)
+
+    def test_no_offloader_plugin_by_default(self, mock_model):
+        agent = Agent(model=mock_model)
+        assert "context_offloader" not in agent._plugin_registry._plugins
+
+
+class TestContextManagerAuto:
+    def test_uses_summarizing_conversation_manager(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert isinstance(agent.conversation_manager, SummarizingConversationManager)
+
+    def test_summary_ratio_is_benchmark_default(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert agent.conversation_manager.summary_ratio == 0.3
+
+    def test_proactive_compression_at_85_percent(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert agent.conversation_manager._compression_threshold == 0.85
+
+    def test_adds_context_offloader_plugin(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert "context_offloader" in agent._plugin_registry._plugins
+
+    def test_offloader_max_result_tokens(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        offloader = agent._plugin_registry._plugins["context_offloader"]
+        assert offloader._max_result_tokens == 1500
+
+    def test_offloader_preview_tokens(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        offloader = agent._plugin_registry._plugins["context_offloader"]
+        assert offloader._preview_tokens == 750
+
+    def test_offloader_uses_in_memory_storage(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        offloader = agent._plugin_registry._plugins["context_offloader"]
+        assert isinstance(offloader._storage, InMemoryStorage)
+
+
+class TestContextManagerCoexistence:
+    def test_user_conversation_manager_is_respected(self, mock_model):
+        user_cm = SlidingWindowConversationManager(window_size=20)
+        agent = Agent(model=mock_model, context_manager="auto", conversation_manager=user_cm)
+        assert agent.conversation_manager is user_cm
+
+    def test_offloader_still_added_with_user_cm(self, mock_model):
+        user_cm = SlidingWindowConversationManager(window_size=20)
+        agent = Agent(model=mock_model, context_manager="auto", conversation_manager=user_cm)
+        assert "context_offloader" in agent._plugin_registry._plugins
+
+    def test_user_offloader_not_overridden(self, mock_model):
+        user_offloader = ContextOffloader(storage=MagicMock(), max_result_tokens=3000, preview_tokens=1000)
+        agent = Agent(model=mock_model, context_manager="auto", plugins=[user_offloader])
+        assert agent._plugin_registry._plugins["context_offloader"]._max_result_tokens == 3000
+
+    def test_user_plugins_preserved(self, mock_model):
+        class MyPlugin:
+            name = "my_plugin"
+            hooks = []
+            tools = []
+
+            def init_agent(self, agent):
+                pass
+
+        plugin = MyPlugin()
+        agent = Agent(model=mock_model, context_manager="auto", plugins=[plugin])
+        assert "my_plugin" in agent._plugin_registry._plugins
+        assert "context_offloader" in agent._plugin_registry._plugins
+
+
+class TestContextManagerStatefulModel:
+    def test_raises_with_stateful_model(self):
+        stateful_model = MagicMock()
+        stateful_model.stateful = True
+        with pytest.raises(ValueError, match="stateful model"):
+            Agent(model=stateful_model, context_manager="auto")