CortexDB Python SDK

The Long-Term Memory Layer for AI Systems.

CortexDB gives your AI agents persistent, structured memory. Capture experiences, then recall a stratified pack (events, facts, beliefs, episodes, concepts) or get a grounded answer — all scoped to a hierarchical scope path.

This SDK targets the v1 API (/v1/experience, /v1/recall, /v1/answer, …), PASETO auth, and hierarchical scopes. The pre-0.3 remember()/tenant_id client was retired — Cortex is now an alias of V1Client.

Installation

pip install cortexdbai

Quick start

import os
from cortexdb import Cortex

# Authenticated local Docker instance (set CORTEX_API_KEY on the server too):
with Cortex("http://localhost:3141", bearer=os.environ["CORTEX_API_KEY"]) as c:
    c.experience(
        "ws:demo",
        text="Priya at Acme signed for 200 seats. Q3 close.",
        wait="indexed",
    )
    pack = c.recall("ws:demo", query="How many seats did Acme sign for?")
    print(pack["context_block"])

Hosted / multi-tenant (PASETO bearer + actor)

with Cortex(
    "https://api-v1.cortexdb.ai",
    actor="user:alice",
    bearer="v4.public…",          # PASETO v4 public or JWT (RS256/ES256)
) as c:
    out = c.answer(
        "org:initech/user:alice",
        "What did Alice say about coffee?",
    )
    print(out["answer"])

One-call signup (no prior credentials)

from cortexdb import Cortex

c = Cortex.signup()                      # POST /v1/auth/signup → free-tier PASETO
c.experience(c.actor, text="hello, memory!", wait="indexed")
print(c.recall(c.actor, query="what did I say?")["context_block"])

Core operations

# Write — async by default (202); use wait="indexed" for read-your-writes.
c.experience("ws:demo", text="Deployed payments-service v3.1 to prod.", wait="indexed")
c.experience("ws:demo", text="Rolled back auth-gateway.", wait="indexed")

# Large content: wait="captured" returns at the durable accept point (~10ms);
# poll write_status to see indexing progress. After a client timeout,
# found=False or status="failed" both mean re-sending is safe (re-processes).
c.experience("ws:demo", text=big_transcript, idempotency_key="k1", wait="captured")
c.write_status(idempotency_key="k1")  # {found, status, stages_completed, ...}

# Preserve exact source structure for citations, replay, and artifact building.
from cortexdb import sha256_fingerprint

text = "Customer approved the renewal."
fingerprint = sha256_fingerprint(text)
c.experience(
    "ws:demo",
    text=text,
    modality="conversation",
    source={
        "schema_version": 2,
        "source_id": "crm:thread:42",
        "modality": "conversation",
        "content_kind": "message",
        "source_fingerprint": fingerprint,
        "fragments": [{
            "fragment_id": "message-7",
            "locator": {
                "kind": "conversation",
                "session_id": "thread-42",
                "turn_index": 7,
            },
            "content_fingerprint": fingerprint,
        }],
    },
)

# Losslessly split a larger logical text source at UTF-8 boundaries. The
# response includes every fragment event ID; no existing one-event API changes.
source_write = c.experience_source_text(
    "ws:demo",
    handbook_text,
    source_id="document:employee-handbook",
    locator_path=["employee-handbook", "body"],
)
assert len(source_write["event_ids"]) == source_write["fragment_count"]
# The server retains replay keys for 24 hours. For longer jobs, persist each
# `on_fragment` checkpoint and pass it back through `completed_fragments`.

# Recall — returns a StratifiedPack { context_block, layers, provenance }.
pack = c.recall("ws:demo", query="recent deploys", view="holistic")

# Answer — recall + LLM render with citations.
ans = c.answer("ws:demo", "What changed in payments recently?")

# Forget — GDPR erasure. confirm_all wipes the whole scope.
c.forget("ws:demo", confirm_all=True, cascade="redact_events",
         reason="GDPR Article 17 request")

# Layer reads (paginated { items, has_more }).
c.events("ws:demo")
c.facts("ws:demo")
c.beliefs("ws:demo")
c.episodes("ws:demo")
c.understanding("ws:demo")

Async usage

AsyncCortex provides the async core surface, including write barriers:

import asyncio
import os
from cortexdb import AsyncCortex

async def main():
    async with AsyncCortex(
        "http://localhost:3141",
        bearer=os.environ["CORTEX_API_KEY"],
    ) as c:
        await c.experience(
            "ws:demo",
            text="async memory works great",
            wait="indexed",
        )
        pack = await c.recall("ws:demo", query="what works?")
        print(pack["context_block"])

asyncio.run(main())

Error handling

import os

from cortexdb import (
    Cortex,
    V1Error,             # base
    V1APIError,          # non-2xx envelope
    V1AuthError,         # 401 / actor mismatch
    V1PolicyDeniedError, # 403 — missing capability
    V1RateLimitError,    # 429
    V1NotConfiguredError,# 503 — feature not configured
    V1ConnectionError,
    V1TimeoutError,
)

with Cortex(
    "http://localhost:3141",
    bearer=os.environ["CORTEX_API_KEY"],
) as c:
    try:
        c.experience("ws:demo", text="important data", wait="indexed")
    except V1RateLimitError as e:
        print(f"rate limited (HTTP {e.status}); retry using your backoff policy")
    except V1PolicyDeniedError as e:
        print(f"missing capability: {e}")
    except V1APIError as e:
        print(f"API error [{e.status}]: {e}")

Integrations

Framework and provider adapters ship as optional extras of the same cortexdbai package — there are no separate cortexdb-<framework> packages to install. Pick an extra and import from cortexdb.integrations.<submodule>:

pip install cortexdbai[langgraph]      # or crewai, llamaindex, autogen, …

from cortexdb.integrations.langgraph import CortexDBCheckpointer, cortexdb_memory_node

Available extras (pip install cortexdbai[<extra>]):

Constructor

Links

• Documentation
• Support

License

Apache-2.0