progress-memory (Core)

Updated on Nov 3, 2026

The Memory Python SDK is split into three packages so you take only the dependencies you need:

PackageMain purposeWhen to use
progress-memoryCore client for add/recall/forget memory operations.You want direct programmatic control of memory calls.
progress-memory-langchainIntegration with LangChain chains and agents.You are building with LangChain and want automatic or tool-based memory.
progress-memory-google-adkIntegration with Google ADK agents via BaseMemoryService.You are using Google ADK and want native ADK memory lifecycle support.

Before configuring the SDK, request early access and prepare the Agent Memory prerequisites.

Installation

Install the core Memory package:

bash
pip install progress-memory

Configuration

Create a ProgressMemoryConfig with the Memory backend URL and your SDK API key. Replace pmem_d_... with your own SDK key. For production, load the key from an environment variable or secret manager rather than hard-coding it.

python
from progress_memory import ProgressMemoryConfig

config = ProgressMemoryConfig(
    backend_base_url="https://memory-api.observability.progress.com",
    sdk_api_key="pmem_d_...",
)

Environment Variables

Alternatively, set the following environment variables and call ProgressMemoryConfig.from_env() to load the configuration:

VariablePurpose
PROGRESS_MEMORY_BACKEND_URLBase URL of the memory-api service
PROGRESS_MEMORY_SDK_API_KEYSDK API key (for example, pmem_d_*)

API Reference

ProgressMemoryClient exposes three async methods (sync variants are also available for blocking callers).

add_memory

Extracts atomic memories from one or more conversation messages and saves them to the knowledge base.

python
async def add_memory(request: AddMemoryRequest, context: ConversationContext | None = None) -> AddMemoryResult

AddMemoryRequest

FieldTypePurpose
user_idstrIdentifies the user whose memory this belongs to.
agent_idstrIdentifies the agent (scopes memory per agent per user).
messageslist[ConversationMessage] | NoneOrdered conversation messages for extraction.
conversation_idstr | NoneOptional conversation identifier to thread multi-turn exchanges.
metadatadict[str, str] | NoneOptional key/value pairs forwarded to the backend.

ConversationMessage

FieldTypePurpose
textstrMessage text.
typestrMessage role: "user" or "assistant".

AddMemoryResult

FieldTypePurpose
original_text_idstrRecord ID for the submitted text.
memories_countintNumber of atomic memories extracted and stored.
conversation_idstr | NoneConversation identifier for threading follow-up turns.
memorieslist[StoredMemory]The individual extracted memories.

StoredMemory

FieldTypePurpose
memory_idstrUnique identifier for the memory. Pass to forget_memory to delete it.
textstrThe extracted atomic memory text.
categorystrCategory assigned during extraction (for example, FACT, PREFERENCE, or MEMORY).

recall_memory

Performs a search over stored memories for a given user/agent pair.

python
async def recall_memory(request: RecallMemoryRequest) -> list[RecallResult]

RecallMemoryRequest

FieldTypePurpose
querystrNatural-language query describing what to look for.
user_idstrScope results to this user.
agent_idstrScope results to this agent.
top_kintMaximum number of results to return (default 10).
include_outdatedboolOptional. Set to True to return both active and outdated memories. If omitted, defaults to False (active memories only).

RecallResult

FieldTypePurpose
memory_idstrUnique identifier for the memory. Pass to forget_memory to delete it.
textstrThe stored atomic memory text.
scorefloatSemantic similarity score (0–1). Higher is more relevant.
categorystr | NoneCategory label assigned during extraction.
original_text_idstr | NoneID of the original text record this memory was derived from.
statusstr | NoneMemory status: active or outdated.
outdated_by_idstr | NoneID of the replacement memory when this memory is outdated.
metadatadict[str, Any]Additional metadata from the backend resource.

forget_memory

Deletes a specific extracted memory by its ID. The original conversation text is not affected. user_id must match the owner of the memory — passing a different user ID raises a 403 error.

python
async def forget_memory(memory_id: str, user_id: str) -> None

forget_memory is idempotent — calling it for a memory that no longer exists returns successfully.

Standalone Usage Example

The following example creates a client, stores a conversation exchange, recalls relevant memories, and deletes a recalled memory:

python
import asyncio
from progress_memory import (
    AddMemoryRequest,
    ConversationMessage,
    ProgressMemoryClient,
    ProgressMemoryConfig,
    RecallMemoryRequest,
)

async def main() -> None:
    config = ProgressMemoryConfig(
        backend_base_url="https://memory-api.observability.progress.com",
        sdk_api_key="pmem_d_...",
    )

    async with ProgressMemoryClient(config) as client:
        # Save a conversation exchange for memory extraction
        result = await client.add_memory(
            AddMemoryRequest(
                user_id="user-123",
                agent_id="support-bot",
                messages=[
                    ConversationMessage(text="I prefer dark mode and metric units.", type="user"),
                    ConversationMessage(text="Noted — I'll remember your display and unit preferences.", type="assistant"),
                ],
            )
        )

        print(f"Extracted {result.memories_count} memory(s):")
        for item in result.memories:
            print(f"  [{item.category}] {item.text}  (id: {item.memory_id})")

        # Recall relevant memories before answering the next question
        memories = await client.recall_memory(
            RecallMemoryRequest(
                query="display preferences",
                user_id="user-123",
                agent_id="support-bot",
                top_k=5,
                include_outdated=False,  # Optional; omit for active-only retrieval, or set to True to include outdated memories.
            )
        )

        for m in memories:
            print(f"[{m.score:.2f}] [{m.category}] {m.text}")

        # Forget a specific memory
        if memories:
            await client.forget_memory(memories[0].memory_id, "user-123")

asyncio.run(main())

See Also