AWS Builder Center

Builder Center

Building a Multi-Agent AWS Platform with Amazon Bedrock AgentCore — Part 5: Memory — Short-Term and Long-Term Context

In Part 5 we integrate AgentCore Memory to give agents persistent context. You'll learn how to store session-scoped memory, retrieve long-term facts across sessions, use semantic search over past interactions, and implement artifact versioning with rollback support.

Mohan Alapati

Sr. Solutions Architect, AWS WWPS State&L Government SA

BODY


Series navigation ←  Part 4: MCP Tool Integration via AgentCore Gateway  →  Part 6: Policy as Code with Cedar 


The memory problem in multi-turn agents

Without memory, every agent invocation starts from scratch. The user says "add a Redis cache layer" and the agent has no idea what diagram they're referring to. You'd have to send the entire conversation history on every request — which quickly exhausts the context window and adds latency.
AgentCore Memory solves this with two tiers:
  • Session memory — key/value pairs scoped to a single conversation session. The Diagram Agent stores the current  DiagramSpec  here so the Terraform Agent can use it as the source of truth.
  • Long-term memory — facts that persist across sessions. When a user always works in  eu-west-1 , that preference is remembered and injected into future prompts automatically.
Both tiers use the  bedrock-agentcore  boto3 client with  create_event  and  list_events  under the hood, with a local in-process dict as fallback for development.


The MemoryClient

Python
# backend/memory/memory_client.py
from __future__ import annotations
import json
import logging
import os
import uuid
from datetime import datetime, timezone
from typing import Any

logger = logging.getLogger(__name__)

_ENABLED = os.environ.get("AGENTCORE_MEMORY_ENABLED", "").lower() in ("1", "true", "yes")
_LONG_TERM_PREFIX = "long_term"

class MemoryClient:
    """AgentCore Memory client with local fallback.

    Set AGENTCORE_MEMORY_ENABLED=true to use the real AgentCore service.
    Without it, an in-process dict provides equivalent behaviour for local dev.
    """

    def __init__(self, memory_id: str, region_name: str = "us-east-1") -> None:
        self._memory_id = memory_id
        self._region = region_name
        self._local_store: dict[str, list[dict]] = {}

        if _ENABLED:
            from backend.aws_session import get_session
            self._client = get_session().client(
                "bedrock-agentcore", region_name=region_name
            )
            logger.info("AgentCore Memory enabled — memory_id=%s", memory_id)
        else:
            self._client = None
            logger.info("AgentCore Memory disabled — using local store")


Session memory: storing context

Session memory is written after every significant agent action. The Diagram Agent stores the current spec so the Terraform Agent can read it:
Python
def store_session_memory(
    self, session_id: str, user_id: str, key: str, value: Any
) -> None:
    """Store a key/value pair in session memory."""
    payload_text = json.dumps({"key": key, "value": value})

    if self._client:
        try:
            self._client.create_event(
                memoryId=self._memory_id,
                actorId=user_id,
                sessionId=session_id,
                eventTimestamp=datetime.now(timezone.utc),
                payload=[{
                    "conversational": {
                        "content": {"text": payload_text},
                        "role": "ASSISTANT",
                    }
                }],
            )
            return
        except Exception as exc:
            logger.warning("AgentCore Memory write failed (%s); using local store", exc)

    # Local fallback
    store_key = f"{user_id}:{session_id}"
    entries = self._local_store.setdefault(store_key, [])
    # Update existing entry if key already exists
    existing = next((e for e in entries if e.get("key") == key), None)
    if existing:
        existing["value"] = value
    else:
        entries.append({
            "entryId": str(uuid.uuid4()),
            "key": key,
            "value": value,
            "createdAt": datetime.now(timezone.utc).isoformat(),
        })


Session memory: loading context

Reading session memory returns a list of  MemoryEntry  objects. Agents walk this list in reverse to find the most recent value for a given key:
Python
def load_session_memory(self, session_id: str, user_id: str) -> list:
    """Retrieve all session-scoped memory entries."""
    from backend.models.memory import MemoryEntry

    if self._client:
        try:
            response = self._client.list_events(
                memoryId=self._memory_id,
                actorId=user_id,
                sessionId=session_id,
                includePayloads=True,
                maxResults=100,
            )
            entries = []
            for event in response.get("events", []):
                for item in event.get("payload", []):
                    text = item.get("conversational", {}).get("content", {}).get("text", "")
                    try:
                        data = json.loads(text)
                        if isinstance(data, dict) and "key" in data:
                            entries.append(MemoryEntry(
                                entry_id=event.get("eventId", str(uuid.uuid4())),
                                user_id=user_id,
                                session_id=session_id,
                                scope="session",
                                key=data["key"],
                                value=data.get("value"),
                                created_at=datetime.now(timezone.utc).isoformat(),
                            ))
                    except (json.JSONDecodeError, TypeError):
                        continue
            if entries:
                return entries
        except Exception as exc:
            logger.debug("AgentCore Memory read failed (%s); using local store", exc)

    # Local fallback
    store_key = f"{user_id}:{session_id}"
    return [
        MemoryEntry(
            entry_id=raw.get("entryId", str(uuid.uuid4())),
            user_id=user_id,
            session_id=session_id,
            scope="session",
            key=raw["key"],
            value=raw.get("value"),
            created_at=raw.get("createdAt", ""),
        )
        for raw in self._local_store.get(store_key, [])
    ]


Long-term memory: cross-session facts

Long-term memory uses a dedicated session ID per user ( long_term_{userId} ). Facts stored here survive across sessions:
Python
def store_long_term_memory(self, user_id: str, key: str, value: Any) -> None:
    """Store a fact in long-term (cross-session) memory."""
    lt_session_id = f"{_LONG_TERM_PREFIX}_{user_id}"
    payload_text = json.dumps({"key": key, "value": value, "scope": "long_term"})

    if self._client:
        try:
            self._client.create_event(
                memoryId=self._memory_id,
                actorId=user_id,
                sessionId=lt_session_id,
                eventTimestamp=datetime.now(timezone.utc),
                payload=[{
                    "conversational": {
                        "content": {"text": payload_text},
                        "role": "ASSISTANT",
                    }
                }],
            )
            return
        except Exception as exc:
            logger.warning("Long-term memory write failed (%s); using local store", exc)

    store_key = f"{user_id}:long_term"
    entries = self._local_store.setdefault(store_key, [])
    existing = next((e for e in entries if e.get("key") == key), None)
    if existing:
        existing["value"] = value
    else:
        entries.append({"key": key, "value": value})
The Orchestrator extracts facts from each interaction and stores them in long-term memory:
Python
# In OrchestratorAgent._store_interaction_summary()
def _store_interaction_summary(
    self, session_id: str, user_id: str,
    message: str, artifact_type: str, artifact: dict
) -> None:
    """Extract and store long-term facts from an interaction."""
    # Extract preferred region from the message
    import re
    region_match = re.search(r'\b(us-east-1|us-west-2|eu-west-1|ap-southeast-1)\b', message)
    if region_match:
        self._memory.store_long_term_memory(
            user_id=user_id,
            key="preferred_region",
            value=region_match.group(1),
        )

    # Store the artifact type preference
    self._memory.store_long_term_memory(
        user_id=user_id,
        key=f"last_{artifact_type}_request",
        value={"message": message[:200], "timestamp": datetime.now(timezone.utc).isoformat()},
    )
On the next session, the Orchestrator loads these facts and injects them into the agent context:
Python
def _load_session_context(self, session_id: str, user_id: str) -> dict:
    """Load session + long-term memory into a context dict."""
    context = {}

    # Session memory
    session_entries = self._memory.load_session_memory(session_id, user_id)
    for entry in session_entries:
        context[entry.key] = entry.value

    # Long-term memory — inject as past_facts string for the prompt
    lt_entries = self._memory.load_long_term_memory(user_id)
    if lt_entries:
        facts = "\n".join(
            f"- {e.key}: {e.value}" for e in lt_entries[:10]
        )
        context["past_facts"] = facts

    return context
The  past_facts  string gets injected into every Claude prompt, giving the agent awareness of the user's history without sending the full conversation.


Semantic search over long-term memory

AgentCore Memory supports semantic search via  retrieve_memory_records . This lets agents find relevant past interactions using natural language queries rather than exact key lookups:
Python
def retrieve_memories(
    self,
    user_id: str,
    query: str,
    top_k: int = 5,
    strategy_id: str | None = None,
) -> list[dict]:
    """Semantic search over long-term memory records.

    Returns the top_k most relevant memories for the query.
    """
    if not self._client:
        return []

    namespace = (
        f"/strategies/{strategy_id}/actors/{user_id}/"
        if strategy_id
        else f"/actors/{user_id}/"
    )

    try:
        response = self._client.retrieve_memory_records(
            memoryId=self._memory_id,
            namespace=namespace,
            searchCriteria={
                "searchQuery": query,
                "topK": top_k,
                **({"memoryStrategyId": strategy_id} if strategy_id else {}),
            },
            maxResults=top_k,
        )

        results = []
        for record in response.get("memoryRecordSummaries", []):
            content = record.get("content", {})
            text = content.get("text") or content.get("value") or str(content)
            results.append({
                "content": text,
                "score": record.get("score", 0.0),
                "memoryRecordId": record.get("memoryRecordId", ""),
            })

        logger.debug(
            "Retrieved %d semantic memory records for query=%r",
            len(results), query[:80],
        )
        return results

    except Exception as exc:
        logger.debug("retrieve_memories failed (%s) — returning empty", exc)
        return []
Usage in the Orchestrator:
Python
# Find relevant past interactions before generating a new diagram
relevant_memories = self._memory.retrieve_memories(
    user_id=user_id,
    query=f"AWS architecture for: {message}",
    top_k=3,
)
if relevant_memories:
    context["relevant_past"] = "\n".join(
        m["content"] for m in relevant_memories
    )


Artifact versioning and rollback

Every time an agent produces a new artifact (diagram or Terraform module), we append a version to the revision history. This enables one-click rollback from the frontend:
Python
# backend/memory/versioning.py
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from backend.memory.memory_client import MemoryClient

_VERSIONS_KEY = "artifact_versions"

def append_artifact_version(
    session_id: str,
    artifact: dict,
    memory_client: MemoryClient,
) -> dict:
    """Append a new artifact version to the session's revision history.

    Monotonically increments the version number and sets parentVersionId.
    """
    existing = get_artifact_versions(session_id, memory_client)
    version_num = len(existing) + 1
    parent_id = existing[-1]["versionId"] if existing else None

    version_entry = {
        "versionId": str(uuid.uuid4()),
        "sessionId": session_id,
        "version": version_num,
        "parentVersionId": parent_id,
        "artifact": artifact,
        "createdAt": datetime.now(timezone.utc).isoformat(),
    }

    updated = existing + [version_entry]
    memory_client.store_session_memory(
        session_id=session_id,
        user_id=artifact.get("userId", ""),
        key=_VERSIONS_KEY,
        value=updated,
    )
    return version_entry

def get_artifact_versions(
    session_id: str, memory_client: MemoryClient
) -> list[dict]:
    """Return all artifact versions for a session, ordered by version number."""
    entries = memory_client.load_session_memory(
        session_id=session_id, user_id=""
    )
    for entry in reversed(entries):
        if entry.key == _VERSIONS_KEY and isinstance(entry.value, list):
            return sorted(entry.value, key=lambda v: v.get("version", 0))
    return []

def get_artifact_version(
    session_id: str, version_id: str, memory_client: MemoryClient
) -> dict | None:
    """Return a specific artifact version by ID."""
    versions = get_artifact_versions(session_id, memory_client)
    return next((v for v in versions if v["versionId"] == version_id), None)
The rollback flow in the API:
Python
# In FastAPI: POST /sessions/{sessionId}/rollback
@app.post("/sessions/{sessionId}/rollback")
async def rollback(sessionId: str, body: RollbackRequest) -> dict:
    orchestrator = get_orchestrator()
    result = orchestrator.handle_rollback(
        version_id=body.version_id,
        session_id=sessionId,
        user_id=session_data["userId"],
    )
    return result


GDPR: deleting user memory

The platform supports a "forget me" endpoint that deletes all memory for a user:
Python
def delete_user_memory(self, user_id: str) -> None:
    """Delete all memory entries for a user (GDPR compliance)."""
    # Delete from local store
    keys_to_delete = [
        k for k in self._local_store if k.startswith(f"{user_id}:")
    ]
    for k in keys_to_delete:
        del self._local_store[k]

    # Best-effort deletion from AgentCore Memory
    # (AgentCore Memory deletion API varies by version — check latest docs)
    logger.info("Deleted all memory for user=%s", user_id)
Python
# DELETE /users/{userId}/memory
@app.delete("/users/{userId}/memory", status_code=204)
async def delete_memory(userId: str) -> Response:
    orchestrator = get_orchestrator()
    orchestrator._memory.delete_user_memory(userId)
    return Response(status_code=204)


What's next

In  Part 6  we add Cedar-based policy enforcement — every tool call is evaluated against a policy bundle before execution, with hot-reload so policy changes take effect without restarting agents.

Series navigation ←  Part 4: MCP Tool Integration via AgentCore Gateway  →  Part 6: Policy as Code with Cedar 

# amazon-bedrock# amazon-bedrock-agentcore# semantic-search
Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.
Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article