Using the Memory SDK

The Python SDK package bedrock_agentcore.memory exposes three layers, each at a different altitude. They all ultimately drive the same two API planes (bedrock-agentcore-control for create/update/get/delete memory, bedrock-agentcore for events and records) — they just differ in how much plumbing they hide.

The mental model: MemoryClient is the thin wrapper over the raw boto3 operations (it maps create_event, retrieve_memories, get_last_k_turns, strategy management, etc. onto the two planes). MemorySessionManager sits above it and gives you session objects so you stop passing memory_id / actor_id / session_id into every call. MemorySession is the per-conversation handle the manager hands back.

# Layer 3 (lower-level, 'legacy' but functional)
from bedrock_agentcore.memory import MemoryClient
client = MemoryClient(region_name="us-east-1")

# Layers 1 & 2 (recommended)
from bedrock_agentcore.memory.session import (
    MemorySessionManager, MemorySession,
    ConversationalMessage, BlobMessage, MessageRole, RetrievalConfig,
)
manager = MemorySessionManager(memory_id="mem-123", region_name="us-east-1")
session = manager.create_memory_session(actor_id="user-123", session_id="session-456")

A memory resource is created on the control plane (bedrock-agentcore-control). It does not become usable instantly — CreateMemory returns with status CREATING, and you must poll get_memory until status is ACTIVE before writing or reading. CreateMemory accepts name, description, eventExpiryDuration (days; min 7, max 365, SDK default 90), the strategy list, an optional memoryExecutionRoleArn (required for override/self-managed strategies), and an optional encryptionKeyArn for a customer-managed KMS key (otherwise AgentCore uses a service-managed key).

Raw control-plane (boto3) — note the boto3 key is memoryStrategies:

import boto3, time

control = boto3.client('bedrock-agentcore-control')
resp = control.create_memory(
    name="ShoppingSupportAgentMemory",
    eventExpiryDuration=90,   # days; min 7, max 365
    memoryStrategies=[
        {'summaryMemoryStrategy': {
            'name': 'SessionSummarizer',
            'namespaceTemplates': ['/summaries/{actorId}/{sessionId}/']}},
        {'userPreferenceMemoryStrategy': {
            'name': 'UserPreferenceExtractor',
            'namespaceTemplates': ['/users/{actorId}/preferences/']}},
    ])
memory_id = resp['memory']['id']

# Poll until ACTIVE before any data-plane call
while control.get_memory(memoryId=memory_id)['memory']['status'] == 'CREATING':
    time.sleep(5)
status = control.get_memory(memoryId=memory_id)['memory']['status']
assert status == 'ACTIVE', f"memory ended in {status}"  # CREATING -> ACTIVE | FAILED

SDK MemoryClient wraps the poll for you with create_memory_and_wait (and the SDK keyword is strategies, not memoryStrategies):

from bedrock_agentcore.memory import MemoryClient
client = MemoryClient(region_name="us-east-1")
memory = client.create_memory_and_wait(          # blocks until ACTIVE
    name="ShoppingSupportAgentMemory",
    strategies=[
        {'summaryMemoryStrategy': {'name': 'SessionSummarizer',
            'namespaceTemplates': ['/summaries/{actorId}/{sessionId}/']}},
    ],
    event_expiry_days=90,
    memory_execution_role_arn=None,   # required only for override / self-managed strategies
)
memory_id = memory['id']

The MemoryClient also offers create_or_get_memory(...) for idempotent setup in app startup code, and per-strategy helpers like add_semantic_strategy / add_summary_strategy / add_user_preference_strategy / add_episodic_strategy (and _and_wait variants) for attaching strategies after creation.

Once the memory is ACTIVE, you work on the data plane (bedrock-agentcore). There are two distinct operations to keep separate in your head:

• Write a turn with create_event — this stores a raw, immutable short-term event scoped by actorId + sessionId. Each event carries one or more messages, each with a role (USER / ASSISTANT / TOOL / OTHER).
• Read derived insight with retrieve_memory_records (boto3 RetrieveMemoryRecords) — this queries the long-term records a strategy extracted, scoped by namespace, ranked by a semantic searchQuery, capped at topK.

Raw boto3 — the conversational payload shape and searchCriteria:

data = boto3.client('bedrock-agentcore')

# Write a USER turn (payload uses the 'conversational' content + role shape)
data.create_event(
    memoryId="mem-123", actorId="customer-456", sessionId="session-789",
    payload=[{"conversational": {"content": {"text": "I'm vegetarian and allergic to peanuts"}, "role": "USER"}}])

# Retrieve long-term records by namespace + search query + topK
records = data.retrieve_memory_records(
    memoryId="mem-123",
    namespace="/users/customer-456/preferences/",
    searchCriteria={"searchQuery": "dietary restrictions", "topK": 5})

SDK MemoryClient equivalents — create_event takes (text, ROLE) tuples, and retrieve_memories exposes query / top_k / namespace:

client.create_event(
    memory_id="mem-123", actor_id="customer-456", session_id="session-789",
    messages=[("I'm vegetarian and allergic to peanuts", "USER")])

recs = client.retrieve_memories(
    memory_id="mem-123",
    namespace="/users/customer-456/preferences/",
    query="dietary restrictions",
    top_k=5)

For short-term reads (the raw turns, not extracted records) use client.get_last_k_turns(...) or client.list_events(...) — those hit ListEvents and are synchronous and immediately consistent, unlike long-term retrieval.

MemorySessionManager is the layer you should reach for on new agents. You construct it once with the memory_id and region, then mint a MemorySession per conversation. The session object remembers its actor_id / session_id, so writes and reads stop carrying repetitive identifiers.

The canonical flow is three calls: create the session → add turns → search long-term memories.

from bedrock_agentcore.memory.session import (
    MemorySessionManager, ConversationalMessage, MessageRole,
)

manager = MemorySessionManager(memory_id="mem-123", region_name="us-east-1")
session = manager.create_memory_session(actor_id="user-123", session_id="session-456")

# Write a USER + ASSISTANT turn as typed ConversationalMessage objects
session.add_turns([
    ConversationalMessage("I love apples", MessageRole.USER),
    ConversationalMessage("Apples are nutritious!", MessageRole.ASSISTANT),
])

# Later (a different session for the same actor), retrieve what was extracted
memories = session.search_long_term_memories(
    query="what food does the user like",
    namespace_path="/food/user-123/",
    top_k=5)

Notice the type system: instead of (text, role) tuples or raw {"conversational": ...} dicts, you pass ConversationalMessage(text, MessageRole.USER) objects (there's also BlobMessage for non-conversational payloads, e.g. tool output blobs). search_long_term_memories is the manager-layer name for the long-term query — same semantics as retrieve_memory_records: a namespace_path, a semantic query, and a top_k cap.

The payoff is plain when you serve many users from one process: build the manager once at startup, and call create_memory_session(actor_id=..., session_id=...) per request. No identifier threading, no accidental cross-actor writes.

process_turn_with_llm collapses the most common agent loop — retrieve relevant memories → call the LLM with them → store the new turn — into a single call. You hand it the user's input, a retrieval_config (a map of namespace → RetrievalConfig), and an llm_callback that does the actual model invocation. It returns a tuple of the retrieved memories, the model's response, and the stored event.

from bedrock_agentcore.memory.session import (
    MemorySessionManager, RetrievalConfig,
)

manager = MemorySessionManager(memory_id="mem-123", region_name="us-east-1")
session = manager.create_memory_session(actor_id="user-123", session_id="session-456")

def my_llm(user_input: str, retrieved_memories) -> str:
    # Build a prompt from the retrieved long-term memories + the user's question,
    # call your model (Bedrock, OpenAI, etc.), and return the assistant text.
    context = "\n".join(str(m) for m in retrieved_memories)
    return call_model(system=f"Known about the user:\n{context}", user=user_input)

memories, response, event = session.process_turn_with_llm(
    user_input="What did we discuss?",
    llm_callback=my_llm,
    retrieval_config={
        "support/facts/{sessionId}/": RetrievalConfig(top_k=5, relevance_score=0.3),
    })

The retrieval_config keys are namespaces (template variables like {sessionId} are substituted for you), and each RetrievalConfig tunes that namespace's retrieval: top_k caps the number of records, relevance_score sets a minimum similarity threshold so low-confidence matches are dropped before they reach your prompt. You can list multiple namespaces to pull, say, both per-session facts and per-actor preferences in one pass.

This helper is convenience, not magic: under the hood it's still search_long_term_memories (or retrieve_memory_records) → your callback → add_turns/create_event. If you need to interleave tool calls or branch the conversation, drop back to the explicit three-call flow.

The single biggest gotcha when coding against Memory: long-term extraction is asynchronous. When you create_event, the raw event is stored synchronously and is immediately readable via get_last_k_turns / list_events. But the extraction pipeline that turns that event into long-term records runs in the background — roughly 20–40 seconds for built-in strategies (retrieval itself is ~200 ms once records exist) (as of writing — verify the live limits page; episodic episodes take longer still). A search_long_term_memories / retrieve_memory_records fired immediately after create_event will often return nothing — not an error, just empty.

Design for this. In tutorials and tests you'll see an explicit wait; in production you retrieve memories that were extracted on earlier turns, so the lag is naturally hidden.

import time

session.add_turns([ConversationalMessage("I'm vegetarian", MessageRole.USER)])

# Extraction is async (~20-40s for built-in strategies). An immediate read may be empty.
time.sleep(40)  # acceptable in a test/tutorial; in prod, read records from prior turns
memories = session.search_long_term_memories(
    query="dietary restrictions", namespace_path="/users/user-123/preferences/", top_k=5)

The exceptions to handle. The SDK surfaces botocore errors; the ones you'll actually see:

from botocore.exceptions import ClientError, NoCredentialsError

try:
    recs = client.retrieve_memories(memory_id="mem-123",
        namespace="/users/customer-456/preferences/", query="diet", top_k=5)
except ClientError as e:
    code = e.response["Error"]["Code"]
    if code == "ResourceNotFoundException":
        ...  # bad id / not ACTIVE
    elif code == "ValidationException":
        ...  # bad request shape — do not blindly retry
    elif code == "ThrottlingException":
        ...  # back off + retry with jitter
    elif code == "AccessDeniedException":
        ...  # IAM / namespace-scoped condition keys
    else:
        raise
except NoCredentialsError:
    ...  # no AWS credentials in the environment