feat: add read_only flag to AgentCoreMemorySessionManager to disable ACM persistence (#389)

* 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.

* fix: address review feedback - store persistence_mode directly and fix lint errors

- Replace self.read_only boolean with self.persistence_mode to store the
  enum directly, making it easier to support future PersistenceMode values
- Fix 5 E501 line-too-long lint errors in test file
- Run ruff format on changed files

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.persistence_mode = agentcore_memory_config.persistence_mode
         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,19 @@ 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 self.persistence_mode is not PersistenceMode.NONE:
+            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 +321,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 self.persistence_mode is not PersistenceMode.NONE:
+                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 +368,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.persistence_mode is PersistenceMode.NONE:
+            return
+
         if self.config.batch_size > 1:
             # Buffer the agent state events
             should_flush = False
@@ -462,14 +469,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 self.persistence_mode is not PersistenceMode.NONE:
+                    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 +554,9 @@ def create_message(
         if not messages:
             return None
 
+        if self.persistence_mode is PersistenceMode.NONE:
+            return {}
+
         is_blob = self.converter.exceeds_conversational_limit(messages[0])
 
         # Build merged metadata from config defaults + per-call overrides
@@ -862,6 +873,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.persistence_mode is PersistenceMode.NONE:
+            return []
+
         with self._message_lock:
             messages_to_send = list(self._message_buffer)
 
@@ -940,6 +954,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.persistence_mode is PersistenceMode.NONE:
+            return []
+
         with self._agent_state_lock:
             agent_states_to_send = list(self._agent_state_buffer)