progress-memory (Core)
The Memory Python SDK is split into three packages so you take only the dependencies you need:
| Package | Main purpose | When to use |
|---|---|---|
| progress-memory | Core client for add/recall/forget memory operations. | You want direct programmatic control of memory calls. |
| progress-memory-langchain | Integration with LangChain chains and agents. | You are building with LangChain and want automatic or tool-based memory. |
| progress-memory-google-adk | Integration 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:
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.
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:
| Variable | Purpose |
|---|---|
PROGRESS_MEMORY_BACKEND_URL | Base URL of the memory-api service |
PROGRESS_MEMORY_SDK_API_KEY | SDK 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.
async def add_memory(request: AddMemoryRequest, context: ConversationContext | None = None) -> AddMemoryResult
AddMemoryRequest
| Field | Type | Purpose |
|---|---|---|
user_id | str | Identifies the user whose memory this belongs to. |
agent_id | str | Identifies the agent (scopes memory per agent per user). |
messages | list[ConversationMessage] | None | Ordered conversation messages for extraction. |
conversation_id | str | None | Optional conversation identifier to thread multi-turn exchanges. |
metadata | dict[str, str] | None | Optional key/value pairs forwarded to the backend. |
ConversationMessage
| Field | Type | Purpose |
|---|---|---|
text | str | Message text. |
type | str | Message role: "user" or "assistant". |
AddMemoryResult
| Field | Type | Purpose |
|---|---|---|
original_text_id | str | Record ID for the submitted text. |
memories_count | int | Number of atomic memories extracted and stored. |
conversation_id | str | None | Conversation identifier for threading follow-up turns. |
memories | list[StoredMemory] | The individual extracted memories. |
StoredMemory
| Field | Type | Purpose |
|---|---|---|
memory_id | str | Unique identifier for the memory. Pass to forget_memory to delete it. |
text | str | The extracted atomic memory text. |
category | str | Category assigned during extraction (for example, FACT, PREFERENCE, or MEMORY). |
recall_memory
Performs a search over stored memories for a given user/agent pair.
async def recall_memory(request: RecallMemoryRequest) -> list[RecallResult]
RecallMemoryRequest
| Field | Type | Purpose |
|---|---|---|
query | str | Natural-language query describing what to look for. |
user_id | str | Scope results to this user. |
agent_id | str | Scope results to this agent. |
top_k | int | Maximum number of results to return (default 10). |
include_outdated | bool | Optional. Set to True to return both active and outdated memories. If omitted, defaults to False (active memories only). |
RecallResult
| Field | Type | Purpose |
|---|---|---|
memory_id | str | Unique identifier for the memory. Pass to forget_memory to delete it. |
text | str | The stored atomic memory text. |
score | float | Semantic similarity score (0–1). Higher is more relevant. |
category | str | None | Category label assigned during extraction. |
original_text_id | str | None | ID of the original text record this memory was derived from. |
status | str | None | Memory status: active or outdated. |
outdated_by_id | str | None | ID of the replacement memory when this memory is outdated. |
metadata | dict[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.
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:
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())