Python SDK

This page shows how to use dakera, the official Python client, against a Dakera v0.12.0 server: installing it, connecting, storing and recalling memories, sessions, text search, knowledge graphs and the admin calls. It is for Python developers adding agent memory to an application; every example uses methods available in dakera 0.13.1 and later (current release: 0.14.1).

PyPI · GitHub · Release history

Watch it run: installing the SDK and a store and recall round trip against a local server, recorded on a clean machine.

python3 -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install dakera              # sync client (requests)
pip install "dakera[async]"     # adds httpx for AsyncDakeraClient
pip show dakera                 # check the installed version

Client setup

Point the client at your Dakera server. Read the URL and key from the environment to keep them out of source code. The key is sent as Authorization: Bearer <key>.

import os
from dakera import DakeraClient

# export DAKERA_API_URL=http://<YOUR_SERVER_IP>:3000
# export DAKERA_API_KEY=your-api-key

client = DakeraClient(
    base_url=os.getenv("DAKERA_API_URL", "http://localhost:3000"),
    api_key=os.getenv("DAKERA_API_KEY"),
)

Memory operations

# Store: returns the stored memory as a dict
memory = client.store_memory(
    agent_id="my-agent",
    content="User is learning Rust",
    memory_type="semantic",          # episodic (default), semantic, procedural, working
    importance=0.8,
    tags=["learning"],
)
memory_id = memory["id"]

# Recall: RecallResponse with .memories (RecalledMemory objects)
response = client.recall(agent_id="my-agent", query="which programming language is the user learning?", top_k=10)
for m in response.memories:
    print(m.id, m.content, round(m.score, 2))

# Get, update and delete one memory
client.get_memory(agent_id="my-agent", memory_id=memory_id)
client.update_memory(agent_id="my-agent", memory_id=memory_id, content="User is learning Rust and Go")
client.update_importance(agent_id="my-agent", memory_ids=[memory_id], importance=0.9)
client.forget(agent_id="my-agent", memory_id=memory_id)

# Filtered search within one agent's memories
client.search_memories(agent_id="my-agent", query="rust", memory_type="semantic", top_k=10)

Document import (server v0.12.3+)

Import a PDF, DOCX, HTML, Markdown or text document as chunk memories of an agent, then recall it with citations. Text formats go through import_document(); an uploaded PDF or DOCX attachment through import_attachment_document(). Both return a job to wait for.

from dakera import document_chunk_meta

job = client.import_document(
    "support-bot",
    open("field-manual.md").read(),
    media_type="text/markdown",
    title="Field manual",
)
done = client.wait_for_document_job(job.job_id, "support-bot")

# Session context in its own list; memories holds at most top_k results, best first
response = client.recall("support-bot", "how do I install it?", top_k=5, neighborhood_mode="separate")
for m in response.memories:
    meta = document_chunk_meta(m)   # None unless the memory is a document chunk
    print(m.content, meta.title if meta else "", meta.heading_path if meta else "")

client.list_documents("support-bot")
client.delete_document(job.document_id, "support-bot")

store_memory() and batch items also take event_time (Unix seconds, an RFC 3339 string or a timezone-aware datetime): when the event happened, so imported histories reason about time from their own dates.

Sessions

# start_session returns the session as a dict
session = client.start_session(agent_id="my-agent", metadata={"task": "review"})

client.store_memory(agent_id="my-agent", content="Found 2 issues", session_id=session["id"])
client.end_session(session["id"], summary="Review complete")

sessions = client.list_sessions(agent_id="my-agent", active_only=True)

ChatMemorySession for chat loops

ChatMemorySession wraps the session and memory calls for an LLM chat loop: it opens a session, stores each turn (tagged with its role) in that session, recalls across all of the agent's memories, and ends the session when the block exits.

from dakera import DakeraClient, ChatMemorySession

client = DakeraClient("http://localhost:3000", api_key="my-key")

with ChatMemorySession.create(client, agent_id="chat-agent") as session:
    session.store("user", "What is the capital of France?")
    session.store("assistant", "The capital of France is Paris.")
    memories = session.recall("capital cities", top_k=5)   # list of RecalledMemory

# Async variant (requires dakera[async])
from dakera import AsyncDakeraClient, AsyncChatMemorySession

async with AsyncDakeraClient("http://localhost:3000", api_key="my-key") as client:
    async with await AsyncChatMemorySession.create(client, agent_id="chat-agent") as session:
        await session.store("user", "Remember: the meeting is at 3pm")
        memories = await session.recall("meeting time")

Text documents and search

These calls work on a namespace of documents rather than an agent's memories. A namespace holds vectors of one dimension; the text routes embed with the server's model.

# Upsert text documents (embedded server-side)
client.upsert_text("docs", documents=[
    {"id": "doc-001", "text": "Quarterly revenue report Q1 2026", "metadata": {"source": "finance"}},
])

# Semantic query
results = client.query_text("docs", text="revenue trends", top_k=5)
for r in results.results:
    print(r.id, r.score)

# Several queries in one call
batch = client.batch_query_text("docs", queries=["revenue report", "product roadmap"], top_k=3)

# Hybrid search (BM25 + vector); vector_weight 0.0 = BM25 only, 1.0 = vector only
hits = client.hybrid_search("docs", query="revenue", top_k=10, vector_weight=0.7)

# Full-text search (BM25 only)
hits = client.fulltext_search("docs", query="quarterly report", top_k=5)

Knowledge graph

# Graph around a seed memory
graph = client.knowledge_graph(agent_id="my-agent", memory_id="mem-abc", depth=2)

# Path between two memories
path = client.knowledge_path(agent_id="my-agent", from_id="mem-a", to_id="mem-b")

# Cross-agent network (admin key with no namespace restriction on v0.12.0)
network = client.cross_agent_network(agent_ids=["agent-a", "agent-b"])

# Find duplicates (preview first)
dupes = client.deduplicate(agent_id="my-agent", threshold=0.93, dry_run=True)

Namespaces, keys and memory policy

# Create a namespace
client.create_namespace("production", dimensions=1024)

# Create an API key scoped to one namespace
key = client.create_namespace_key("production", name="reader", expires_in_days=90)

# Memory lifecycle policy: read, change, write back
policy = client.get_memory_policy("production")
policy.episodic_ttl_seconds = 604800   # 7 days
client.set_memory_policy("production", policy)

Entity extraction

# Extract entities from text without storing it
entities = client.extract_entities("Alice met Bob at Google HQ on Tuesday",
                                   entity_types=["person", "organization", "location"])

# Extract entities automatically for memories stored in a namespace
client.configure_namespace_ner("chat", extract_entities=True,
                               entity_types=["person", "organization", "location"])
config = client.get_namespace_entity_config("chat")

Import and export

# Export an agent's memories (jsonl, mem0, zep or csv)
export = client.export_memories(format="jsonl", agent_id="my-agent")

# Import from Mem0 format
client.import_memories(mem0_records, format="mem0", agent_id="my-agent")

Feedback

# Signal: upvote, downvote or flag
client.feedback_memory(memory_id="mem-abc", agent_id="my-agent", signal="upvote")

# Set importance directly
client.patch_memory_importance(memory_id="mem-abc", agent_id="my-agent", importance=0.9)

# Feedback summary for an agent
summary = client.get_agent_feedback_summary("my-agent")

Batch operations

from dakera import BatchForgetRequest, BatchMemoryFilter, BatchRecallRequest

# List memories by filter (no embedding)
resp = client.batch_recall(BatchRecallRequest(
    agent_id="my-agent",
    filter=BatchMemoryFilter(tags=["learning"], min_importance=0.6),
    limit=50,
))
print(resp.filtered, "matching memories")

# Delete memories by filter (at least one filter is required)
resp = client.batch_forget(BatchForgetRequest(
    agent_id="my-agent",
    filter=BatchMemoryFilter(tags=["obsolete"], max_importance=0.3),
))
print(f"Deleted {resp.deleted_count} memories")

AutoPilot and decay (admin)

status = client.autopilot_status()
client.autopilot_update_config(enabled=True, dedup_threshold=0.93, dedup_interval_hours=6)
client.autopilot_trigger("dedup")                       # dedup, consolidate or all

decay = client.decay_config()
client.decay_update_config(strategy="exponential", half_life_hours=168, min_importance=0.01)

Health probes

client.health()        # GET /health — status, version, degraded components
client.health_ready()  # GET /health/ready — succeeds once models are loaded and storage answers
client.health_live()   # GET /health/live — the process is up

Vector bulk operations

# Update metadata of every vector matching a filter
client.bulk_update_vectors("docs", filter={"source": {"$eq": "finance"}}, update={"reviewed": True})

# Delete every vector matching a filter
client.bulk_delete_vectors("docs", filter={"source": {"$eq": "drafts"}})

# Count vectors, optionally filtered
count = client.count_vectors("docs")

Full-text index management

stats = client.fulltext_stats("docs")
print(f"Documents: {stats.document_count}, terms: {stats.unique_terms}")

client.fulltext_delete("docs", ids=["doc-001"])

TTL and storage

ttl = client.ttl_stats()
print(f"With TTL: {ttl.total_with_ttl}, expired: {ttl.total_expired}")

tiers = client.storage_tier_overview()
type_stats = client.memory_type_stats()   # episodic, semantic, procedural, working

Agent consolidation

client.consolidate_agent("my-agent")
log = client.get_consolidation_log("my-agent")
client.patch_consolidation_config("my-agent", enabled=True)   # also: epsilon, min_samples, soft_deprecation_days

Namespace configuration

config = client.get_namespace_entity_config("production")
extractor = client.get_namespace_extractor("production")

# Re-embed namespaces at a new dimension (admin)
client.migrate_namespace_dimensions(target_dimension=1024, namespaces=["production"])

Admin and operations

On v0.12.0 these calls need an admin key with no namespace restriction (or authentication off); downloading, uploading or restoring a backup needs super_admin. See Upgrading to v0.12.

cluster = client.cluster_status()

# Maintenance mode
client.admin_enable_maintenance(reason="scheduled upgrade")
client.admin_disable_maintenance()

# Namespace quotas
quotas = client.admin_list_quotas()
client.admin_set_quota("production", {"max_vectors": 100000})

# Slow queries
slow = client.slow_queries(limit=20)

# Backups
backup = client.create_backup(include_data=True)
backups = client.list_backups()

# Diagnostics, background jobs, compaction, graceful shutdown
diag = client.ops_diagnostics()
jobs = client.ops_list_jobs()
client.ops_compact(namespace="production")
client.ops_shutdown()

Server-sent events

# Memory lifecycle events (stored, recalled, forgotten, …)
for event in client.stream_memory_events():
    print(event.event_type, event.agent_id, event.memory_id)

# Audit events, optionally filtered by agent
for event in client.stream_audit_events(agent_id="my-agent"):
    print(event.type, event.message)

On v0.12.0 both streams need an admin key with no namespace restriction.

Async client

AsyncDakeraClient (installed with dakera[async]) has the same methods as DakeraClient, as coroutines.

from dakera import AsyncDakeraClient

async with AsyncDakeraClient("http://localhost:3000", api_key="key") as client:
    await client.store_memory(agent_id="my-agent", content="async memory")
    response = await client.recall(agent_id="my-agent", query="async")

Error handling

from dakera import DakeraError, NotFoundError, RateLimitError

try:
    memory = client.get_memory(agent_id="my-agent", memory_id="mem-abc")
except NotFoundError:
    print("Memory not found")
except RateLimitError as e:
    print(f"Rate limited — retry after {e.retry_after}s")
except DakeraError as e:
    print(f"Error {e.status_code} ({e.code}): {e.message}")

Exceptions: DakeraError (base, with status_code, code, message), ConnectionError, NotFoundError, ValidationError, RateLimitError, ServerError, AuthenticationError, AuthorizationError, and TimeoutError (in dakera.exceptions).

Connection configuration

from dakera import DakeraClient, RetryConfig

client = DakeraClient(
    base_url="http://localhost:3000",
    api_key="your-key",
    timeout=30.0,                 # seconds per request (default 30.0)
    retry_config=RetryConfig(
        max_retries=3,            # defaults shown
        base_delay=0.1,
        max_delay=60.0,
        jitter=True,
    ),
)

See also: REST API reference · Quick Start · LangChain integration · CrewAI integration · LlamaIndex integration