AgentCore Memory

ACTOR_CLAIM = "username"
"""Cognito puts the signed-in user's name in this claim of its access tokens (M2M client tokens have none)."""


def actor_from(context: RequestContext) -> str:
    """The signed-in user, from the verified token: never trust caller-supplied identity (a header anyone can set).

    Raises:
        PermissionError: The token names no user (e.g. a machine-to-machine client token).
    """
    actor = token_claims(context).get(ACTOR_CLAIM)
    if not actor:
        raise PermissionError(f"the bearer token has no '{ACTOR_CLAIM}' claim: sign in as a user")
    return str(actor)


def token_claims(context: RequestContext) -> dict:
    """The claims of the request's bearer token ({} when there is none).

    Decoding without checking the signature is safe only because the runtime's JWT authorizer already verified the
    signature, issuer, expiry and client before the request reached this code.
    """
    headers = {name.lower(): value for name, value in (context.request_headers or {}).items()}
    token = headers.get("authorization", "").split(" ")[-1]
    try:
        payload = token.split(".")[1]
        return json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
    except (IndexError, ValueError):
        return {}

invoke --session S1 --> agent --CreateEvent--> SHORT-TERM: events, per actor + session (raw turns, kept 30 days)
                                                   |  strategies extract in the background (~1-2 min)
                                                   v
                         LONG-TERM: records, per strategy namespace
                           /strategies/<UserFacts>/actors/alice-chen               "Alice owns the Q3 report"
                           /strategies/<UserPreferences>/actors/alice-chen         "prefers answers in French"
                           /strategies/<SessionSummaries>/actors/alice-chen/sessions/S1   "<topic>Headcount ...</topic>"
                                                   |
invoke (new session S2) --RetrieveMemoryRecords(query = your question)--> <user_context> before the model call
invoke --session last  --ListEvents--> the conversation so far (short-term), no extraction needed

MEMORY_RELEVANCE_SCORE = 0.3
"""Minimum relevance a long-term fact or past-session summary needs to be retrieved into the conversation."""


ALWAYS = 0.0
"""No relevance threshold: preferences (language, format, units) apply to every question, whatever it is."""


def long_term_retrieval() -> dict[str, RetrievalConfig]:
    """Which long-term records to search before each model call: one namespace prefix per strategy, per actor.

    The session manager fills in {actorId}; a namespace is a path prefix, so the summary entry covers every past
    session of this actor (`.../sessions/<id>`).
    """
    return {
        f"/strategies/{MEMORY_STRATEGY_ID}/actors/{{actorId}}": RetrievalConfig(
            top_k=5, relevance_score=MEMORY_RELEVANCE_SCORE
        ),
        f"/strategies/{MEMORY_PREFERENCE_STRATEGY_ID}/actors/{{actorId}}": RetrievalConfig(
            top_k=5, relevance_score=ALWAYS
        ),
        f"/strategies/{MEMORY_SUMMARY_STRATEGY_ID}/actors/{{actorId}}": RetrievalConfig(
            top_k=2, relevance_score=MEMORY_RELEVANCE_SCORE
        ),
    }


def memory_session_manager(session_id: str, actor_id: str) -> AgentCoreMemorySessionManager:
    """Short-term: this session's turns (events). Long-term: this actor's facts, preferences and session summaries."""
    config = AgentCoreMemoryConfig(
        memory_id=MEMORY_ID, session_id=session_id, actor_id=actor_id, retrieval_config=long_term_retrieval()
    )
    return AgentCoreMemorySessionManager(agentcore_memory_config=config, region_name=REGION)

uv run bootcamp.py test --only 3

[CHALLENGE] SKIP stage 3 memories in the reply: the reply has no memories field yet
# after your change:
[CHALLENGE] PASS stage 3 memories in the reply: signed in as bootcamp-check-8560e352: actor_id=bootcamp-check-8560e352 memories=[]