Iris Development

Iris: Redis Agent Memory

Iris is the umbrella brand for Redis's AI-focused products. This skill currently covers one product in that family: Redis Agent Memory (RAM) — the persistent memory layer for AI agents, delivered as a managed service on Redis Cloud. Additional Iris products will be added as separate sections when they ship.

Redis Agent Memory exposes a REST/JSON data-plane API with two memory tiers:

• Session memory — append-only conversation history per session (working memory).
• Long-term memory — semantically searchable records extracted from sessions (or created directly).

A background promotion worker — managed by Redis Cloud — extracts durable facts from session events and writes them into long-term memory.

Official SDKs

All code samples use the official SDKs:

Both SDKs read the bearer token from AGENT_MEMORY_API_KEY and the default store ID from AGENT_MEMORY_STORE_ID. The production data-plane URL is https://gcp-us-east4.memory.redis.io; the exact URL for your service is also shown in the Cloud console after provisioning.

When to Apply

Reference these guidelines when:

• Creating a memory service on Redis Cloud (https://cloud.redis.io/#/agent-memory)
• Wiring an agent to call AgentMemory.add_session_event(...) / addSessionEvent(...)
• Searching long-term memory with search_long_term_memory(...) / searchLongTermMemory(...)
• Choosing between session events and direct long-term memory writes

Rule Categories by Priority

Quick Reference

1. Setup & Cloud Service (HIGH)

• `setup-cloud-service` - Create a Memory service on Redis Cloud
• `setup-auth-token` - Authenticate the SDK with a store API key

2. Session Memory / Events (HIGH)

• `session-when-to-use` - Choose session events vs direct long-term memory
• `session-add-event` - Append a session event correctly
• `session-retrieval` - Retrieve session memory and individual events

3. Long-Term Memory (HIGH)

• `ltm-bulk-create` - Create long-term memories in bulk with idempotent IDs
• `ltm-search` - Search long-term memory semantically with filters
• `ltm-organize` - Organize records with namespace, ownerId, topics, and memoryType

4. Memory Promotion (MEDIUM)

• `promotion-overview` - How background promotion works

How to Use

Read individual rule files under references/ for detailed explanations and code examples:

references/setup-cloud-service.md
references/session-add-event.md
references/promotion-overview.md

Each rule file contains:

• Brief explanation of why it matters
• Correct example(s) with Python and TypeScript SDK code
• Either an "Incorrect" example or "When to use / When NOT needed" guidance
• Additional context and references

Create Long-Term Memories in Bulk with Idempotent IDs

bulk_create_long_term_memories (Python) / bulkCreateLongTermMemories (TypeScript) accepts up to 100 records per call. The client supplies the id for each record so a retry never creates a duplicate. The response splits into created (IDs that landed) and errors (per-ID failures).

Correct: Generate a deterministic ID per logical fact and batch up to 100.

Python:

import uuid
from redis_agent_memory import AgentMemory, models

def upsert_facts(agent_memory: AgentMemory, facts: list[dict]):
    # Cap at 100 per call — the API enforces this.
    res = agent_memory.bulk_create_long_term_memories(memories=[
        {
            "id":          fact["id"],                           # stable, deterministic
            "text":        fact["text"],                         # 1–50000 chars
            "memory_type": fact.get("memory_type", models.MemoryType.SEMANTIC),
            "session_id":  fact.get("session_id"),
            "owner_id":    fact.get("owner_id"),
            "namespace":   fact.get("namespace"),
            "topics":      fact.get("topics", []),
        }
        for fact in facts[:100]
    ])
    # res.created = [...ids...], res.errors = [BulkOperationError(...)]
    return res

# Deterministic IDs make retries safe: same fact → same id → no duplicate.
facts = [{
    "id":         f"user-42-pref-{uuid.uuid5(uuid.NAMESPACE_OID, 'theme:dark')}",
    "text":       "User 42 prefers dark mode.",
    "owner_id":   "user-42",
    "topics":     ["profile", "ui-preferences"],
}]
upsert_facts(agent_memory, facts)

TypeScript:

import { AgentMemory } from "@redis-iris/agent-memory";

async function upsertFacts(
  agentMemory: AgentMemory,
  facts: Array<{
    id: string; text: string;
    memoryType?: "semantic" | "episodic" | "message";
    sessionId?: string; ownerId?: string; namespace?: string;
    topics?: string[];
  }>,
) {
  const res = await agentMemory.bulkCreateLongTermMemories({
    memories: facts.slice(0, 100).map((f) => ({
      id:         f.id,
      text:       f.text,
      memoryType: f.memoryType ?? "semantic",
      sessionId:  f.sessionId,
      ownerId:    f.ownerId,
      namespace:  f.namespace,
      topics:     f.topics ?? [],
    })),
  });
  // res.created: string[], res.errors?: Array<{id: string; error: string}>
  return res;
}

Incorrect: One call per memory, or random IDs on every retry.

# Bad: N round-trips + N embedding calls — slow and hammers your rate limit.
for fact in facts:
    agent_memory.bulk_create_long_term_memories(memories=[{
        "id":   str(uuid.uuid4()),            # <-- new id on every retry → duplicates on transient failures
        "text": fact["text"],
    }])

Partial-success contract — always inspect `errors`:

res = upsert_facts(agent_memory, facts)
if res.errors:
    for err in res.errors:
        log.warning("LTM create failed", id=err.id, reason=err.error)
    # res.created IS persisted; do not retry those.
    failed_ids = {e.id for e in res.errors}
    retry_later([f for f in facts if f["id"] in failed_ids])

const res = await upsertFacts(agentMemory, facts);
if (res.errors?.length) {
  for (const err of res.errors) {
    console.warn("LTM create failed", err.id, err.error);
  }
  const failedIds = new Set(res.errors.map((e) => e.id));
  await retryLater(facts.filter((f) => failedIds.has(f.id)));
}

Constraints:

• memories: 1–100 items per call.
• id: 1–64 chars, [a-zA-Z0-9-].
• text: 1–50000 chars.
• memory_type / memoryType: semantic | episodic | message.
• topics: up to 50, each 1–100 chars.
• TTL: defaults to 1 year (31_536_000 seconds) unless the store's long-term-memory TTL overrides it.

To update a record's text or tags later, use update_long_term_memory(memory_id=...) / updateLongTermMemory(memoryId, ...) rather than re-creating with the same ID.

Organize Long-Term Memory with namespace, ownerId, topics, and memoryType

LTM records carry four structured fields that exist purely to scope search. They cost nothing extra to populate at write time and make every later search call faster and more precise.

`memory_type` semantics:

• semantic — a durable fact ("user prefers dark mode"). Cheapest to keep around long-term; survives across sessions.
• episodic — something that happened at a point in time ("user asked about pricing on 2026-05-10"). Pair with created_at filters.
• message — a raw conversational turn that was deemed worth retaining verbatim.

Correct: Populate every applicable field at create time.

Python:

from redis_agent_memory import models

agent_memory.bulk_create_long_term_memories(memories=[
    {
        "id":          "user-42-pref-theme",
        "text":        "User 42 prefers dark mode in the dashboard.",
        "memory_type": models.MemoryType.SEMANTIC,
        "owner_id":    "user-42",
        "namespace":   "preferences",
        "topics":      ["ui", "theme"],
    },
    {
        "id":          "user-42-incident-7821",
        "text":        "User 42 hit a 500 on /api/checkout on 2026-05-10 and was refunded.",
        "memory_type": models.MemoryType.EPISODIC,
        "owner_id":    "user-42",
        "namespace":   "interactions",
        "topics":      ["incident", "billing"],
    },
])

TypeScript:

await agentMemory.bulkCreateLongTermMemories({
  memories: [
    {
      id:         "user-42-pref-theme",
      text:       "User 42 prefers dark mode in the dashboard.",
      memoryType: "semantic",
      ownerId:    "user-42",
      namespace:  "preferences",
      topics:     ["ui", "theme"],
    },
    {
      id:         "user-42-incident-7821",
      text:       "User 42 hit a 500 on /api/checkout on 2026-05-10 and was refunded.",
      memoryType: "episodic",
      ownerId:    "user-42",
      namespace:  "interactions",
      topics:     ["incident", "billing"],
    },
  ],
});

Later searches can then scope cheaply:

Python:

from datetime import datetime, timedelta, timezone

# All preferences for one user
agent_memory.search_long_term_memory(
    filter_={"owner_id": {"eq": "user-42"}, "namespace": {"eq": "preferences"}},
)

# Incidents across all users in the last 7 days
seven_days_ago = datetime.now(timezone.utc) - timedelta(days=7)
agent_memory.search_long_term_memory(
    text="checkout failure",
    filter_={
        "topics":     {"all": ["incident", "billing"]},
        "created_at": {"gte": seven_days_ago},     # tz-aware UTC datetime
    },
)

TypeScript:

// All preferences for one user
await agentMemory.searchLongTermMemory({
  filter: { ownerId: { eq: "user-42" }, namespace: { eq: "preferences" } },
});

// Incidents across all users in the last 7 days
const sevenDaysAgo = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000);
await agentMemory.searchLongTermMemory({
  text: "checkout failure",
  filter: {
    topics:    { all: ["incident", "billing"] },
    createdAt: { gte: sevenDaysAgo },             // Date
  },
});

Incorrect: Stuffing all of these into the text field.

# Bad: structured signals hidden inside free text. Search can't filter on them
# without an LLM re-parse, and similarity threshold becomes the only knob.
agent_memory.bulk_create_long_term_memories(memories=[{
    "id":   "fact-1",
    "text": "[owner=user-42][namespace=preferences][topic=ui] prefers dark mode",
}])

Updating organization later: update_long_term_memory(memory_id=..., ...) / updateLongTermMemory(memoryId, ...) accepts namespace, owner_id, session_id, topics, and memory_type. To clear a field, send an empty string ("") — omitting the field leaves it unchanged.

Avoid leakage between owners. If a record can be attributed to one user, set owner_id. A search request without an owner_id filter will happily return facts from any user in the same store.

Search Long-Term Memory Semantically with Filters

search_long_term_memory(...) (Python) / searchLongTermMemory(...) (TypeScript) runs a vector search over LTM records and applies structured filters in the same call. Combining both is the supported path — do not pull a wide vector result and filter on the client.

Correct: Pre-filter by the structured fields you already know, then rank by semantic similarity.

Python:

from redis_agent_memory import AgentMemory, models

def recall(
    agent_memory: AgentMemory,
    *,
    owner_id:  str,
    query:     str,
    namespace: str | None = None,
    k:         int        = 5,
):
    filt = {
        "owner_id":    {"eq": owner_id},
        "memory_type": {"in": ["semantic", "episodic"]},
    }
    if namespace is not None:
        filt["namespace"] = {"eq": namespace}

    res = agent_memory.search_long_term_memory(
        text=query,                              # embedded server-side
        similarity_threshold=0.7,                # normalized cosine, 0–1
        filter_op=models.FilterConjunction.ALL,  # AND across filter keys
        filter_=filt,                            # NB: trailing underscore — `filter` is reserved in Python
        limit=k,                                 # 1–100, default 10
    )
    return res.memories

TypeScript:

import { AgentMemory } from "@redis-iris/agent-memory";

async function recall(
  agentMemory: AgentMemory,
  args: { ownerId: string; query: string; namespace?: string; k?: number },
) {
  const res = await agentMemory.searchLongTermMemory({
    text:                args.query,
    similarityThreshold: 0.7,
    filterOp:            "all",                  // AND across filter keys
    filter: {
      ownerId:    { eq: args.ownerId },
      ...(args.namespace ? { namespace: { eq: args.namespace } } : {}),
      memoryType: { in: ["semantic", "episodic"] },
    },
    limit: args.k ?? 5,
  });
  return res.memories;
}

Incorrect: Querying with only text and filtering client-side.

# Bad: pulls up to 100 unrelated records per user, then re-filters in Python.
# Pays the vector-search cost on the full store, and capped at 100 results
# you may miss the one you needed.
hits = agent_memory.search_long_term_memory(text=query, limit=100).memories
for m in hits:
    if m.owner_id == owner_id and m.namespace == namespace:
        ...

Filter operators (per field):

filter_op / filterOp controls how the top-level filter fields combine: "all" (default, AND) or "any" (OR). Inside one field, eq / ne / in / all are mutually exclusive — set exactly one.

Similarity threshold: Normalized cosine similarity (0–1). Start at 0.7 and tune per workload — too high returns empty pages; too low returns noise.

Pagination: Pass next_page_token / nextPageToken back verbatim. Don't decode it; the server may change the encoding.

def iter_results(agent_memory, *, query: str, owner_id: str):
    token = None
    while True:
        page = agent_memory.search_long_term_memory(
            text=query,
            filter_={"owner_id": {"eq": owner_id}},
            limit=50,
            page_token=token,
        )
        yield from page.memories
        token = page.next_page_token
        if not token:
            return

async function* iterResults(
  agentMemory: AgentMemory,
  args: { query: string; ownerId: string },
) {
  let pageToken: string | undefined;
  while (true) {
    const page = await agentMemory.searchLongTermMemory({
      text: args.query,
      filter: { ownerId: { eq: args.ownerId } },
      limit: 50,
      pageToken,
    });
    yield* page.memories;
    if (!page.nextPageToken) return;
    pageToken = page.nextPageToken;
  }
}

No-query browsing: Omit text to apply only the structured filters (vector ranking is skipped, results are returned in record order).

Understand Background Memory Promotion

Every successful add_session_event / addSessionEvent enqueues a promote-working-memory job, fire-and-forget. The data plane never blocks on the LLM call; Redis Cloud's worker pool consumes the job, reads the session's events, calls an LLM to extract durable facts, and writes resulting records into long-term memory.

┌──────────┐  addSessionEvent  ┌──────────┐   enqueue   ┌──────────────┐
│  Agent   │ ────────────────► │ Data plane│ ─────────► │  Job queue   │
└──────────┘   200 OK          └──────────┘             │  (managed)   │
                                                       └──────┬───────┘
                                                              │ poll
                                                              ▼
                                                       ┌──────────────┐
                                                       │   Worker     │
                                                       │ • read session│
                                                       │ • call LLM   │
                                                       │ • write LTM  │
                                                       └──────────────┘

Deduplication window

Submitting a job per event would mean an LLM call per turn. To prevent that, the worker groups events into time windows. Jobs whose deduplication key would collide are run only once for that window.

• Two events landing in the same window for the same session share a deduplication key, so only one promotion job runs for that bucket.
• Window: 5 minutes (managed by Redis Cloud — not user-configurable today).
• The job is delayed until the end of the window so it sees every event in that bucket.

Eventually consistent — design for it

After an add_session_event returns 200, a search_long_term_memory for the extracted facts may not see them for up to one deduplication window plus the LLM round-trip. Don't assert synchronously in tests; poll.

Python:

import time
from redis_agent_memory import AgentMemory

def wait_for_ltm(
    agent_memory: AgentMemory,
    *,
    query:     str,
    owner_id:  str,
    timeout_s: float = 30,
):
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        hits = agent_memory.search_long_term_memory(
            text=query,
            filter_={"owner_id": {"eq": owner_id}},
            limit=5,
        ).memories
        if hits:
            return hits
        time.sleep(1.0)
    raise AssertionError("promotion did not materialize in time")

TypeScript:

import { AgentMemory } from "@redis-iris/agent-memory";

async function waitForLtm(
  agentMemory: AgentMemory,
  args: { query: string; ownerId: string; timeoutMs?: number },
) {
  const deadline = Date.now() + (args.timeoutMs ?? 30_000);
  while (Date.now() < deadline) {
    const res = await agentMemory.searchLongTermMemory({
      text:   args.query,
      filter: { ownerId: { eq: args.ownerId } },
      limit:  5,
    });
    if (res.memories.length) return res.memories;
    await new Promise((r) => setTimeout(r, 1000));
  }
  throw new Error("promotion did not materialize in time");
}

Incorrect: Assuming LTM is updated synchronously with the session write.

# Bad: race. The promotion job is enqueued but the worker hasn't run yet.
agent_memory.add_session_event(...)
results = agent_memory.search_long_term_memory(text="...").memories
assert results, "expected the new fact to be retrievable"   # flaky

What if a promotion fails?

• Submission errors are logged on the data plane but do not fail the write — add_session_event still returns 200. The trade-off is that a queue outage silently delays promotion until Cloud's monitoring picks it up.
• Worker-side failures (LLM timeout, embedding-provider 429) are retried by the workflow engine.
• Sessions that have stopped receiving events may keep trailing turns un-promoted until the next event for that session arrives.

Append a Session Event Correctly

AgentMemory.add_session_event(...) (Python) / agentMemory.addSessionEvent(...) (TypeScript) appends a single event to a session. The session is created on first write; if session_id / sessionId is omitted the server generates one (32-char UUID without dashes) and returns it on the response. Every successful write also enqueues a promotion job — so payload quality directly affects what lands in long-term memory.

Correct: Pass actor_id, role, content, and a tz-aware UTC `created_at` on every turn. Carry the same session_id for the whole conversation.

Python — created_at is a datetime.datetime (UTC, tz-aware):

from datetime import datetime, timezone
from redis_agent_memory import AgentMemory, models

def append_event(
    agent_memory: AgentMemory,
    *,
    session_id: str,
    actor_id:   str,
    role:       models.MessageRole,
    text:       str,
    metadata:   dict | None = None,
):
    return agent_memory.add_session_event(
        session_id=session_id,                       # client-supplied — keeps the turn ordered with prior turns
        actor_id=actor_id,                           # who said this (user-42, agent-1, system)
        role=role,                                   # MessageRole.USER | .ASSISTANT | .SYSTEM
        content=[{"text": text}],                    # list of typed content parts
        created_at=datetime.now(timezone.utc),       # tz-aware UTC datetime — required
        metadata=metadata,                           # any JSON, ≤ 16 KB
    ).event                                          # → server-assigned eventId, etc.

append_event(
    agent_memory,
    session_id="chat-2026-05-18-42",
    actor_id="user-42",
    role=models.MessageRole.USER,
    text="What did we agree on yesterday?",
)

TypeScript — createdAt is a Date (SDK serializes to UTC ISO-8601):

import { AgentMemory } from "@redis-iris/agent-memory";

async function appendEvent(
  agentMemory: AgentMemory,
  args: {
    sessionId: string;
    actorId:   string;
    role:      "USER" | "ASSISTANT" | "SYSTEM";
    text:      string;
    metadata?: Record<string, unknown>;
  },
) {
  const res = await agentMemory.addSessionEvent({
    sessionId: args.sessionId,
    actorId:   args.actorId,
    role:      args.role,
    content:   [{ text: args.text }],
    createdAt: new Date(),                         // UTC Date — required
    metadata:  args.metadata,
  });
  return res.event;                                // server-assigned eventId, etc.
}

await appendEvent(agentMemory, {
  sessionId: "chat-2026-05-18-42",
  actorId:   "user-42",
  role:      "USER",
  text:      "What did we agree on yesterday?",
});

Incorrect: Letting the server generate a new session_id on every turn, or passing a naive (tz-less) datetime in Python.

from datetime import datetime

# Bad: omitting session_id on every call creates a new session per turn,
# so the session memory contains exactly one event and promotion has no
# context to extract from.
agent_memory.add_session_event(
    actor_id="user-42",
    role=models.MessageRole.USER,
    content=[{"text": msg}],
    created_at=datetime.now(),                     # <-- naive datetime; ambiguous timezone.
                                                   #     Use datetime.now(timezone.utc).
)

Constraints worth remembering:

• store_id, session_id, actor_id: 1–64 chars, [a-zA-Z0-9-] only.
• role: one of USER, ASSISTANT, SYSTEM.
• content: list of typed parts; today only {"text": "..."} is supported.
• created_at / createdAt: tz-aware UTC datetime (Python) or Date (TypeScript). The SDKs serialize to ISO-8601 on the wire.
• metadata: any valid JSON document, ≤ 16 KB.
• Session TTL is governed by the store's short-memory TTL (configured at store creation). Each new event refreshes the TTL on the session key.

The response (res.event / result.event) includes the server-generated event_id / eventId (32-char UUID without dashes) — store it if you might need delete_session_event / deleteSessionEvent later. It also includes a system_timestamp / systemTimestamp (set by the data plane on ingestion) alongside the client-supplied created_at — see `session-retrieval` for how to use the two timestamps.

Async (Python)

The Python SDK exposes an _async variant for every method when used inside an async function:

import asyncio
from datetime import datetime, timezone

async def main():
    async with AgentMemory(URL, store_id=SID, api_key=KEY) as agent_memory:
        await agent_memory.add_session_event_async(
            session_id="chat-1",
            actor_id="user-42",
            role=models.MessageRole.USER,
            content=[{"text": "hi"}],
            created_at=datetime.now(timezone.utc),
        )

asyncio.run(main())

Retrieve Session Memory and Individual Events

The SDK exposes three read paths against session memory; pick the narrowest one for the job.

The session's owner_id is set from the first event's actor_id and is immutable for the lifetime of the session.

Correct: Fetch the whole session when reconstructing the agent's working context.

Python:

def load_session(agent_memory, session_id: str) -> list:
    res = agent_memory.get_session_memory(session_id=session_id)
    # res.session_id, res.owner_id, res.events (ordered by created_at)
    return res.events

TypeScript:

async function loadSession(agentMemory: AgentMemory, sessionId: string) {
  const res = await agentMemory.getSessionMemory(sessionId);
  // res.sessionId, res.ownerId, res.events (ordered by createdAt)
  return res.events;
}

Two timestamps on each event. Every SessionEvent in the response carries:

• created_at / createdAt — the client-supplied UTC timestamp you passed at write time. This is what the agent considers "when the turn happened" and what events are ordered by.
• system_timestamp / systemTimestamp — a server-set UTC timestamp recording when the data plane ingested the event. Useful for diagnostics (e.g. clock skew between agent and server, or detecting replayed events with stale created_at values).

Both are datetime (Python) / Date (TypeScript) on the SDK side; serialized as UTC ISO-8601 on the wire.

Correct: Page through sessions for admin tools.

Python:

def iter_session_ids(agent_memory):
    token = None
    while True:
        page = agent_memory.list_sessions(limit=200, page_token=token)
        yield from page.sessions
        token = page.next_page_token
        if not token:
            return

TypeScript:

async function* iterSessionIds(agentMemory: AgentMemory) {
  let pageToken: string | undefined;
  while (true) {
    const page = await agentMemory.listSessions({ limit: 200, pageToken });
    yield* page.sessions;
    if (!page.nextPageToken) return;
    pageToken = page.nextPageToken;
  }
}

Incorrect: Paging the full session list to find one event you already have an ID for.

# Bad: O(sessions) just to find one eventId you already have.
for sid in iter_session_ids(agent_memory):
    for ev in load_session(agent_memory, sid):
        if ev.event_id == target_event_id:
            return ev

Use get_session_event / getSessionEvent instead — it is an O(1) lookup.

Pagination limits:

• list_sessions.limit defaults to 100, max 1000.
• next_page_token / nextPageToken is opaque — pass it back verbatim, don't try to decode it.

Deletion:

• delete_session_memory(session_id=...) / deleteSessionMemory(sessionId) removes the entire session and all its events.
• delete_session_event(session_id=..., event_id=...) / deleteSessionEvent(sessionId, eventId) removes one event.
• Already-promoted long-term memories are not affected — delete those separately via bulk_delete_long_term_memories / bulkDeleteLongTermMemories.

Choose Session Events vs Long-Term Memory

Redis Agent Memory has two tiers. They serve different jobs — picking the wrong one is the single biggest source of cost and correctness problems.

Correct: Append every turn of the conversation as a session event. Let the background promotion worker decide what becomes long-term memory.

Python:

from datetime import datetime, timezone
from redis_agent_memory import AgentMemory, models

# Every user/assistant turn → add_session_event. That's it.
agent_memory.add_session_event(
    session_id=session_id,
    actor_id="user-42",
    role=models.MessageRole.USER,
    content=[{"text": user_msg}],
    created_at=datetime.now(timezone.utc),
)
agent_memory.add_session_event(
    session_id=session_id,
    actor_id="agent-1",
    role=models.MessageRole.ASSISTANT,
    content=[{"text": reply}],
    created_at=datetime.now(timezone.utc),
)
# Promotion happens asynchronously — see promotion-overview.

TypeScript:

await agentMemory.addSessionEvent({
  sessionId: sessionId,
  actorId:   "user-42",
  role:      "USER",
  content:   [{ text: userMsg }],
  createdAt: new Date(),
});
await agentMemory.addSessionEvent({
  sessionId: sessionId,
  actorId:   "agent-1",
  role:      "ASSISTANT",
  content:   [{ text: reply }],
  createdAt: new Date(),
});

Correct: Write to long-term memory directly when you already have a structured fact and don't want to pay for extraction.

Python:

# Pre-known fact — skip the LLM and write LTM directly.
agent_memory.bulk_create_long_term_memories(memories=[
    {
        "id":          "user-42-timezone",
        "text":        "User 42 is in Europe/Sofia (UTC+2/+3).",
        "memory_type": models.MemoryType.SEMANTIC,
        "owner_id":    "user-42",
        "topics":      ["profile", "timezone"],
    },
])

TypeScript:

await agentMemory.bulkCreateLongTermMemories({
  memories: [
    {
      id:         "user-42-timezone",
      text:       "User 42 is in Europe/Sofia (UTC+2/+3).",
      memoryType: "semantic",
      ownerId:    "user-42",
      topics:     ["profile", "timezone"],
    },
  ],
});

Incorrect: Using long-term memory as the conversation buffer.

# Bad: each turn pays for embedding + LTM write, and the agent loses turn order.
for turn in conversation:
    agent_memory.bulk_create_long_term_memories(memories=[{
        "id":          f"{session_id}-{turn.idx}",
        "text":        turn.text,
        "memory_type": models.MemoryType.MESSAGE,
    }])

Why it's bad: LTM is vector-indexed (cost per write) and unordered (you re-paginate to reconstruct a session). Session memory is append-only and keeps createdAt order for free.

Rule of thumb: if you'd want to retrieve it in a different future conversation, it belongs in LTM (usually via promotion). If you only need it for the current turn or the rest of this session, it stays in session memory.

Authenticate the SDK with a Store API Key

Every data-plane request carries Authorization: Bearer <store-api-key>. The SDKs add this header for you — just pass the key (and store ID) at client construction. Both SDKs follow the same convention:

store_id / storeId is a global parameter — set it on the client once and every per-store operation uses it by default. You can still override it per call.

Correct: Read the key from a secrets manager (or environment) and construct the client once per process.

Python:

import os
from redis_agent_memory import AgentMemory

# Construct once at startup; reuse across requests.
# The `with` block ensures the underlying httpx client is closed cleanly.
def make_client() -> AgentMemory:
    return AgentMemory(
        os.environ["AGENT_MEMORY_BASE_URL"],         # e.g. https://gcp-us-east4.memory.redis.io
        store_id=os.environ["AGENT_MEMORY_STORE_ID"],
        api_key=os.environ["AGENT_MEMORY_API_KEY"],
    )

with make_client() as agent_memory:
    agent_memory.health()

TypeScript:

import { AgentMemory } from "@redis-iris/agent-memory";

// Construct once at module scope; the SDK is safe to share across requests.
export const agentMemory = new AgentMemory({
  serverURL: process.env.AGENT_MEMORY_BASE_URL!,
  storeId:   process.env.AGENT_MEMORY_STORE_ID!,
  apiKey:    process.env.AGENT_MEMORY_API_KEY!,
});

await agentMemory.health();

Incorrect: Hard-coding the key, building the client per request, or constructing the bearer header by hand.

# Bad: key in source — leaks through git history and any traceback log.
agent_memory = AgentMemory(
    "https://gcp-us-east4.memory.redis.io",
    store_id="01HZ...",
    api_key="sk-live-abc123...",       # <-- never
)

# Bad: new client per request — wastes the underlying TCP/TLS pool.
def handle(req):
    with AgentMemory(URL, store_id=SID, api_key=KEY) as agent_memory:
        return agent_memory.get_session_memory(session_id=req.session_id)

// Bad: rebuilding the Authorization header manually defeats the SDK's typing,
// retry, and error-class machinery.
const res = await fetch(`${URL}/v1/stores/${SID}/session-memory/events`, {
  method:  "POST",
  headers: { Authorization: `Bearer ${KEY}` },
  body:    JSON.stringify(event),
});

Rotation: Regenerate the key from the Cloud console. There is no short-lived-token flow — rotation is the only mitigation if a key leaks.

Per-operation override: Both SDKs accept a store_id / storeId argument on every call, which overrides the global. Use this when one process talks to multiple stores; do not use it to "scope" calls — global is fine for single-store apps.

Reference: Redis Cloud Agent Memory

Create a Memory Service on Redis Cloud

Redis Cloud provisions the memory store, the backing Redis database, the background promotion worker, and the LLM/embedding provider credentials. Each store gets a unique storeId and a store API key used as a bearer token on every data-plane request.

Correct: Provision through the Redis Cloud Agent Memory console.

1. Sign in at https://cloud.redis.io/#/agent-memory. 2. Click New service and pick the correct settings for the user. 3. After provisioning, copy three values from the console:

• Server URL — the production data-plane URL is https://gcp-us-east4.memory.redis.io (your exact URL is shown in the Cloud console)
• Store ID — 32-character UUID without dashes
• Store API key — Bearer token (treat like a secret)

4. Export them so the SDKs can read them from the environment:

export AGENT_MEMORY_BASE_URL="https://gcp-us-east4.memory.redis.io"
export AGENT_MEMORY_STORE_ID="<your-store-id>"
export AGENT_MEMORY_API_KEY="<your-store-api-key>"

5. Install the SDK and run a smoke test.

Python — pip install redis-agent-memory:

import os
from datetime import datetime, timezone
from redis_agent_memory import AgentMemory, models

with AgentMemory(
    os.environ["AGENT_MEMORY_BASE_URL"],
    store_id=os.environ["AGENT_MEMORY_STORE_ID"],
    api_key=os.environ["AGENT_MEMORY_API_KEY"],
) as agent_memory:
    # Health check
    print(agent_memory.health())

    # Sanity write
    res = agent_memory.add_session_event(
        actor_id="user-42",
        role=models.MessageRole.USER,
        content=[{"text": "hello"}],
        created_at=datetime.now(timezone.utc),   # tz-aware UTC datetime
    )
    print(res.event.event_id)

TypeScript — npm add @redis-iris/agent-memory:

import { AgentMemory } from "@redis-iris/agent-memory";

const agentMemory = new AgentMemory({
  serverURL: process.env.AGENT_MEMORY_BASE_URL!,
  storeId:   process.env.AGENT_MEMORY_STORE_ID!,
  apiKey:    process.env.AGENT_MEMORY_API_KEY!,
});

async function smokeTest() {
  console.log(await agentMemory.health());

  const res = await agentMemory.addSessionEvent({
    actorId:   "user-42",
    role:      "USER",
    content:   [{ text: "hello" }],
    createdAt: new Date(),                       // SDK serializes to UTC ISO-8601
  });
  console.log(res.event.eventId);
}

smokeTest();

Store the API key in a secrets manager. It scopes access to a single store; rotating it requires regenerating from the Cloud console.

Reference: Redis Cloud Agent Memory · Python SDK · TypeScript SDK