feat: add persistence_mode to AgentCoreMemorySessionManager to control ACM persistence

Add PersistenceMode enum (FULL | NONE) and persistence_mode config option
to AgentCoreMemoryConfig. When set to NONE, disables persistence to
AgentCore Memory (create_event calls) while keeping local session/agent
state management and memory injection (LTM retrieval) working.

This allows customers who only need memory injection into agent context
without persisting agent state to set persistence_mode=PersistenceMode.NONE
on their existing configuration.

Author: mohanasudhan

src/bedrock_agentcore/memory/integrations/strands/config.py

@@ -1,5 +1,6 @@
 """Configuration for AgentCore Memory Session Manager."""
 
+from enum import Enum
 from typing import Any, Callable, Dict, Optional
 
 from pydantic import BaseModel, Field, field_validator
@@ -10,6 +11,19 @@ def normalize_metadata(raw: Dict[str, Any]) -> Dict[str, Any]:
     return {k: {"stringValue": v} if isinstance(v, str) else v for k, v in raw.items()}
 
 
+class PersistenceMode(str, Enum):
+    """Controls what gets persisted to AgentCore Memory.
+
+    Attributes:
+        FULL: Persist everything (session, agent state, messages) to AgentCore Memory. Default behavior.
+        NONE: Disable all persistence. Local session/agent state management and memory injection
+            (LTM retrieval) still work, but no create_event calls are made to AgentCore Memory.
+    """
+
+    FULL = "FULL"
+    NONE = "NONE"
+
+
 class RetrievalConfig(BaseModel):
     """Configuration for memory retrieval operations.
 
@@ -51,6 +65,9 @@ class AgentCoreMemoryConfig(BaseModel):
             event creation, so it can return dynamic values (e.g. current traceId). The returned
             dict is merged after default_metadata but before per-call metadata.
             Accepts plain strings (auto-wrapped) or explicit MetadataValue dicts.
+        persistence_mode: Controls what gets persisted to AgentCore Memory.
+            FULL (default): persist everything. NONE: disable all persistence while keeping
+            local state management and memory injection working.
     """
 
     memory_id: str = Field(min_length=1)
@@ -63,6 +80,7 @@ class AgentCoreMemoryConfig(BaseModel):
     filter_restored_tool_context: bool = Field(default=False)
     default_metadata: Optional[Dict[str, Any]] = None
     metadata_provider: Optional[Callable[[], Dict[str, Any]]] = None
+    persistence_mode: PersistenceMode = Field(default=PersistenceMode.FULL)
 
     @field_validator("default_metadata", mode="before")
     @classmethod

src/bedrock_agentcore/memory/integrations/strands/session_manager.py

@@ -29,7 +29,7 @@
 )
 
 from .bedrock_converter import AgentCoreMemoryConverter
-from .config import AgentCoreMemoryConfig, RetrievalConfig, normalize_metadata
+from .config import AgentCoreMemoryConfig, PersistenceMode, RetrievalConfig, normalize_metadata
 from .converters import MemoryConverter
 
 if TYPE_CHECKING:
@@ -142,6 +142,7 @@ def __init__(
         """
         self.converter = converter or AgentCoreMemoryConverter
         self.config = agentcore_memory_config
+        self.read_only = agentcore_memory_config.persistence_mode is PersistenceMode.NONE
         self.memory_client = MemoryClient(region_name=region_name)
         session = boto_session or boto3.Session(region_name=region_name)
         self.has_existing_agent = False
@@ -255,17 +256,21 @@ def create_session(self, session: Session, **kwargs: Any) -> Session:
         if session.session_id != self.config.session_id:
             raise SessionException(f"Session ID mismatch: expected {self.config.session_id}, got {session.session_id}")
 
-        event = self.memory_client.gmdp_client.create_event(
-            memoryId=self.config.memory_id,
-            actorId=self.config.actor_id,
-            sessionId=self.session_id,
-            payload=[
-                {"blob": json.dumps(session.to_dict())},
-            ],
-            eventTimestamp=self._get_monotonic_timestamp(),
-            metadata={STATE_TYPE_KEY: {"stringValue": StateType.SESSION.value}},
-        )
-        logger.info("Created session: %s with event: %s", session.session_id, event.get("event", {}).get("eventId"))
+        if not self.read_only:
+            event = self.memory_client.gmdp_client.create_event(
+                memoryId=self.config.memory_id,
+                actorId=self.config.actor_id,
+                sessionId=self.session_id,
+                payload=[
+                    {"blob": json.dumps(session.to_dict())},
+                ],
+                eventTimestamp=self._get_monotonic_timestamp(),
+                metadata={STATE_TYPE_KEY: {"stringValue": StateType.SESSION.value}},
+            )
+            logger.info(
+                "Created session: %s with event: %s", session.session_id, event.get("event", {}).get("eventId")
+            )
+
         return session
 
     def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
@@ -318,14 +323,15 @@ def read_session(self, session_id: str, **kwargs: Any) -> Optional[Session]:
             session_data = json.loads(old_event.get("payload", {})[0].get("blob"))
             session = Session.from_dict(session_data)
             # Migrate: create new event with metadata, delete old
-            self.create_session(session)
-            self.memory_client.gmdp_client.delete_event(
-                memoryId=self.config.memory_id,
-                actorId=legacy_actor_id,
-                sessionId=session_id,
-                eventId=old_event.get("eventId"),
-            )
-            logger.info("Migrated legacy session event for session: %s", session_id)
+            if not self.read_only:
+                self.create_session(session)
+                self.memory_client.gmdp_client.delete_event(
+                    memoryId=self.config.memory_id,
+                    actorId=legacy_actor_id,
+                    sessionId=session_id,
+                    eventId=old_event.get("eventId"),
+                )
+                logger.info("Migrated legacy session event for session: %s", session_id)
             return session
 
         return None
@@ -364,6 +370,9 @@ def create_agent(self, session_id: str, session_agent: SessionAgent, **kwargs: A
         if session_agent.created_at:
             self._agent_created_at_cache[session_agent.agent_id] = session_agent.created_at
 
+        if self.read_only:
+            return
+
         if self.config.batch_size > 1:
             # Buffer the agent state events
             should_flush = False
@@ -462,14 +471,15 @@ def read_agent(self, session_id: str, agent_id: str, **kwargs: Any) -> Optional[
                 agent_data = json.loads(old_event.get("payload", {})[0].get("blob"))
                 agent = SessionAgent.from_dict(agent_data)
                 # Migrate: create new event with metadata, delete old
-                self.create_agent(session_id, agent)
-                self.memory_client.gmdp_client.delete_event(
-                    memoryId=self.config.memory_id,
-                    actorId=legacy_actor_id,
-                    sessionId=session_id,
-                    eventId=old_event.get("eventId"),
-                )
-                logger.info("Migrated legacy agent event for agent: %s", agent_id)
+                if not self.read_only:
+                    self.create_agent(session_id, agent)
+                    self.memory_client.gmdp_client.delete_event(
+                        memoryId=self.config.memory_id,
+                        actorId=legacy_actor_id,
+                        sessionId=session_id,
+                        eventId=old_event.get("eventId"),
+                    )
+                    logger.info("Migrated legacy agent event for agent: %s", agent_id)
                 return agent
 
             return None
@@ -546,6 +556,9 @@ def create_message(
         if not messages:
             return None
 
+        if self.read_only:
+            return {}
+
         is_blob = self.converter.exceeds_conversational_limit(messages[0])
 
         # Build merged metadata from config defaults + per-call overrides
@@ -862,6 +875,9 @@ def _flush_messages_only(self) -> list[dict[str, Any]]:
         Raises:
             SessionException: If message creation fails. On failure, messages remain in the buffer.
         """
+        if self.read_only:
+            return []
+
         with self._message_lock:
             messages_to_send = list(self._message_buffer)
 
@@ -940,6 +956,9 @@ def _flush_agent_states_only(self) -> list[dict[str, Any]]:
         Raises:
             SessionException: If agent state creation fails. On failure, agent states remain in the buffer.
         """
+        if self.read_only:
+            return []
+
         with self._agent_state_lock:
             agent_states_to_send = list(self._agent_state_buffer)
 

tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py

@@ -18,7 +18,7 @@
     CONVERSATIONAL_MAX_SIZE,
     AgentCoreMemoryConverter,
 )
-from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, RetrievalConfig
+from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, PersistenceMode, RetrievalConfig
 from bedrock_agentcore.memory.integrations.strands.session_manager import (
     AgentCoreMemorySessionManager,
     BufferedMessage,
@@ -3234,3 +3234,200 @@ def test_default_metadata_plain_strings_normalized(self, mock_memory_client):
 
         kwargs = mock_memory_client.create_event.call_args[1]
         assert kwargs["metadata"]["project"] == {"stringValue": "atlas"}
+
+
+class TestPersistenceMode:
+    """Test persistence_mode=NONE disables ACM persistence but keeps local state management and LTM retrieval."""
+
+    @pytest.fixture
+    def no_persist_config(self):
+        return AgentCoreMemoryConfig(
+            memory_id="test-memory-123",
+            session_id="test-session-456",
+            actor_id="test-actor-789",
+            persistence_mode=PersistenceMode.NONE,
+        )
+
+    @pytest.fixture
+    def no_persist_manager(self, no_persist_config, mock_memory_client):
+        return _create_session_manager(no_persist_config, mock_memory_client)
+
+    # --- Config ---
+
+    def test_config_defaults_to_full(self):
+        config = AgentCoreMemoryConfig(memory_id="m", session_id="s", actor_id="a")
+        assert config.persistence_mode is PersistenceMode.FULL
+
+    def test_config_accepts_none_mode(self):
+        config = AgentCoreMemoryConfig(memory_id="m", session_id="s", actor_id="a", persistence_mode=PersistenceMode.NONE)
+        assert config.persistence_mode is PersistenceMode.NONE
+
+    def test_config_accepts_string_value(self):
+        config = AgentCoreMemoryConfig(memory_id="m", session_id="s", actor_id="a", persistence_mode="NONE")
+        assert config.persistence_mode is PersistenceMode.NONE
+
+    # --- create_session: local state works, no ACM write ---
+
+    def test_create_session_returns_session(self, no_persist_manager, mock_memory_client):
+        session = Session(session_id="test-session-456", session_type=SessionType.AGENT)
+        result = no_persist_manager.create_session(session)
+        assert result.session_id == "test-session-456"
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    def test_create_session_still_validates(self, no_persist_manager):
+        session = Session(session_id="wrong-id", session_type=SessionType.AGENT)
+        with pytest.raises(SessionException, match="Session ID mismatch"):
+            no_persist_manager.create_session(session)
+
+    # --- create_agent: local cache works, no ACM write ---
+
+    def test_create_agent_caches_timestamp(self, no_persist_manager, mock_memory_client):
+        agent = SessionAgent(agent_id="a1", state={}, conversation_manager_state={})
+        no_persist_manager.create_agent("test-session-456", agent)
+        assert "a1" in no_persist_manager._agent_created_at_cache
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    def test_create_agent_still_validates(self, no_persist_manager):
+        agent = SessionAgent(agent_id="a1", state={}, conversation_manager_state={})
+        with pytest.raises(SessionException, match="Session ID mismatch"):
+            no_persist_manager.create_agent("wrong-id", agent)
+
+    # --- update_agent: local cache works, no ACM write ---
+
+    def test_update_agent_no_acm_write(self, no_persist_manager, mock_memory_client):
+        no_persist_manager._agent_created_at_cache["a1"] = "2024-01-01T00:00:00+00:00"
+        agent = SessionAgent(agent_id="a1", state={"k": "v"}, conversation_manager_state={})
+        no_persist_manager.update_agent("test-session-456", agent)
+        assert agent.created_at == "2024-01-01T00:00:00+00:00"
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    # --- create_message: validation works, returns non-None for append_message, no ACM write ---
+
+    def test_create_message_returns_empty_dict(self, no_persist_manager, mock_memory_client):
+        msg = SessionMessage(
+            message={"role": "user", "content": [{"text": "hi"}]},
+            message_id=1,
+            created_at="2024-01-01T12:00:00Z",
+        )
+        result = no_persist_manager.create_message("test-session-456", "a1", msg)
+        assert result == {}
+        mock_memory_client.create_event.assert_not_called()
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    def test_create_message_still_validates(self, no_persist_manager):
+        msg = SessionMessage(
+            message={"role": "user", "content": [{"text": "hi"}]},
+            message_id=1,
+            created_at="2024-01-01T12:00:00Z",
+        )
+        with pytest.raises(SessionException, match="Session ID mismatch"):
+            no_persist_manager.create_message("wrong-id", "a1", msg)
+
+    def test_create_message_returns_none_for_empty_payload(self, no_persist_manager):
+        msg = SessionMessage(
+            message={"role": "user", "content": []},
+            message_id=1,
+            created_at="2024-01-01T12:00:00Z",
+        )
+        result = no_persist_manager.create_message("test-session-456", "a1", msg)
+        assert result is None
+
+    # --- append_message: local state tracking works ---
+
+    def test_append_message_tracks_local_state(self, no_persist_manager, mock_memory_client):
+        no_persist_manager._latest_agent_message = {}
+        mock_agent = Mock()
+        mock_agent.agent_id = "a1"
+        message = {"role": "user", "content": [{"text": "hello"}]}
+        no_persist_manager.append_message(message, mock_agent)
+        assert "a1" in no_persist_manager._latest_agent_message
+        mock_memory_client.create_event.assert_not_called()
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    # --- flush: no-ops ---
+
+    def test_flush_messages_only_noop(self, no_persist_manager, mock_memory_client):
+        assert no_persist_manager._flush_messages_only() == []
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    def test_flush_agent_states_only_noop(self, no_persist_manager, mock_memory_client):
+        assert no_persist_manager._flush_agent_states_only() == []
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    def test_close_no_acm_writes(self, no_persist_manager, mock_memory_client):
+        no_persist_manager.close()
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+
+    # --- reads still work ---
+
+    def test_read_session_works(self, no_persist_manager, mock_memory_client):
+        mock_memory_client.list_events.return_value = [
+            {"eventId": "e1", "payload": [{"blob": '{"session_id": "test-session-456", "session_type": "AGENT"}'}]},
+        ]
+        result = no_persist_manager.read_session("test-session-456")
+        assert result is not None
+        assert result.session_id == "test-session-456"
+
+    def test_read_agent_works(self, no_persist_manager, mock_memory_client):
+        mock_memory_client.list_events.return_value = [
+            {"eventId": "e1", "payload": [{"blob": '{"agent_id": "a1", "state": {}, "conversation_manager_state": {}}'}]},
+        ]
+        result = no_persist_manager.read_agent("test-session-456", "a1")
+        assert result is not None
+        assert result.agent_id == "a1"
+
+    def test_list_messages_works(self, no_persist_manager, mock_memory_client):
+        mock_memory_client.list_events.return_value = [
+            {
+                "eventId": "e1",
+                "eventTimestamp": "2024-01-01T12:00:00Z",
+                "payload": [{"conversational": {"content": {"text": '{"message": {"role": "user", "content": [{"text": "Hello"}]}, "message_id": 1}'}, "role": "USER"}}],
+            },
+        ]
+        messages = no_persist_manager.list_messages("test-session-456", "a1")
+        assert len(messages) == 1
+
+    # --- legacy migration skipped ---
+
+    def test_legacy_session_migration_skipped(self, no_persist_manager, mock_memory_client):
+        mock_memory_client.list_events.side_effect = [
+            [],
+            [{"eventId": "legacy-1", "payload": [{"blob": '{"session_id": "test-session-456", "session_type": "AGENT"}'}]}],
+        ]
+        result = no_persist_manager.read_session("test-session-456")
+        assert result is not None
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+        mock_memory_client.gmdp_client.delete_event.assert_not_called()
+
+    def test_legacy_agent_migration_skipped(self, no_persist_manager, mock_memory_client):
+        mock_memory_client.list_events.side_effect = [
+            [],
+            [{"eventId": "legacy-1", "payload": [{"blob": '{"agent_id": "a1", "state": {}, "conversation_manager_state": {}}'}]}],
+        ]
+        result = no_persist_manager.read_agent("test-session-456", "a1")
+        assert result is not None
+        mock_memory_client.gmdp_client.create_event.assert_not_called()
+        mock_memory_client.gmdp_client.delete_event.assert_not_called()
+
+    # --- LTM retrieval still works ---
+
+    def test_retrieve_customer_context_works(self, mock_memory_client):
+        config = AgentCoreMemoryConfig(
+            memory_id="test-memory-123",
+            session_id="test-session-456",
+            actor_id="test-actor-789",
+            persistence_mode=PersistenceMode.NONE,
+            retrieval_config={"ns/": RetrievalConfig(top_k=5, relevance_score=0.3)},
+        )
+        manager = _create_session_manager(config, mock_memory_client)
+        mock_memory_client.retrieve_memories.return_value = [
+            {"content": {"text": "remembered fact"}},
+        ]
+
+        mock_agent = Mock()
+        mock_agent.messages = [{"role": "user", "content": [{"text": "query"}]}]
+        event = MessageAddedEvent(agent=mock_agent, message={"role": "user", "content": [{"text": "query"}]})
+        manager.retrieve_customer_context(event)
+
+        mock_memory_client.retrieve_memories.assert_called_once()
+        assert "<user_context>" in mock_agent.messages[0]["content"][0]["text"]