AgentCore Stateful Runtime + Django: Building AI Agents That Remember Across Sessions

How AWS AgentCore’s managed memory and session persistence turns your Django-backed AI agents from goldfish into colleagues — without building a single memory system yourself.

The Amnesia Problem

Every LLM conversation starts from zero.

You ask your AI assistant to “use the same tone as last time.” It doesn’t know what last time was. You mention “the client we discussed on Tuesday.” It has no Tuesday. You say “continue where we left off.” There is no left-off.

The standard workaround is to stuff conversation history into the context window on every call. This works until it doesn’t — context windows fill up, token costs compound, and you’re still only solving within-session memory. The moment the user closes the browser tab, everything is gone.

What teams actually need is cross-session memory: an agent that accumulates knowledge about a user, task, or project over days and weeks. An agent that remembers a user’s preferences after ten conversations. An agent that recalls a debugging session from three days ago when a new related error surfaces.

AWS AgentCore, launched in 2025, includes a Stateful Runtime that provides exactly this — a managed, persistent memory and session layer for AI agents. Instead of building your own vector store, summary compressor, and session manager, you get a first-class API for storing, retrieving, and associating memories across time.

This post wires AgentCore’s Stateful Runtime into a Django application: persistent user preferences, cross-session conversation continuity, long-term task context, and a clean Django interface that makes stateful agents feel as natural as any other Django service.

What AgentCore Stateful Runtime Actually Is

Before the code, a clear mental model of what’s being managed:

Sessions are discrete interaction windows — a single chat conversation, a single task run, a single form wizard. AgentCore creates, manages, and expires sessions. Each session has an ID, a start time, and associated turn history.

Memory is agent-owned, user-associated persistent storage that outlives sessions. There are three memory types:

• Semantic memory — facts about the user or domain. “User prefers Python over JavaScript.” “This project uses PostgreSQL.” “Customer is on the Enterprise plan.”
• Episodic memory — what happened. “On session 2025–06–01, user asked about deployment and we resolved a Dockerfile issue.” “User attempted the onboarding flow twice and dropped off at step 3 both times.”
• Working memory — scoped to the current session. Cleared when the session ends. Used for intermediate reasoning state within a conversation.

Memory retrieval happens via semantic search over stored memories. When a new session starts, AgentCore retrieves the most relevant memories given the session’s initial context — so the agent walks into every conversation already briefed on what matters.

This is the architecture:

User Request
     │
     ▼
Django View
     │
     ├── AgentCore.create_session()        ← or resume existing
     │
     ├── AgentCore.retrieve_memories()     ← fetch relevant past context
     │
     ├── Bedrock LLM call                  ← agent reasoning with memory context
     │
     ├── AgentCore.store_memory()          ← persist anything worth remembering
     │
     └── AgentCore.end_turn()              ← save turn to session history
          │
          ▼
     Django Response

Project Setup

pip install boto3 django djangorestframework

AgentCore is accessed via the standard boto3 Bedrock Agent Runtime client. Add to settings.py:

# settings.py

AWS_REGION = "us-east-1"

AGENTCORE_CONFIG = {
    "memory_store_id": "your-memory-store-id",      # created in AWS console
    "agent_id": "your-agent-id",                    # your AgentCore agent
    "agent_alias_id": "TSTALIASID",                 # or production alias
    "session_ttl_minutes": 30,                      # idle session expiry
    "memory_retrieval_limit": 10,                   # max memories per retrieval
    "model_id": "us.anthropic.claude-3-7-sonnet-20250219-v1:0",
}

Directory layout:

myapp/
├── agentcore/
│   ├── __init__.py
│   ├── client.py          ← AgentCore boto3 wrapper
│   ├── memory.py          ← memory CRUD and retrieval
│   ├── session.py         ← session lifecycle management
│   └── agent.py           ← stateful agent: the main interface
├── models.py              ← Django session index + memory log
├── views.py
└── serializers.py

Step 1: The AgentCore Client

# myapp/agentcore/client.py

import boto3
from functools import lru_cache
from django.conf import settings

@lru_cache(maxsize=1)
def get_agentcore_client():
    """
    Cached boto3 client for AgentCore (Bedrock Agent Runtime).
    Uses a single client instance per Django process.
    """
    return boto3.client(
        "bedrock-agent-runtime",
        region_name=settings.AWS_REGION,
    )

@lru_cache(maxsize=1)
def get_bedrock_client():
    """Cached boto3 client for direct Bedrock model invocation."""
    return boto3.client(
        "bedrock-runtime",
        region_name=settings.AWS_REGION,
    )

Step 2: Memory Management

# myapp/agentcore/memory.py

from __future__ import annotations
import json
import uuid
import logging
from datetime import datetime
from typing import Literal
from django.conf import settings
from .client import get_agentcore_client

logger = logging.getLogger(__name__)

MemoryType = Literal["SEMANTIC", "EPISODIC", "WORKING"]

class MemoryManager:
    """
    Manages persistent memory for AI agents via AgentCore.

    Memory types:
    - SEMANTIC: facts about the user/domain that age slowly
    - EPISODIC: what happened during past sessions
    - WORKING: temporary state for the current session only
    """

    def __init__(self):
        self.config = settings.AGENTCORE_CONFIG
        self.memory_store_id = self.config["memory_store_id"]

    @property
    def client(self):
        return get_agentcore_client()

    # ── Store ──────────────────────────────────────────────────────────────

    def store(
        self,
        user_id: str,
        content: str,
        memory_type: MemoryType = "SEMANTIC",
        metadata: dict | None = None,
        session_id: str | None = None,
    ) -> str:
        """
        Store a memory for a user. Returns the memory ID.

        Examples:
            store(user_id, "Prefers concise bullet-point answers", "SEMANTIC")
            store(user_id, "Resolved Celery timeout issue on 2025-06-01", "EPISODIC")
        """
        memory_id = str(uuid.uuid4())
        payload = {
            "memoryStoreId": self.memory_store_id,
            "memoryId": memory_id,
            "content": content,
            "memoryType": memory_type,
            "userId": user_id,
        }
        if metadata:
            payload["metadata"] = {k: str(v) for k, v in metadata.items()}
        if session_id:
            payload["sessionId"] = session_id

        try:
            self.client.create_memory(**payload)
            logger.debug(f"Stored {memory_type} memory {memory_id} for user {user_id}")
            return memory_id
        except Exception as e:
            logger.error(f"Failed to store memory for {user_id}: {e}")
            raise

    def store_semantic(self, user_id: str, fact: str, **metadata) -> str:
        """Shortcut for storing a factual memory about a user."""
        return self.store(user_id, fact, "SEMANTIC", metadata or None)

    def store_episodic(
        self,
        user_id: str,
        what_happened: str,
        session_id: str,
        **metadata
    ) -> str:
        """Shortcut for storing what happened in a session."""
        meta = {"session_id": session_id, **metadata}
        return self.store(user_id, what_happened, "EPISODIC", meta, session_id)

    def store_working(self, user_id: str, state: str, session_id: str) -> str:
        """Shortcut for temporary within-session state."""
        return self.store(user_id, state, "WORKING", {"session_id": session_id}, session_id)

    # ── Retrieve ───────────────────────────────────────────────────────────

    def retrieve(
        self,
        user_id: str,
        query: str,
        memory_types: list[MemoryType] | None = None,
        limit: int | None = None,
    ) -> list[dict]:
        """
        Retrieve relevant memories for a user given a query.
        AgentCore uses semantic search over stored memories.

        Returns list of {"content": str, "type": str, "score": float, "id": str}
        """
        limit = limit or self.config["memory_retrieval_limit"]
        payload = {
            "memoryStoreId": self.memory_store_id,
            "userId": user_id,
            "searchQuery": query,
            "maxResults": limit,
        }
        if memory_types:
            payload["memoryTypes"] = memory_types

        try:
            response = self.client.retrieve_memories(**payload)
            memories = response.get("memoryItems", [])
            return [
                {
                    "id": m["memoryId"],
                    "content": m["content"],
                    "type": m["memoryType"],
                    "score": m.get("relevanceScore", 0.0),
                    "created_at": m.get("createdAt", ""),
                }
                for m in memories
            ]
        except Exception as e:
            logger.error(f"Memory retrieval failed for {user_id}: {e}")
            return []

    def retrieve_for_session_start(self, user_id: str, initial_message: str) -> list[dict]:
        """
        Retrieve memories relevant to the start of a new session.
        Called before the first LLM turn to brief the agent.
        Returns both semantic facts and recent episodic memories.
        """
        semantic = self.retrieve(user_id, initial_message, ["SEMANTIC"], limit=6)
        episodic = self.retrieve(user_id, initial_message, ["EPISODIC"], limit=4)
        return semantic + episodic

    # ── Delete ─────────────────────────────────────────────────────────────

    def delete(self, user_id: str, memory_id: str) -> None:
        """Delete a specific memory (e.g. if user requests data erasure)."""
        try:
            self.client.delete_memory(
                memoryStoreId=self.memory_store_id,
                userId=user_id,
                memoryId=memory_id,
            )
        except Exception as e:
            logger.error(f"Failed to delete memory {memory_id}: {e}")
            raise

    def delete_all_user_memories(self, user_id: str) -> int:
        """
        GDPR/privacy: delete all memories for a user.
        Returns count of deleted memories.
        """
        all_memories = self.retrieve(user_id, "", limit=1000)
        deleted = 0
        for m in all_memories:
            try:
                self.delete(user_id, m["id"])
                deleted += 1
            except Exception:
                pass
        logger.info(f"Deleted {deleted} memories for user {user_id}")
        return deleted

Step 3: Session Management

# myapp/agentcore/session.py

from __future__ import annotations
import uuid
import logging
from datetime import datetime
from django.conf import settings
from django.utils import timezone
from .client import get_agentcore_client
from myapp.models import AgentSession

logger = logging.getLogger(__name__)

class SessionManager:
    """
    Manages AgentCore session lifecycle:
    create → use → end → expire.
    Mirrors sessions in a Django model for queryability.
    """

    def __init__(self):
        self.config = settings.AGENTCORE_CONFIG

    @property
    def client(self):
        return get_agentcore_client()

    # ── Create ─────────────────────────────────────────────────────────────

    def create_session(
        self,
        user_id: str,
        initial_context: str = "",
        channel: str = "web",
    ) -> "AgentSession":
        """
        Create a new session in AgentCore and record it in Django.
        Returns the Django AgentSession model instance.
        """
        session_id = str(uuid.uuid4())

        try:
            self.client.create_session(
                agentId=self.config["agent_id"],
                agentAliasId=self.config["agent_alias_id"],
                sessionId=session_id,
                sessionAttributes={
                    "userId": user_id,
                    "channel": channel,
                    "created_at": datetime.utcnow().isoformat(),
                },
            )
        except Exception as e:
            logger.error(f"AgentCore session creation failed: {e}")
            raise

        session = AgentSession.objects.create(
            session_id=session_id,
            user_id=user_id,
            channel=channel,
            initial_context=initial_context[:1000],
            status="active",
        )
        logger.debug(f"Created session {session_id} for user {user_id}")
        return session

    # ── Resume ─────────────────────────────────────────────────────────────

    def get_or_create_session(
        self,
        user_id: str,
        session_id: str | None = None,
        initial_context: str = "",
    ) -> tuple["AgentSession", bool]:
        """
        Resume an existing session or create a new one.
        Returns (session, created: bool).

        Automatically creates a new session if:
        - No session_id provided
        - Session not found
        - Session has expired
        """
        if session_id:
            try:
                session = AgentSession.objects.get(
                    session_id=session_id,
                    user_id=user_id,
                    status="active",
                )
                # Check idle expiry
                ttl = self.config["session_ttl_minutes"]
                idle_minutes = (timezone.now() - session.last_activity).seconds // 60
                if idle_minutes > ttl:
                    self.end_session(session)
                    # fall through to create new
                else:
                    return session, False
            except AgentSession.DoesNotExist:
                pass

        new_session = self.create_session(
            user_id=user_id,
            initial_context=initial_context,
        )
        return new_session, True

    # ── End ────────────────────────────────────────────────────────────────

    def end_session(self, session: "AgentSession") -> None:
        """
        Mark session as ended in Django and in AgentCore.
        Triggers episodic memory creation for the session.
        """
        try:
            self.client.end_session(
                agentId=self.config["agent_id"],
                agentAliasId=self.config["agent_alias_id"],
                sessionId=session.session_id,
            )
        except Exception as e:
            logger.warning(f"AgentCore end_session failed for {session.session_id}: {e}")

        session.status = "ended"
        session.ended_at = timezone.now()
        session.save(update_fields=["status", "ended_at"])
        logger.debug(f"Ended session {session.session_id}")

    # ── Touch ──────────────────────────────────────────────────────────────

    def touch_session(self, session: "AgentSession") -> None:
        """Update last_activity timestamp to prevent idle expiry."""
        AgentSession.objects.filter(pk=session.pk).update(
            last_activity=timezone.now(),
            turn_count=session.turn_count + 1,
        )

Step 4: The Stateful Agent

This is where everything comes together. The StatefulAgent manages the full loop: retrieve memories, call the LLM with memory context, extract new memories from the response, store them.

# myapp/agentcore/agent.py

from __future__ import annotations
import json
import re
import logging
import time
from dataclasses import dataclass, field
from typing import Generator
from django.conf import settings
from .memory import MemoryManager
from .session import SessionManager
from .client import get_bedrock_client
from myapp.models import AgentSession, TurnLog

logger = logging.getLogger(__name__)

@dataclass
class AgentResponse:
    """Full response from a stateful agent turn."""
    text: str
    session_id: str
    turn_number: int
    memories_retrieved: list[dict] = field(default_factory=list)
    memories_stored: list[str] = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    latency_ms: int = 0
    new_session: bool = False

# ── Memory extraction prompt ─────────────────────────────────────────────────
# Appended to every agent system prompt to enable self-directed memory storage

MEMORY_EXTRACTION_SUFFIX = """

## Memory Instructions
At the end of your response, if this conversation revealed anything worth
remembering for future sessions, output a memory block in this exact format:

<memories>
SEMANTIC: User prefers TypeScript over JavaScript for all new projects.
EPISODIC: User resolved a Redis connection error by updating REDIS_URL to use rediss://.
SEMANTIC: User is building a multi-tenant SaaS on Django with row-level security.
</memories>

Only include memories that are:
- Durable facts about the user, their project, or their preferences (SEMANTIC)
- Significant events or resolutions worth recalling in future sessions (EPISODIC)

Omit the <memories> block entirely if nothing is worth storing.
Do not include trivial, transient, or session-specific information.
"""

class StatefulAgent:
    """
    A Django-integrated AI agent with cross-session memory.

    Usage:
        agent = StatefulAgent()
        response = agent.chat(
            user_id="user_42",
            message="How should I structure my Celery tasks?",
            session_id=request.session.get("agent_session_id"),
        )
        request.session["agent_session_id"] = response.session_id
    """

    def __init__(self, system_prompt: str = ""):
        self.config = settings.AGENTCORE_CONFIG
        self.memory = MemoryManager()
        self.sessions = SessionManager()
        self.base_system_prompt = system_prompt or self._default_system_prompt()

    def _default_system_prompt(self) -> str:
        return (
            "You are a helpful AI assistant with persistent memory. "
            "You remember details from previous conversations and use them "
            "to provide personalized, context-aware assistance."
        )

    @property
    def bedrock(self):
        return get_bedrock_client()

    # ── Memory context builder ────────────────────────────────────────────

    def _build_memory_context(self, memories: list[dict]) -> str:
        """Format retrieved memories into a clear context block for the LLM."""
        if not memories:
            return ""

        semantic = [m for m in memories if m["type"] == "SEMANTIC"]
        episodic = [m for m in memories if m["type"] == "EPISODIC"]

        lines = ["## What I remember about you\n"]

        if semantic:
            lines.append("**Preferences and facts:**")
            for m in semantic:
                lines.append(f"- {m['content']}")

        if episodic:
            lines.append("\n**Past sessions:**")
            for m in episodic:
                lines.append(f"- {m['content']}")

        return "\n".join(lines)

    # ── Memory extraction from response ───────────────────────────────────

    def _extract_and_store_memories(
        self,
        response_text: str,
        user_id: str,
        session_id: str,
    ) -> tuple[str, list[str]]:
        """
        Parse <memories> block from the LLM response.
        Store each memory and return (clean_response, memory_ids).
        """
        memory_ids = []
        clean_text = response_text

        match = re.search(r"<memories>(.*?)</memories>", response_text, re.DOTALL)
        if not match:
            return clean_text, memory_ids

        # Strip the memory block from the user-facing response
        clean_text = response_text[:match.start()].strip()

        memory_block = match.group(1).strip()
        for line in memory_block.splitlines():
            line = line.strip()
            if not line:
                continue
            if line.startswith("SEMANTIC:"):
                content = line[len("SEMANTIC:"):].strip()
                mid = self.memory.store_semantic(user_id, content)
                memory_ids.append(mid)
                logger.debug(f"Stored semantic memory: {content[:60]}...")
            elif line.startswith("EPISODIC:"):
                content = line[len("EPISODIC:"):].strip()
                mid = self.memory.store_episodic(user_id, content, session_id)
                memory_ids.append(mid)
                logger.debug(f"Stored episodic memory: {content[:60]}...")

        return clean_text, memory_ids

    # ── LLM call ──────────────────────────────────────────────────────────

    def _invoke_model(
        self,
        messages: list[dict],
        system_prompt: str,
        max_tokens: int = 2048,
    ) -> tuple[str, int, int]:
        """
        Call Bedrock with the full conversation + memory context.
        Returns (response_text, input_tokens, output_tokens).
        """
        payload = {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": max_tokens,
            "system": system_prompt,
            "messages": messages,
        }
        response = self.bedrock.invoke_model(
            modelId=self.config["model_id"],
            body=json.dumps(payload),
        )
        body = json.loads(response["body"].read())
        text = body["content"][0]["text"]
        usage = body.get("usage", {})
        return text, usage.get("input_tokens", 0), usage.get("output_tokens", 0)

    # ── Main chat interface ────────────────────────────────────────────────

    def chat(
        self,
        user_id: str,
        message: str,
        session_id: str | None = None,
        conversation_history: list[dict] | None = None,
        max_tokens: int = 2048,
    ) -> AgentResponse:
        """
        Single-turn chat with full memory lifecycle.

        - Retrieves relevant memories from past sessions
        - Injects memory context into the system prompt
        - Calls the LLM
        - Extracts and persists new memories from the response
        - Logs the turn

        Returns AgentResponse with clean text + full metadata.
        """
        start = time.monotonic()

        # 1. Session management
        session, is_new = self.sessions.get_or_create_session(
            user_id=user_id,
            session_id=session_id,
            initial_context=message,
        )

        # 2. Retrieve relevant memories
        memories = self.memory.retrieve_for_session_start(user_id, message)

        # 3. Build system prompt with memory context
        memory_context = self._build_memory_context(memories)
        full_system_prompt = "\n\n".join(filter(None, [
            self.base_system_prompt,
            memory_context,
            MEMORY_EXTRACTION_SUFFIX,
        ]))

        # 4. Build message list (history + current turn)
        messages = list(conversation_history or [])
        messages.append({"role": "user", "content": message})

        # 5. Call the model
        raw_response, input_tokens, output_tokens = self._invoke_model(
            messages=messages,
            system_prompt=full_system_prompt,
            max_tokens=max_tokens,
        )

        # 6. Extract memories from response and get clean text
        clean_response, stored_ids = self._extract_and_store_memories(
            raw_response, user_id, session.session_id
        )

        # 7. Update session activity
        self.sessions.touch_session(session)

        latency_ms = int((time.monotonic() - start) * 1000)
        turn_number = session.turn_count + 1

        # 8. Log the turn
        TurnLog.objects.create(
            session_id=session.session_id,
            user_id=user_id,
            turn_number=turn_number,
            user_message=message[:2000],
            agent_response=clean_response[:4000],
            memories_retrieved=len(memories),
            memories_stored=len(stored_ids),
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            latency_ms=latency_ms,
        )

        return AgentResponse(
            text=clean_response,
            session_id=session.session_id,
            turn_number=turn_number,
            memories_retrieved=memories,
            memories_stored=stored_ids,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            latency_ms=latency_ms,
            new_session=is_new,
        )

Step 5: Django Models

# myapp/models.py

from django.db import models
from django.contrib.auth import get_user_model

User = get_user_model()

class AgentSession(models.Model):
    STATUS_CHOICES = [
        ("active", "Active"),
        ("ended", "Ended"),
        ("expired", "Expired"),
    ]
    CHANNEL_CHOICES = [
        ("web", "Web"),
        ("api", "API"),
        ("mobile", "Mobile"),
        ("slack", "Slack"),
    ]

    session_id = models.CharField(max_length=100, unique=True, db_index=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="agent_sessions")
    channel = models.CharField(max_length=20, choices=CHANNEL_CHOICES, default="web")
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="active")
    initial_context = models.TextField(blank=True)
    turn_count = models.IntegerField(default=0)
    last_activity = models.DateTimeField(auto_now=True)
    started_at = models.DateTimeField(auto_now_add=True)
    ended_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        ordering = ["-started_at"]
        indexes = [
            models.Index(fields=["user", "status", "last_activity"]),
        ]

    def __str__(self):
        return f"Session {self.session_id[:8]}… ({self.user} / {self.status})"

class TurnLog(models.Model):
    """Individual turn record within a session."""
    session_id = models.CharField(max_length=100, db_index=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="turn_logs")
    turn_number = models.IntegerField()
    user_message = models.TextField()
    agent_response = models.TextField()
    memories_retrieved = models.IntegerField(default=0)
    memories_stored = models.IntegerField(default=0)
    input_tokens = models.IntegerField(default=0)
    output_tokens = models.IntegerField(default=0)
    latency_ms = models.IntegerField(null=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ["session_id", "turn_number"]
        indexes = [
            models.Index(fields=["session_id", "turn_number"]),
            models.Index(fields=["user", "created_at"]),
        ]

class MemoryAuditLog(models.Model):
    """Tracks what was stored and when, for debugging and privacy compliance."""
    MEMORY_TYPE_CHOICES = [
        ("SEMANTIC", "Semantic"),
        ("EPISODIC", "Episodic"),
        ("WORKING", "Working"),
    ]
    memory_id = models.CharField(max_length=100, unique=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="memory_logs")
    memory_type = models.CharField(max_length=20, choices=MEMORY_TYPE_CHOICES)
    content_preview = models.CharField(max_length=300)
    session_id = models.CharField(max_length=100, blank=True)
    deleted = models.BooleanField(default=False)
    deleted_at = models.DateTimeField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ["-created_at"]
        indexes = [
            models.Index(fields=["user", "memory_type"]),
            models.Index(fields=["user", "deleted"]),
        ]

Step 6: Django Views

# myapp/views.py

import json
import logging
from django.http import JsonResponse
from django.contrib.auth.decorators import login_required
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST, require_GET
from .agentcore.agent import StatefulAgent
from .agentcore.memory import MemoryManager
from .models import AgentSession, TurnLog

logger = logging.getLogger(__name__)

# Module-level agent instance — one per Django process
agent = StatefulAgent(
    system_prompt=(
        "You are a knowledgeable engineering assistant for a SaaS development team. "
        "You help with code reviews, architecture decisions, debugging, and best practices. "
        "You have persistent memory of past conversations and use it to give context-aware advice."
    )
)
memory_manager = MemoryManager()

@login_required
@csrf_exempt
@require_POST
def chat(request):
    """
    Main stateful chat endpoint.
    Manages session continuity via a session_id stored in Django's session framework.
    """
    try:
        payload = json.loads(request.body)
        message = payload.get("message", "").strip()
        if not message:
            return JsonResponse({"error": "message is required"}, status=400)
    except json.JSONDecodeError:
        return JsonResponse({"error": "Invalid JSON"}, status=400)

    # Retrieve conversation history for within-session continuity
    # (stored in Django session, separate from AgentCore cross-session memory)
    conversation_history = request.session.get("conversation_history", [])

    # Retrieve persisted AgentCore session ID if one exists
    agentcore_session_id = request.session.get("agentcore_session_id")

    try:
        response = agent.chat(
            user_id=str(request.user.id),
            message=message,
            session_id=agentcore_session_id,
            conversation_history=conversation_history,
        )
    except Exception as e:
        logger.exception(f"Agent chat failed for user {request.user.id}")
        return JsonResponse({"error": "Agent error", "detail": str(e)}, status=500)

    # Persist session ID and update history for next turn
    request.session["agentcore_session_id"] = response.session_id
    conversation_history.append({"role": "user", "content": message})
    conversation_history.append({"role": "assistant", "content": response.text})
    # Keep last 20 turns in session to manage context window
    request.session["conversation_history"] = conversation_history[-20:]

    return JsonResponse({
        "reply": response.text,
        "session_id": response.session_id,
        "new_session": response.new_session,
        "turn": response.turn_number,
        "memories_active": len(response.memories_retrieved),
        "memories_learned": len(response.memories_stored),
        "meta": {
            "input_tokens": response.input_tokens,
            "output_tokens": response.output_tokens,
            "latency_ms": response.latency_ms,
        }
    })

@login_required
def session_history(request):
    """Return the user's session history with turn counts."""
    sessions = AgentSession.objects.filter(
        user=request.user
    ).values(
        "session_id", "status", "turn_count",
        "channel", "started_at", "ended_at", "last_activity"
    )[:20]
    return JsonResponse({"sessions": list(sessions)})

@login_required
def turn_history(request, session_id):
    """Return all turns for a specific session."""
    turns = TurnLog.objects.filter(
        session_id=session_id,
        user=request.user,
    ).values(
        "turn_number", "user_message", "agent_response",
        "memories_retrieved", "memories_stored", "latency_ms", "created_at"
    )
    return JsonResponse({"turns": list(turns)})

@login_required
def memory_list(request):
    """Return all stored memories for the current user."""
    memories = memory_manager.retrieve(
        user_id=str(request.user.id),
        query="",        # empty query returns all
        limit=100,
    )
    return JsonResponse({"memories": memories, "count": len(memories)})

@login_required
@csrf_exempt
@require_POST
def memory_delete(request, memory_id):
    """GDPR/privacy: delete a specific memory."""
    try:
        memory_manager.delete(
            user_id=str(request.user.id),
            memory_id=memory_id,
        )
        return JsonResponse({"deleted": memory_id})
    except Exception as e:
        return JsonResponse({"error": str(e)}, status=500)

@login_required
@csrf_exempt
@require_POST
def memory_delete_all(request):
    """GDPR/privacy: delete all memories for the current user."""
    count = memory_manager.delete_all_user_memories(str(request.user.id))
    # Also clear the active session
    request.session.pop("agentcore_session_id", None)
    request.session.pop("conversation_history", None)
    return JsonResponse({"deleted_count": count})

@login_required
@csrf_exempt
@require_POST
def end_session(request):
    """Explicitly end the current session and trigger episodic memory creation."""
    session_id = request.session.get("agentcore_session_id")
    if not session_id:
        return JsonResponse({"status": "no active session"})

    try:
        session = AgentSession.objects.get(
            session_id=session_id, user=request.user
        )
        from .agentcore.session import SessionManager
        sm = SessionManager()
        sm.end_session(session)

        # Store a high-level episodic summary of this session
        turn_count = session.turn_count
        memory_manager.store_episodic(
            user_id=str(request.user.id),
            what_happened=f"Session on {session.started_at.strftime('%Y-%m-%d')} "
                          f"({turn_count} turns via {session.channel}). "
                          f"Context: {session.initial_context[:200]}",
            session_id=session_id,
        )

        request.session.pop("agentcore_session_id", None)
        request.session.pop("conversation_history", None)

        return JsonResponse({"ended": session_id, "turns": turn_count})
    except AgentSession.DoesNotExist:
        return JsonResponse({"error": "Session not found"}, status=404)

Step 7: URL Configuration

# myapp/urls.py

from django.urls import path
from . import views

urlpatterns = [
    # Chat
    path("api/agent/chat/", views.chat, name="agent-chat"),
    path("api/agent/session/end/", views.end_session, name="agent-end-session"),

    # History
    path("api/agent/sessions/", views.session_history, name="agent-sessions"),
    path("api/agent/sessions/<str:session_id>/turns/", views.turn_history, name="agent-turns"),

    # Memory management
    path("api/agent/memories/", views.memory_list, name="agent-memories"),
    path("api/agent/memories/<str:memory_id>/delete/", views.memory_delete, name="memory-delete"),
    path("api/agent/memories/delete-all/", views.memory_delete_all, name="memory-delete-all"),
]

How Memory Flows: A Walk-Through

Here’s what happens across three separate sessions for a single user:

Session 1 — Monday

User: “I’m building a multi-tenant SaaS on Django. Each customer needs complete data isolation.”

The agent responds with row-level security approaches. At the end of its response, it includes:

<memories>
SEMANTIC: User is building a multi-tenant Django SaaS requiring strict per-customer data isolation.
SEMANTIC: User is evaluating row-level security strategies for multi-tenancy.
</memories>

Both memories are stored. Session ends.

Session 2 — Wednesday (new browser tab, fresh HTTP session)

User: “What’s the safest way to run database migrations in this setup?”

Before calling the LLM, retrieve_for_session_start() fetches the Monday memories. The system prompt now includes:

## What I remember about you

**Preferences and facts:**
- User is building a multi-tenant Django SaaS requiring strict per-customer data isolation.
- User is evaluating row-level security strategies for multi-tenancy.

The agent responds with migration advice specifically scoped to multi-tenant row-level security — without the user mentioning any of that context. The user experiences the agent as genuinely knowledgeable about their project.

Session 3 — Friday

User: “We’re ready to go live. What should I check first?”

Episodic memories from sessions 1 and 2 are retrieved. The agent produces a launch checklist that references the specific architectural decisions from those sessions. It feels like consulting a teammate who was there, not re-explaining everything to a new tool.

Handling Memory Quality

Not every fact is worth storing. The MEMORY_EXTRACTION_SUFFIX instructs the model to be selective. But you can go further by adding quality filters before storing:

# myapp/agentcore/memory.py — add to MemoryManager

MIN_MEMORY_LENGTH = 20
MAX_MEMORY_LENGTH = 500
NOISE_PHRASES = [
    "the user said", "user mentioned", "user asked",
    "you can", "you should", "let me know",
]

def is_quality_memory(self, content: str) -> bool:
    """Basic quality gate before storing a memory."""
    if len(content) < self.MIN_MEMORY_LENGTH:
        return False
    if len(content) > self.MAX_MEMORY_LENGTH:
        return False
    content_lower = content.lower()
    if any(phrase in content_lower for phrase in self.NOISE_PHRASES):
        return False
    return True

Apply it in store():

def store(self, user_id, content, memory_type="SEMANTIC", ...):
    if not self.is_quality_memory(content):
        logger.debug(f"Rejected low-quality memory: {content[:60]}")
        return None
    # ... rest of store logic

Cross-Session Conversation Continuity vs. Memory: The Distinction

There are two different things in play here and it’s important not to conflate them:

Within-session continuity is handled by passing conversation_history — the list of message dicts — in every Bedrock call. This is the standard "give the model the full transcript" approach. It lives in Django's session framework (or Redis/database-backed sessions) and expires when the session ends.

Cross-session memory is handled by AgentCore. It’s not transcripts — it’s distilled knowledge that persists indefinitely. The agent learns that the user prefers TypeScript. It doesn’t store every message about TypeScript forever.

The two layers complement each other. Within-session history provides exact turn-by-turn context. Cross-session memory provides the “who is this user and what do I know about them” briefing at session start.

Production Considerations

Memory Staleness

Semantic memories can become outdated. A user who preferred React in February might have switched to Vue by May. Add a created_at filter to deprioritize old memories:

def retrieve_for_session_start(self, user_id, initial_message):
    recent = self.retrieve(user_id, initial_message, ["SEMANTIC"], limit=6)
    # For episodic, only last 90 days are relevant
    episodic = self.retrieve(
        user_id, initial_message, ["EPISODIC"], limit=4
    )
    # Filter episodic to last 90 days in post-processing
    from datetime import datetime, timedelta
    cutoff = (datetime.utcnow() - timedelta(days=90)).isoformat()
    episodic = [m for m in episodic if m.get("created_at", "") >= cutoff]
    return recent + episodic

Memory Store Costs

AgentCore’s memory store charges per stored memory and per retrieval. Keep costs predictable:

• Cap semantic memories per user (e.g., 200 max — delete oldest when full)
• Limit episodic memories to last 60 days
• Gate episodic storage to sessions with 3+ turns (trivial sessions rarely produce useful episodic memories)

Privacy and Data Residency

AgentCore stores memories in AWS. For GDPR compliance:

• Implement delete_all_user_memories() and wire it to your account deletion flow
• Expose the /api/agent/memories/ endpoint so users can see exactly what's stored
• Log all memory storage events in MemoryAuditLog for audit trails
• Set explicit memory retention policies in the AgentCore console

What Stateful Memory Unlocks

The functional difference between a stateless and stateful agent is not subtle. Here’s what becomes possible once memory is in place:

Personalization without onboarding. The agent learns a user’s stack, preferences, and communication style through conversation. No questionnaire required.

Continuity across interruptions. A multi-day debugging session picks up where it left off even after the user closes the app and comes back tomorrow.

Progressively smarter responses. The 50th conversation with a user is materially better than the first, because the agent has 49 sessions of context distilled into memories.

Proactive context application. The agent can reference a past issue without being prompted — “You mentioned migrating to a new database setup last week — does that affect this?” — because episodic memory surfaces relevant history automatically.

These are the features that make users feel like they have a dedicated assistant rather than a generic chatbot. And with AgentCore’s Stateful Runtime, the memory infrastructure is managed. You’re not building a vector store, a summary compressor, or a session reconciliation system. You’re writing Django application code.

Conclusion

AgentCore’s Stateful Runtime removes the hardest part of building memory-capable agents: the infrastructure. The memory store, semantic retrieval, session lifecycle, and cross-session persistence are all managed services. What remains is wiring them into your Django application — which, as this post shows, is a few hundred lines of clean Python.

The pattern laid out here — retrieve memories at session start, inject as system prompt context, extract new memories from responses, persist via AgentCore — is a complete, production-ready approach. Add the Django models for auditability, the GDPR deletion endpoints for compliance, and the quality filters for memory hygiene, and you have an agent that genuinely improves with every conversation.

That’s the bar users have started expecting from AI tools. AgentCore makes it achievable without a dedicated ML infrastructure team.

Resources

• AWS AgentCore documentation
• Amazon Bedrock Agent Runtime API
• AWS Strands Agents SDK — pairs naturally with AgentCore memory
• Django session framework
• PyO3 for performance-critical Django extensions

If this shaped how you’re thinking about stateful agents, leave a comment or clap. Questions about scaling to millions of users? The memory store cost model gets interesting at that scale — happy to dig into it.