Memory

/memory                    # uses name from agentcore.json, or prompts
/memory UserContext        # targets a specific memory resource by name

Which memory strategy fits your use case?

SEMANTIC
  Best for: remembering facts about users across sessions
  How it works: extracts facts and stores them as embeddings; retrieves
  relevant context via similarity search at session start
  Cost: higher (embedding model + vector search per session)
  Example: "Remember that Alex prefers bullet points and works in fintech"

USER_PREFERENCE
  Best for: remembering explicit settings and preferences
  How it works: extracts structured preference data; optimized for
  key-value retrieval
  Cost: lower (structured extraction, no vector search)
  Example: "Remember my preferred response format and language"

SUMMARIZATION
  Best for: remembering what you talked about last time
  How it works: compresses conversation history into summaries; injects
  the summary at the start of each new session
  Cost: medium (summarization model runs at session end)
  Example: "Pick up where we left off last time"

EPISODIC
  Best for: remembering sequences of events or interactions over time
  How it works: stores episodic records of interactions with temporal
  context
  Cost: medium

Common combinations:
  SEMANTIC + USER_PREFERENCE  →  facts + preferences (most common)
  SEMANTIC + SUMMARIZATION    →  full episodic memory (highest capability, highest cost)
  USER_PREFERENCE alone       →  lightweight preference store

Which strategy (or combination) do you want?

agentcore add memory --name <MemoryName> --strategies <STRATEGY1,STRATEGY2> --expiry 30

{
  "memories": [{
    "type": "AgentCoreMemory",
    "name": "MyMemory",
    "eventExpiryDuration": 30,
    "strategies": [
      {"type": "SEMANTIC"},
      {"type": "USER_PREFERENCE"}
    ]
  }]
}

import os
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, RetrievalConfig
from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager
from strands import Agent
from model.load import load_model  # scaffolded by `agentcore create`

app = BedrockAgentCoreApp()

# AgentCore injects this env var after deploy.
# Format: MEMORY_<UPPERCASENAME>_ID
MEMORY_ID = os.getenv("MEMORY_<UPPERCASENAME>_ID")
REGION = os.getenv("AWS_REGION", "us-east-1")

@app.entrypoint
def invoke(payload, context):
    actor_id = payload.get("userId", "default-user")
    session_id = getattr(context, "session_id", "default-session")

    session_manager = None
    if MEMORY_ID:
        # RetrievalConfig parameters:
        #   top_k: max number of memory records to retrieve per namespace (SDK default: 10)
        #   relevance_score: similarity threshold, 0 = return anything, 1 = exact match (SDK default: 0.2)
        # The CLI template deviates from SDK defaults to favor precision over recall:
        #   top_k=3 limits context window usage; relevance_score=0.5 filters low-quality matches.
        # Tune these if retrieval misses relevant facts (lower) or surfaces irrelevant ones (raise).
        memory_config = AgentCoreMemoryConfig(
            memory_id=MEMORY_ID,
            session_id=session_id,
            actor_id=actor_id,
            retrieval_config={
                f"/users/{actor_id}/facts": RetrievalConfig(top_k=3, relevance_score=0.5),
                f"/users/{actor_id}/preferences": RetrievalConfig(top_k=3, relevance_score=0.5),
            }
        )
        session_manager = AgentCoreMemorySessionManager(memory_config, REGION)

    agent = Agent(
        model=load_model(),
        session_manager=session_manager,  # None is safe — agent runs without memory
        system_prompt="You are a helpful assistant.",
    )

    result = agent(payload.get("prompt", ""))
    return {"response": str(result)}

if __name__ == "__main__":
    app.run()

import os
from bedrock_agentcore.memory import MemoryClient
from strands.hooks import AgentInitializedEvent, HookProvider, MessageAddedEvent

MEMORY_ID = os.getenv("MEMORY_<UPPERCASENAME>_ID")
memory_client = MemoryClient(region_name=os.getenv("AWS_REGION", "us-east-1")) if MEMORY_ID else None

class MemoryHook(HookProvider):
    def on_agent_initialized(self, event):
        """Load recent conversation turns into the agent's context."""
        if not MEMORY_ID:
            return
        session_id = event.agent.state.get("session_id", "default")
        turns = memory_client.get_last_k_turns(
            memory_id=MEMORY_ID,
            actor_id="user",
            session_id=session_id,
            k=3
        )
        if turns:
            context = "\n".join([
                f"{m['role']}: {m['content']['text']}"
                for t in turns for m in t
            ])
            event.agent.system_prompt += f"\n\nPrevious conversation:\n{context}"

    def on_message_added(self, event):
        """Save each message to memory after it's processed."""
        if not MEMORY_ID:
            return
        session_id = event.agent.state.get("session_id", "default")
        msg = event.agent.messages[-1]
        memory_client.create_event(
            memory_id=MEMORY_ID,
            actor_id="user",
            session_id=session_id,
            messages=[(str(msg["content"]), msg["role"])]
        )

    def register_hooks(self, registry):
        registry.add_callback(AgentInitializedEvent, self.on_agent_initialized)
        registry.add_callback(MessageAddedEvent, self.on_message_added)

# Add to your existing agent:
agent = Agent(
    # ... your existing config ...
    hooks=[MemoryHook()] if MEMORY_ID else [],
    state={"session_id": "default"},
)

pip install langgraph-checkpoint-aws

import os
from langgraph.prebuilt import create_react_agent
from langgraph_checkpoint_aws import AgentCoreMemorySaver
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from model.load import load_model  # scaffolded by `agentcore create`

app = BedrockAgentCoreApp()

MEMORY_ID = os.getenv("MEMORY_<UPPERCASENAME>_ID")
REGION = os.getenv("AWS_REGION", "us-east-1")

# Only wire checkpointing if memory is available (deployed)
checkpointer = AgentCoreMemorySaver(MEMORY_ID, region_name=REGION) if MEMORY_ID else None

@app.entrypoint
async def invoke(payload, context):
    actor_id = payload.get("userId", "default-user")
    session_id = getattr(context, "session_id", "default-session")

    graph = create_react_agent(
        model=load_model(),
        tools=tools,
        checkpointer=checkpointer,  # None is safe — graph runs without persistence
    )

    # LangGraph's RunnableConfig maps thread_id → AgentCore session_id,
    # actor_id → AgentCore actor_id under the hood
    config = {
        "configurable": {
            "thread_id": session_id,
            "actor_id": actor_id,
        }
    }

    result = await graph.ainvoke(
        {"messages": [("human", payload["prompt"])]},
        config=config,
    )
    return {"response": result["messages"][-1].content}

import os
import uuid
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableConfig
from langgraph.prebuilt import create_react_agent
from langgraph.store.base import BaseStore
from langgraph_checkpoint_aws import AgentCoreMemorySaver, AgentCoreMemoryStore

MEMORY_ID = os.getenv("MEMORY_<UPPERCASENAME>_ID")
REGION = os.getenv("AWS_REGION", "us-east-1")

checkpointer = AgentCoreMemorySaver(MEMORY_ID, region_name=REGION) if MEMORY_ID else None
store = AgentCoreMemoryStore(MEMORY_ID, region_name=REGION) if MEMORY_ID else None

def pre_model_hook(state, config: RunnableConfig, *, store: BaseStore):
    """Save the latest human message for async extraction; optionally retrieve preferences."""
    actor_id = config["configurable"]["actor_id"]
    thread_id = config["configurable"]["thread_id"]
    namespace = (actor_id, thread_id)

    messages = state.get("messages", [])
    for msg in reversed(messages):
        if isinstance(msg, HumanMessage):
            store.put(namespace, str(uuid.uuid4()), {"message": msg})
            break

    # Optional: retrieve user preferences to inject into context
    # preferences_ns = ("preferences", actor_id)
    # preferences = store.search(preferences_ns, query=msg.content, limit=5)

    return {"llm_input_messages": messages}

graph = create_react_agent(
    model=load_model(),
    tools=tools,
    checkpointer=checkpointer,
    store=store,
    pre_model_hook=pre_model_hook if store else None,
)

config = {"configurable": {"thread_id": "session-1", "actor_id": "user-alice"}}
response = graph.invoke({"messages": [("human", "I prefer short answers.")]}, config=config)

# New session for the same actor — long-term memories are retrieved
new_config = {"configurable": {"thread_id": "session-2", "actor_id": "user-alice"}}
response = graph.invoke({"messages": [("human", "Summarize my latest report.")]}, config=new_config)

from bedrock_agentcore.memory import MemoryClient

client = MemoryClient(region_name=REGION)
# client.create_event(...), client.retrieve_memories(...), client.get_last_k_turns(...)

import os
from agents import Agent, Runner, function_tool
from bedrock_agentcore.memory import MemoryClient

MEMORY_ID = os.getenv("MEMORY_<UPPERCASENAME>_ID")
REGION = os.getenv("AWS_REGION", "us-east-1")
_client = MemoryClient(region_name=REGION) if MEMORY_ID else None

def _build_memory_tools(actor_id: str, session_id: str):
    """Factory — binds actor/session into tool closures."""

    @function_tool
    def recall_context(query: str, top_k: int = 3) -> str:
        """Search long-term memory for facts or preferences about the user."""
        if not _client or not MEMORY_ID:
            return "Memory unavailable."
        try:
            memories = _client.retrieve_memories(
                memory_id=MEMORY_ID,
                namespace=f"/users/{actor_id}/facts",
                query=query,
                top_k=top_k,
            )
            return "\n".join(m.get("content", {}).get("text", "") for m in memories) or "No relevant memories."
        except Exception as e:
            return f"Memory error: {e}"

    @function_tool
    def save_fact(content: str) -> str:
        """Save a fact to long-term memory."""
        if not _client or not MEMORY_ID:
            return "Memory unavailable."
        try:
            _client.create_event(
                memory_id=MEMORY_ID,
                actor_id=actor_id,
                session_id=session_id,
                messages=[(content, "ASSISTANT")],
            )
            return "Saved."
        except Exception as e:
            return f"Error: {e}"

    return [recall_context, save_fact]

@app.entrypoint
async def invoke(payload, context):
    actor_id = payload.get("userId", "default-user")
    session_id = getattr(context, "session_id", "default-session")

    agent = Agent(
        name="Assistant",
        instructions="Use recall_context at the start of each session to check what you know about the user. Use save_fact when the user tells you something worth remembering.",
        tools=_build_memory_tools(actor_id, session_id),
    )
    result = await Runner.run(agent, payload["prompt"])
    return {"response": result.final_output}

⚠️  Memory is not available during local development (agentcore dev).

The MEMORY_<NAME>_ID env var is only injected after deploy. The code above
handles this gracefully — it runs without memory when the env var isn't set.

To test memory:
  agentcore deploy -y
  agentcore invoke "My name is Alex and I prefer concise answers"
  agentcore invoke "What do you know about me?"

If using long-term memory (SEMANTIC or USER_PREFERENCE), wait 5–30 seconds
between the first and second invoke — extraction runs asynchronously after
each session ends.

Session ID note: use UUIDs (v4) for session IDs — they satisfy the platform's
minimum length requirement (33 characters) and are what `agentcore invoke`
generates by default. Short or sequential session IDs (e.g., "session-1",
"test") can cause long-term memory extraction to fail silently.

from bedrock_agentcore.memory import MemoryClient

client = MemoryClient(region_name="us-east-1")

# Create memory and wait for it to become ACTIVE (takes 2-5 minutes)
memory = client.create_memory_and_wait(
    name="UserMemory",
    strategies=[
        {"userPreferenceMemoryStrategy": {
            "name": "prefs",
            "namespaces": ["/user/preferences/"]
        }},
        {"semanticMemoryStrategy": {
            "name": "facts",
            "namespaces": ["/user/facts/"]
        }}
    ],
    event_expiry_days=30
)

MEMORY_ID = memory["id"]
print(f"Memory created: {MEMORY_ID}")
# Set this as an env var or hardcode for testing:
# export MEMORY_ID=<value>

agentcore status --type memory

import boto3, json

client = boto3.client("bedrock-agentcore-control")
memory_id = "<MEMORY_ID>"  # from: agentcore status --type memory --json

client.put_memory_resource_policy(
    memoryId=memory_id,
    policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": {"AWS": "arn:aws:iam::111122223333:root"},
            "Action": [
                "bedrock-agentcore:CreateEvent",
                "bedrock-agentcore:RetrieveMemories",
                "bedrock-agentcore:ListEvents"
            ],
            "Resource": "*"
        }]
    })
)