MemgraphAI

Python SDK

The official Python SDK for Memgraph. Available on PyPI as memgraph-sdk (v0.7.2).

Installation

bash
# Sync client (requests-based)
pip install memgraph-sdk

# With async support (adds httpx)
pip install memgraph-sdk[async]

Requires Python 3.9+.

Sync Client

The MemgraphClient uses the requests library. Best for scripts, CLI tools, and synchronous applications.

python
from memgraph_sdk import MemgraphClient

client = MemgraphClient(
    api_key="mg_your_api_key",
    tenant_id="your-tenant-uuid",
    base_url=None,       # defaults to MEMGRAPH_API_URL env var
    timeout=30.0,        # request timeout in seconds
    max_retries=3,       # automatic retries with exponential backoff
)

Core methods

ping()→ Dict

Check server connectivity and return health status.

python
status = client.ping()
# {"status": "healthy", "database": "connected"}
remember(text, user_id, category="general", domain=None, confidence=0.90)→ Dict

Store a memory as a belief with vector embedding. Immediately searchable. Use this when you want the fact available right away.

python
client.remember(
    text="User prefers dark mode",
    user_id="alice",
    category="preference",  # decision, architecture, bug_fix, preference, general
    domain="settings",       # optional domain categorization
    confidence=0.95,
)
add(text, user_id)→ Dict

Add a raw event to the memory pipeline. Goes through event → episode → belief extraction (asynchronous). Use this for raw interaction logging.

python
client.add(
    text="User asked about Kubernetes deployment",
    user_id="alice",
)
search(query, user_id, agent_id=None, limit=10)→ Dict

Search memories relevant to a query. Returns scored results with content, relevance score, and metadata.

python
result = client.search(
    query="What does Alice prefer?",
    user_id="alice",
)
# result["results"] is a list of {content, score, metadata}
for r in result["results"]:
    print(f'{r["content"]} (score: {r["score"]:.2f})')
get_beliefs(user_id, limit=50, cursor=None)→ Dict

Fetch beliefs for a user with cursor-based pagination.

python
beliefs = client.get_beliefs(user_id="alice", limit=20)
# beliefs["items"] contains the belief objects
# beliefs["next_cursor"] for pagination

Memory Intelligence

Methods for monitoring memory quality and health.

health(user_id=None)→ Dict

Get memory health metrics: belief counts, confidence distribution, domain breakdown, contradiction rate, staleness rate.

python
health = client.health()
# {"total_beliefs": 42, "avg_confidence": 0.87, "contradiction_rate": 0.02}
contradictions(user_id=None)→ Dict

Get contradiction report showing conflicting beliefs.

python
report = client.contradictions(user_id="alice")
mcis(user_id=None, save=False)→ Dict

Compute the Memgraph Cognitive Integrity Score (0-100). A composite score of memory quality.

python
score = client.mcis(save=True)
# {"score": 85, "breakdown": {"accuracy": 90, "completeness": 80, ...}}
evaluate(query, user_id)→ Dict

Run a retrieval query with detailed scoring breakdown — useful for debugging search quality.

python
result = client.evaluate(query="deployment preferences", user_id="alice")
benchmark(scenario)→ Dict

Run a memory benchmark scenario against your tenant's data.

python
result = client.benchmark(scenario="preference_recall")

Async Client

The AsyncMemgraphClient uses httpx for non-blocking HTTP calls. Ideal for FastAPI, aiohttp, or high-throughput pipelines.

python
from memgraph_sdk import AsyncMemgraphClient

async with AsyncMemgraphClient(
    api_key="mg_your_key",
    tenant_id="your-tenant-uuid",
) as client:
    # Same API as sync client, but all methods are awaitable
    await client.remember("User prefers Python 3.12", user_id="alice", category="preference")
    context = await client.search("What language does Alice prefer?", user_id="alice")
    health = await client.health()

Requires pip install memgraph-sdk[async] to install the httpx dependency. The async client has the same methods as the sync client.

CLI Tool

The SDK includes a memgraph command-line tool for quick interactions.

bash
# Initialize configuration (saves to ~/.memgraph/config.json)
memgraph init --api-key mg_your_key --tenant-id your-tenant-uuid

# Store a memory as a belief
memgraph remember "User prefers dark mode" --user alice --category preference

# Search memories
memgraph recall "What does alice prefer?" --user alice

# Check server connection
memgraph status

Cognitive Sidecar

Always-on memory layer that auto-recalls before every LLM call and auto-learns after. See the Sidecar guide for full details.

python
from memgraph_sdk.middleware import CognitiveSidecar

sidecar = CognitiveSidecar(
    client=client,
    user_id="alice",
    agent_id="support-bot",
    token_budget=4000,
    auto_learn=True,
)

# Wrap your messages — memory is injected automatically
messages = [{"role": "user", "content": "How should I deploy?"}]
enriched, learn = sidecar.wrap(messages)

# Call your LLM with enriched messages...
# Then call learn(response_text) to store learnings

Sidecar methods

pre_flight(messages)→ List[Dict]

Inject memory context as a system message before your LLM call.

python
enriched = sidecar.pre_flight(messages)
print(f"Found {sidecar.memories_found} memories")
post_flight(messages)→ None

Extract and store learnings from a message exchange.

python
sidecar.post_flight(full_messages)
wrap(messages)→ Tuple[List, Callable]

Pre-flight + returns a learn callback for post-flight.

python
enriched, learn = sidecar.wrap(messages)
# ... call LLM ...
learn(response_text)
process(messages)→ List[Dict]

Combined pre-flight + post-flight in a single API call.

python
enriched = sidecar.process(messages)

v2 Cognitive Infrastructure

Methods for entities, relationships, decisions, and graph traversal. See the v2 API docs for endpoint details.

Context Graph

get_context_graph(query, user_id=None, agent_id=None, include_decisions=True, include_graph=True)→ Dict

6-stage context retrieval: semantic → entity resolution → graph expansion → temporal filter → confidence weight → assembly.

python
context = client.get_context_graph(
    query="What does Alice prefer?",
    user_id="alice",
    include_decisions=True,
)

Entities

create_entity(entity_type, name=None, properties=None, aliases=None)→ Dict

Create a named entity in the knowledge graph.

python
entity = client.create_entity(
    entity_type="person",
    name="Alice Chen",
    properties={"role": "engineer"},
    aliases=["alice"],
)
search_entities(query, max_results=5)→ List[Dict]

Fuzzy + semantic entity search.

python
results = client.search_entities("Alice", max_results=5)

Decisions

record_decision(goal, reasoning_steps=None, tools_used=None, beliefs_used=None, confidence=None, outcome=None, agent_id=None, user_id=None)→ Dict

Record a decision with full reasoning trace.

python
decision = client.record_decision(
    goal="Recommend deployment strategy",
    agent_id="bot",
    confidence=0.85,
    outcome="SUCCESS",
)
explain_decision(decision_id)→ Dict

Get full explanation: beliefs, context snapshot, entity graph at decision time.

python
explanation = client.explain_decision("decision-uuid")
list_decisions(agent_id=None, user_id=None, outcome=None, limit=50)→ List[Dict]

List decisions with optional filters.

python
decisions = client.list_decisions(agent_id="bot", outcome="FAILURE")

Relationships & Graph

create_relationship(source_entity_id, target_entity_id, relation_type, confidence=1.0)→ Dict

Create a temporal relationship between two entities.

python
rel = client.create_relationship(
    source_entity_id="entity-1",
    target_entity_id="entity-2",
    relation_type="works_with",
)
traverse_graph(entity_ids, max_depth=2, temporal_filter=True)→ Dict

Traverse entity graph from seed entities.

python
graph = client.traverse_graph(entity_ids=["entity-1"], max_depth=2)

Contradictions

list_contradictions(status=None, limit=50)→ List[Dict]

List detected contradictions between beliefs.

python
contradictions = client.list_contradictions(status="active")
resolve_contradiction(contradiction_id, resolution_status="resolved", resolved_by="manual")→ Dict

Resolve or dismiss a contradiction.

python
client.resolve_contradiction("contradiction-uuid")

OpenAI Agents SDK

First-class integration with the OpenAI Agents SDK. Auto-inject memory into agent instructions and capture decisions with lifecycle hooks.

python
from memgraph_sdk import MemgraphClient
from memgraph_sdk.openai_agents import create_memgraph_agent, MemgraphAgentHooks
from agents import Runner

client = MemgraphClient(api_key="mg_your_key")

# All-in-one: agent with memory-enriched instructions + decision capture
agent = create_memgraph_agent(
    client=client,
    user_id="alice",
    name="Support Agent",
    instructions="You are a helpful support agent.",
    agent_id="support-bot",
    include_memory_tools=True,  # adds search_memory + store_memory tools
    token_budget=4000,
)

result = await Runner.run(agent, "How should I deploy to production?")

See the Integrations page for more OpenAI Agents patterns including hooks and multi-agent orchestration.

Memory Management

forget(belief_id)→ Dict

Delete a specific belief by ID.

python
client.forget("belief-uuid")
forget_all(user_id, domain=None, soft=False)→ Dict

Bulk delete beliefs for a user. Use soft=True for soft delete.

python
client.forget_all(user_id="alice", domain="settings")
belief_history(user_id, key, as_of=None, include_inactive=True)→ List[Dict]

Version history for a specific belief key.

python
history = client.belief_history(user_id="alice", key="preferred_language")
belief_timeline(user_id, domain=None, since=None, limit=50)→ Dict

Chronological timeline of all belief changes.

python
timeline = client.belief_timeline(user_id="alice", since="2025-01-01")

MCP Server

Give Claude Code, Cursor, and other MCP clients persistent memory. See the MCP guide for full setup.

bash
# One-command setup
memgraph setup --key mg_your_api_key

Error Handling

The SDK uses a typed exception hierarchy for clean error handling:

python
from memgraph_sdk import (
    MemgraphError,            # Base exception
    MemgraphConnectionError,  # Cannot reach server
    MemgraphAuthError,        # 401/403 - bad API key or token
    MemgraphValidationError,  # 400-499 - bad request data
    MemgraphRateLimitError,   # 429 - rate limit exceeded
    MemgraphAPIError,         # 500+ - server error
)

try:
    client.remember("test", user_id="alice")
except MemgraphAuthError:
    print("Check your API key")
except MemgraphRateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except MemgraphConnectionError:
    print("Cannot reach server")
except MemgraphError as e:
    print(f"Memgraph error: {e}")

Automatic retries: The SDK retries on rate limits (429), server errors (5xx), and connection errors with exponential backoff. Auth errors (401/403) and validation errors (4xx) are not retried.

Environment Variables

VariableDefaultDescription
MEMGRAPH_API_URLhttps://api.memgraph.ai/v1Base URL for the Memgraph API
MEMGRAPH_API_KEY-API key (used by MCP server)
MEMGRAPH_TENANT_ID-Tenant UUID (used by MCP server)