MemoryClient API Reference

from prme import MemoryClient
 
# From a directory path (creates if needed)
client = MemoryClient("./my_memories")
 
# With explicit config
from prme.config import PRMEConfig
config = PRMEConfig(db_path="./custom.duckdb", vector_path="./vecs.usearch", lexical_path="./lex")
client = MemoryClient(config=config)

# Context manager (recommended)
with MemoryClient("./my_memories") as client:
    client.store("hello", user_id="alice")
 
# Manual lifecycle
client = MemoryClient("./my_memories")
try:
    client.store("hello", user_id="alice")
finally:
    client.close()

def store(
    content: str,
    *,
    user_id: str,
    session_id: str | None = None,
    role: str = "user",
    speaker: str | None = None,
    node_type: NodeType = NodeType.NOTE,
    scope: Scope = Scope.PERSONAL,
    metadata: dict | None = None,
    confidence: float | None = None,
    epistemic_type: EpistemicType | None = None,
    source_type: SourceType | None = None,
    event_time: datetime | None = None,
    ttl_days: int | None = ...,
) -> str  # returns event UUID

from prme.types import NodeType, Scope
 
event_id = client.store(
    "The deployment target is AWS us-east-1.",
    user_id="alice",
    node_type=NodeType.DECISION,
    scope=Scope.PROJECT,
    metadata={"source": "architecture-review"},
)

def retrieve(
    query: str,
    *,
    user_id: str,
    scope: Scope | list[Scope] | None = None,
    time_from: datetime | None = None,
    time_to: datetime | None = None,
    reference_time: datetime | None = None,
    knowledge_at: datetime | None = None,
    token_budget: int | None = None,
) -> RetrievalResponse

response = client.retrieve("What tech stack are we using?", user_id="alice")
 
for result in response.results[:5]:
    node = result.node
    print(f"[{result.composite_score:.3f}] {node.node_type.value}: {node.content}")

event_id = client.ingest_fast("Alice prefers dark mode", user_id="alice")
result = client.process_pending(user_id="alice", budget_ms=1000)
status = client.processing_status(event_id, user_id="alice")

def ingest(
    content: str,
    *,
    user_id: str,
    role: str = "user",
    speaker: str | None = None,
    session_id: str | None = None,
    scope: Scope = Scope.PERSONAL,
) -> str  # returns event UUID

event_id = client.ingest(
    "I just finished migrating our auth system from JWT to session tokens.",
    user_id="alice",
    scope=Scope.PROJECT,
)

def ingest_batch(
    messages: list[dict],
    *,
    user_id: str,
    session_id: str | None = None,
    scope: Scope = Scope.PERSONAL,
) -> list[str]  # returns list of event UUIDs

event_ids = client.ingest_batch(
    [
        {"role": "user", "content": "Let's use Redis for caching."},
        {"role": "assistant", "content": "Good call. I'll note that decision."},
    ],
    user_id="alice",
    session_id="planning-session",
)

def observe(
    self,
    source: str | os.PathLike[str] | bytes,
    *,
    user_id: str,
    captured_at: datetime | None = None,
    source_name: str | None = None,
    session_id: str | None = None,
    metadata: dict | None = None,
    scope: Scope = Scope.PERSONAL,
) -> list[str]

ids = client.observe("~/Pictures/receipt.jpg", user_id="alice",
                     captured_at=datetime(2026, 10, 12, 8, 41, tzinfo=timezone.utc))
event = client.get_event(ids[0], user_id="alice")
thumbnail = client.get_blob(event.metadata["thumbnail"])  # JPEG bytes

def get_node(node_id: str) -> MemoryNode | None

node = client.get_node("550e8400-e29b-41d4-a716-446655440000")
if node:
    print(f"{node.node_type.value}: {node.content}")
    print(f"State: {node.lifecycle_state.value}, Confidence: {node.confidence:.2f}")

def query_nodes(**kwargs) -> list[MemoryNode]

from prme.types import NodeType
 
facts = client.query_nodes(user_id="alice", node_type=NodeType.FACT, limit=20)
for fact in facts:
    print(f"[{fact.lifecycle_state.value}] {fact.content}")

def organize(
    *,
    user_id: str | None = None,
    jobs: list[str] | None = None,
    budget_ms: int = 5000,
) -> OrganizeResult

# Run all jobs
result = client.organize(user_id="alice")
print(f"Jobs run: {result.jobs_run}")
 
# Run specific jobs
result = client.organize(jobs=["promote", "deduplicate"], budget_ms=2000)

client.close()

from fastapi import FastAPI
from prme import MemoryClient
 
app = FastAPI()
client = MemoryClient("./memories")
 
@app.on_event("shutdown")
def shutdown():
    client.close()
 
@app.post("/remember")
def remember(content: str, user_id: str):
    event_id = client.store(content, user_id=user_id)
    return {"event_id": event_id}
 
@app.get("/recall")
def recall(query: str, user_id: str):
    response = client.retrieve(query, user_id=user_id)
    return [{"content": r.node.content, "score": r.composite_score} for r in response.results]

from prme import MemoryClient
 
client = MemoryClient("./agent_memory")
 
def store_memory(content: str, user_id: str):
    """Tool for the agent to store memories."""
    return client.store(content, user_id=user_id)
 
def recall_memory(query: str, user_id: str):
    """Tool for the agent to recall relevant memories."""
    response = client.retrieve(query, user_id=user_id)
    return "\n".join(r.node.content for r in response.results[:5])

from prme import MemoryClient
 
client = MemoryClient("./notebook_memories")
 
# Store during analysis
client.store("Customer churn rate is 5.2% this quarter.", user_id="analyst")
 
# Query later
response = client.retrieve("What is the churn rate?", user_id="analyst")
response.results[0].node.content
# → "Customer churn rate is 5.2% this quarter."
 
# Don't forget to close
client.close()