Long-term memory

Remember user preferences across sessions. Your Agent Tools decide what to save and when to retrieve it; Harnest does not extract memories or add them to prompts automatically.

Connect your existing database

Create lifecycle/memory.py in your agent folder. Both database providers ship with the managed Harnest runtime.

import os

from harnest import lifecycle
from harnest_postgres import PostgresMemoryStore

@lifecycle.storage.memory
def memory() -> PostgresMemoryStore:
    """Store explicit memories in our existing PostgreSQL database."""
    return PostgresMemoryStore(
        os.environ["DATABASE_URL"],
        pool_options={"min_size": 1, "max_size": 5},
    )

import os

from harnest import lifecycle
from harnest_redis import RedisMemoryStore

@lifecycle.storage.memory
def memory() -> RedisMemoryStore:
    """Use a persistent Redis deployment for cross-session memories."""
    return RedisMemoryStore(os.environ["REDIS_URL"])

from harnest import lifecycle
from harnest.memory import InMemoryStore

@lifecycle.storage.memory
def memory() -> InMemoryStore:
    """Keep test memories local to this process."""
    return InMemoryStore()

Supply the connection URL through your environment or deployment secrets. From your agent folder:

harnest env sync .
harnest test .
harnest serve .

Provider	Setup
PostgreSQL	Creates `harnest_memories` using its own pool; no vector extension needed. Requires table-creation permission, or `setup_schema=False` for a pre-provisioned table.
Redis	Configure persistence, backups, replication, and no eviction. A cache-only deployment is not durable.
`InMemoryStore`	For tests; loses memories when the process exits.

Harnest owns startup and shutdown. The memory factory is optional; remove it when unused. For shared configuration or direct Python driver installation, see Task and cron storage.

Write deliberately from a tool

Save a preference only when the user asks:

tools/remember_report_preference.py

from harnest import context
from harnest.agent import tool

@tool
async def remember_report_preference(preference: str) -> dict:
    """Remember a reporting preference the user explicitly asks us to save."""
    record = await context.memory.put(
        key="report-preference",
        content=preference,
        metadata={"category": "preferences"},
    )
    return {"saved": True, "revision": record.revision}

context.memory is scoped to the current application and authenticated user, including calls from SubAgents. Replicas share an application identity; independent agents need distinct identities. Use namespaces to organize one user’s memories:

preferences = context.memory.namespace("preferences")
record = await preferences.get("report-format")
await preferences.put("report-format", "Start with an executive summary.")

Await operations within the invocation; handles cannot be retained for later use.

Read, search, update, and forget

MemoryContext supports explicit writes, reads, bounded listing, search, and deletion. See the operation reference for signatures and return values. All operations are async. Search is literal text matching, not semantic search. Treat retrieved content as untrusted data.

page = await context.memory.search("executive summary", limit=5)
for record in page.items:
    print(record.key)  # Avoid logging private memory content.

if page.next_cursor is not None:
    next_page = await context.memory.search(
        "executive summary", limit=5, after=page.next_cursor
    )

Continue until next_cursor is None, even after an empty page. Concurrent edits can change later pages. Use a revision to avoid overwriting a concurrent update:

record = await context.memory.get("report-preference")
if record is not None:
    updated = await context.memory.put(
        record.key,
        "Use bullet points and include a summary.",
        expected_revision=record.revision,
    )
    await context.memory.delete(updated.key, expected_revision=updated.revision)

Catch harnest.memory.MemoryConflictError for stale revisions. Without expected_revision, put() replaces the current record unconditionally.

Retention and limits

Set expires_at to hide expired memories, then purge them for physical deletion. Session deletion does not erase memory. See record and pagination limits. Use a separate database for evaluations that write memories.

Run trusted storage maintenance

These Python provider methods are for trusted application code—not built-in HTTP endpoints or generated Agent Tools. Any route you add must enforce authentication and authorization. Use MemoryStore.delete_all for one namespace or delete_user for account erasure. Use purge_expired for bounded cleanup sweeps. Sweep expired records using your configured provider:

from harnest.memory import MemoryScope, MemoryStore

async def purge_namespace(store: MemoryStore, scope: MemoryScope) -> int:
    """Sweep expired records in an application-authorized namespace."""
    cursor, removed = None, 0
    while True:
        page = await store.purge_expired(scope, limit=100, after=cursor)
        removed += page.deleted
        cursor = page.next_cursor
        if cursor is None:
            return removed

Schedule repeated sweeps with your maintenance worker or Harnest cron. Before account erasure, authorize the owner and revoke writes to prevent new memories. Redis deletes namespaces in batches; retry after a partial failure. Deletion does not erase backups, database logs, transcripts, or previously returned copies.

Handle connection failures

MemoryStorageError can mean the database committed but its acknowledgement was lost. Read the key again before retrying, and use revision checks to avoid overwriting a later update. Harnest does not automatically replay Redis writes after a lost acknowledgement.

Implement a custom provider

Implement the MemoryStore contract and its associated record, scope, and error types. Register your provider with @lifecycle.storage.memory. Follow MemoryStore for isolation, atomic revisions, and detached results; subclassing is optional. Keep explicit MemoryScope values in trusted code. Run the bundled conformance suite against an isolated real database:

import unittest

from harnest.testing import MemoryStoreConformanceMixin
from my_storage import MyMemoryStore

class CustomMemoryTests(MemoryStoreConformanceMixin,
                        unittest.IsolatedAsyncioTestCase):
    async def make_store(self):
        """Return an unstarted provider for an isolated test database."""
        return MyMemoryStore(...)

The mixin uses a unique application scope per test. Add test-data cleanup and provider-specific crash, rollback, and connection-loss checks; also verify your deployment’s persistence settings.

API reference

See harnest_postgres, harnest_redis for signatures, options, and return values.

Overview

Get started

Harnest Studio

Guides

API reference

Long-term memory

Connect your existing database

Write deliberately from a tool

Read, search, update, and forget

Retention and limits

Run trusted storage maintenance

Handle connection failures

Implement a custom provider

API reference

​Connect your existing database

​Write deliberately from a tool

​Read, search, update, and forget

​Retention and limits

​Run trusted storage maintenance

​Handle connection failures

​Implement a custom provider

​API reference

Connect your existing database

Write deliberately from a tool

Read, search, update, and forget

Retention and limits

Run trusted storage maintenance

Handle connection failures

Implement a custom provider

API reference