Memory Integration

This guide shows how to integrate memory into your agents and workflows. Memory lets agents remember information across sessions, accumulate domain knowledge, and personalize responses. Memory is built into the Utils service (em-runtime-utils) — the same service that handles schedules, metadata, and catalog tags. No separate service is required.

Prerequisites

A valid JWT token from the platform (user or service account)
The JWT must contain an org_id claim for organization scoping
An X-Project-ID header for project scoping on all memory requests

Basic Integration Pattern

The most common pattern for an agent:

Before responding, list memories from the relevant Context Pack for context
After responding, extract and store new memories from the conversation

import httpx

UTILS_BASE = "https://api.example.com/utils"

async def agent_respond(user_message: str, token: str, pack_name: str, project_id: str) -> str:
    headers = {
        "Authorization": f"Bearer {token}",
        "X-Project-ID": project_id,
    }

    async with httpx.AsyncClient() as client:
        # Step 1: List memories from the relevant Context Pack
        mem_resp = await client.get(
            f"{UTILS_BASE}/context-packs/{pack_name}/memories",
            headers=headers,
        )
        memories = mem_resp.json().get("data", [])

        # Step 2: Build context from memories
        context = "\n".join(f"- {m['content']}" for m in memories)

        # Step 3: Call LLM with memory context
        prompt = f"Relevant context:\n{context}\n\nUser: {user_message}"
        response = await call_llm(prompt)

        # Step 4: Store new memories from this interaction
        if should_memorize(user_message, response):
            await client.post(
                f"{UTILS_BASE}/context-packs/{pack_name}/memories",
                headers=headers,
                json={
                    "name": "session_fact",
                    "content": {
                        "type": "text",
                        "text": extract_memorable_fact(user_message, response),
                    },
                    "memory_type": "fact",
                    "source": {"type": "agent"},
                },
            )

    return response

Storing Memories

The following snippets show individual API calls. In an async agent, wrap them in async with httpx.AsyncClient() as client: and use await client.post(...) / await client.get(...) as shown in the integration pattern above.

Basic Memory Creation

Memories are created within a named Context Pack:

# Store a user preference in a context pack
await client.post(
    f"{UTILS_BASE}/context-packs/{pack_name}/memories",
    headers=headers,
    json={
        "name": "metric_explanation_style",
        "content": {
            "type": "text",
            "text": "User prefers metric definitions before SQL queries",
        },
        "memory_type": "preference",
        "source": {"type": "user"},
    },
)

Memory Types

Choose the type based on the nature of the information:

Type	When to use	Example
`fact`	Factual knowledge	”The `orders` table has 50M rows”
`experience`	Learned behaviors	”Query optimization worked better with CTEs”
`observation`	Observed patterns	”Revenue spikes every Q4”
`instruction`	Directives / guidelines	”Always include date filters for large tables”
`preference`	User/system preferences	”Prefers concise output without verbose explanations”
`summary`	Condensed summaries	”Session covered revenue analysis across 3 regions”
`glossary`	Business terms	”ARR = Annual Recurring Revenue”
`join`	Table relationships	”Table `orders` has FK to `customers`”
`ontology`	Domain classifications	”Column `status` uses enum: active, inactive, churned”
`policy`	Governance rules	”PII columns must not appear in SQL results”
`textual_pattern`	Text patterns in data	”Email column matches a standard email regex pattern”
`kpi`	KPI definitions	”Churn Rate = lost customers / total customers x 100”
`exemplar`	Representative examples	”Typical order: id=1234, amount=49.99, status=completed”
`numeric_pattern`	Numeric distributions	”Revenue column: mean=1250, median=980, stddev=450”

Named Memories

Use name for memories you want to retrieve directly by name within a pack:

# Store with a name for direct retrieval
await client.post(
    f"{UTILS_BASE}/context-packs/{pack_name}/memories",
    headers=headers,
    json={
        "name": "user_data_source_primary",
        "content": {
            "type": "text",
            "text": "User's primary data source is Snowflake, database: analytics_prod",
        },
        "memory_type": "fact",
        "source": {"type": "user"},
    },
)

# Retrieve by name within a pack
response = await client.get(
    f"{UTILS_BASE}/context-packs/{pack_name}/memories/user_data_source_primary",
    headers=headers,
)

# Retrieve by memory ID
response = await client.get(
    f"{UTILS_BASE}/memories/{memory_id}",
    headers=headers,
)

Retrieving Memories

List All Memories in a Context Pack

response = await client.get(
    f"{UTILS_BASE}/context-packs/{pack_name}/memories",
    headers=headers,
)
memories = response.json().get("data", [])
for memory in memories:
    print(f"[{memory['memory_type']}] {memory['content']}")

List All Memories in the Project

response = await client.get(
    f"{UTILS_BASE}/memories",
    headers=headers,
)
all_memories = response.json()  # {"data": [...], "pagination": {...}}

Context Packs

For structured memory management, use Context Packs to group related memories by domain:

# Create a domain knowledge pack
pack_resp = await client.post(
    f"{UTILS_BASE}/context-packs",
    headers=headers,
    json={
        "name": "user_data_domain",
        "pack_type": "domain",
        "description": "User's data domain knowledge: schema, metrics, business rules",
    },
)
pack_name = pack_resp.json()["name"]

# Add a memory to the pack
await client.post(
    f"{UTILS_BASE}/context-packs/{pack_name}/memories",
    headers=headers,
    json={
        "name": "revenue_metric_definition",
        "content": {
            "type": "text",
            "text": "Revenue metric = sum of completed order amounts, excluding refunds",
        },
        "memory_type": "fact",
        "source": {"type": "agent"},
    },
)

# Retrieve all memories from the pack
response = await client.get(
    f"{UTILS_BASE}/context-packs/{pack_name}/memories",
    headers=headers,
)
memories = response.json().get("data", [])

Updating and Deleting

Update a Memory

# When a fact changes, update rather than delete
await client.patch(
    f"{UTILS_BASE}/memories/{memory_id}",
    headers=headers,
    json={
        "content": {
            "type": "text",
            "text": "User's primary data source is BigQuery, project: analytics_prod (migrated from Snowflake)",
        },
    },
)

Delete a Memory

# Soft delete (recoverable)
await client.delete(
    f"{UTILS_BASE}/memories/{memory_id}",
    headers=headers,
)

# Permanent delete (irreversible)
await client.post(
    f"{UTILS_BASE}/memories/{memory_id}/permanent-delete",
    headers=headers,
)

Pin a Memory

Pinned memories stay in the Active tier and bypass inactivity decay:

await client.post(
    f"{UTILS_BASE}/memories/{memory_id}/pin",
    headers=headers,
)

Best Practices

Keep memories self-contained

Each memory should make sense without knowing the conversation that produced it. Instead of “Yes, that’s right” store “User confirmed that the ‘revenue’ column uses gross revenue, not net.”

Use specific memory types

Typed memories retrieve more accurately. If you know a memory is a preference, mark it preference, use specific types for better retrieval.

Check for existing memories before storing

Before creating a new memory, list the pack’s memories and check whether an existing one covers the same information. Update existing memories rather than creating duplicates.

Use Context Packs for domain knowledge

For domain-specific knowledge bases (data dictionaries, business rules, entity relationships), group memories into typed Context Packs. This improves retrieval precision and enables pack-level operations.

Scope memories to organizations and projects

Memories are scoped by the org_id claim in the JWT token and the X-Project-ID header. Ensure each agent uses a token scoped to the correct organization and passes the correct project ID to prevent information leakage.

Memory Service

Concepts overview and API reference.

Context Packs

Detailed Context Pack architecture and management.

​Memory Integration

​Prerequisites

​Basic Integration Pattern

​Storing Memories

​Basic Memory Creation

​Memory Types

​Named Memories

​Retrieving Memories

​List All Memories in a Context Pack

​List All Memories in the Project

​Context Packs

​Updating and Deleting

​Update a Memory

​Delete a Memory

​Pin a Memory

​Best Practices

​Related

Memory Service

Context Packs

Memory Integration

Prerequisites

Basic Integration Pattern

Storing Memories

Basic Memory Creation

Memory Types

Named Memories

Retrieving Memories

List All Memories in a Context Pack

List All Memories in the Project

Context Packs

Updating and Deleting

Update a Memory

Delete a Memory

Pin a Memory

Best Practices

Related