MemoryClient API Reference

The main entry point for all memory operations.

Overview

MemoryClient manages the Neo4j connection and provides access to all three memory types.

from neo4j_agent_memory import MemoryClient, MemorySettings

settings = MemorySettings(
    neo4j={"uri": "bolt://localhost:7687", "password": "password"}
)

async with MemoryClient(settings) as client:
    # Access memory types
    client.short_term  # ShortTermMemory
    client.long_term   # LongTermMemory
    client.reasoning   # ReasoningMemory

Constructor

MemoryClient(
    settings: MemorySettings,
    embedder: Embedder | None = None,
    extractor: EntityExtractor | None = None,
    resolver: EntityResolver | None = None,
)

Parameters

Parameter Type Description

settings

MemorySettings

Configuration settings (required)

embedder

Embedder

Custom embedder (optional, created from settings if not provided)

extractor

EntityExtractor

Custom extractor (optional, created from settings if not provided)

resolver

EntityResolver

Custom resolver (optional, created from settings if not provided)

Properties

Property Type Description

short_term

ShortTermMemory

Access to conversation and message operations

long_term

LongTermMemory

Access to entity, preference, and fact operations

reasoning

ReasoningMemory

Access to reasoning trace operations

settings

MemorySettings

Current configuration settings

graph

Neo4jClient

Access the underlying Neo4j graph client for custom Cypher queries

Methods

connect

Establish connection to Neo4j.

await client.connect()
Called automatically when using async with.

close

Close the Neo4j connection.

await client.close()
Called automatically when exiting async with.

graph

Access the underlying Neo4jClient for custom Cypher queries alongside memory operations. This allows applications to query domain-specific data stored in the same Neo4j database without creating a separate database connection.

async with MemoryClient(settings) as client:
    # Run custom Cypher queries on the same connection used for memory
    results = await client.graph.execute_read(
        "MATCH (c:Customer {id: $id}) RETURN c.name AS name",
        {"id": "CUST-001"}
    )

    # Write operations
    await client.graph.execute_write(
        "CREATE (n:MyNode {name: $name})",
        {"name": "example"}
    )

The Neo4jClient provides:

  • execute_read(query, parameters) - Run read-only Cypher queries

  • execute_write(query, parameters) - Run write Cypher queries

  • execute_batch(queries) - Run multiple queries in a single transaction

  • vector_search(…​) - Run vector similarity searches

Raises NotConnectedError if the client is not connected. Always use within async with or after calling connect().

get_context

Get combined context from all memory types for LLM prompts.

context = await client.get_context(
    query: str,
    session_id: str | None = None,
    user_id: str | None = None,
    limit: int = 10,
    include_short_term: bool = True,
    include_long_term: bool = True,
    include_reasoning: bool = True,
)

Parameters

Parameter Type Default Description

query

str

required

Search query for semantic matching

session_id

str

None

Filter to specific session

user_id

str

None

Filter to specific user

limit

int

10

Maximum items per memory type

include_short_term

bool

True

Include recent messages

include_long_term

bool

True

Include entities and preferences

include_reasoning

bool

True

Include similar reasoning traces

Returns

MemoryContext object containing:

@dataclass
class MemoryContext:
    messages: list[Message]
    entities: list[Entity]
    preferences: list[Preference]
    reasoning_traces: list[ReasoningTrace]

    def to_prompt(self) -> str:
        """Format context for LLM prompt."""
        ...

Example

context = await client.get_context(
    "restaurant recommendations",
    session_id="user-123",
    limit=5,
)

# Use in LLM prompt
prompt = f"""
Context from memory:
{context.to_prompt()}

User question: What's a good Italian restaurant?
"""

get_stats

Get memory statistics.

stats = await client.get_stats()

Returns

@dataclass
class MemoryStats:
    short_term: ShortTermStats
    long_term: LongTermStats
    reasoning: ReasoningStats

@dataclass
class ShortTermStats:
    conversation_count: int
    message_count: int

@dataclass
class LongTermStats:
    entity_count: int
    entity_counts_by_type: dict[str, int]
    preference_count: int
    fact_count: int

@dataclass
class ReasoningStats:
    trace_count: int
    step_count: int
    tool_call_count: int

export_graph

Export memory graph as a subgraph.

graph = await client.export_graph(
    session_id: str | None = None,
    include_entities: bool = True,
    include_messages: bool = True,
    include_reasoning: bool = True,
)

Returns

Dictionary with nodes and relationships lists.

Context Manager

The recommended way to use MemoryClient is as an async context manager:

async with MemoryClient(settings) as client:
    # Connection is established
    await client.short_term.add_message(...)
# Connection is closed automatically

This ensures proper connection handling and resource cleanup.

Thread Safety

MemoryClient is not thread-safe. Create separate instances for concurrent access:

# In async context - OK (single thread)
async with MemoryClient(settings) as client:
    await asyncio.gather(
        client.short_term.add_message(...),
        client.long_term.add_entity(...),
    )

# For multiple threads - create separate clients
async def worker(settings):
    async with MemoryClient(settings) as client:
        await client.short_term.add_message(...)

See Also