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)

The client acts as its token's subject: X-Cortex-Actor is the token's sub unless you pass actor (actor=None sends no header, as a token whose subject the server maps itself needs).

with Cortex(
    "https://api-v1.cortexdb.ai",
    bearer="v4.public…",          # PASETO v4 public or JWT (RS256/ES256); acts as its sub
) 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, its
# `resend` says what to do: "no" means stored, and a write that failed indexing
# (status="failed") is stored too: never re-send it, at any age or under any
# key (a new key would store it twice); the operator re-includes it in place.
# "if_sent_within_24h" means no receipt: re-send if you sent it within a day.
c.experience("ws:demo", text=big_transcript, idempotency_key="k1", wait="captured")
c.write_status(idempotency_key="k1")  # {found, status, stages_completed, resend, note}

# 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 has every method Cortex has, under the same name, with the same parameters; each is a coroutine (streams and pages are async iterators):

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

Every refusal raises the class of its error_code, from cortexdb.errors (one per code: WAIT_TIMEOUT raises errors.WaitTimeoutError), which subclasses the class of its HTTP status family: V1BadRequestError (400, and V1ValidationError for 422), V1AuthError (401), V1PaymentRequiredError (402), V1PolicyDeniedError (403), V1NotFoundError (404), V1ConflictError (409, and V1PreconditionFailedError for 412), V1GoneError (410), V1PayloadTooLargeError (413), V1RateLimitError (429), V1ServerTimeoutError (408 and 504), V1ServerError (5xx) and V1UnavailableError (503, with V1NotConfiguredError for a server missing a provider). All of them are V1APIErrors, with status, error_code, message, request_id, retriable (sending the same request again can succeed), details, headers and retry_after (the Retry-After header in seconds); each per-code class reads its documented details keys as properties (err.waiting_on, err.limit, …). A code this SDK does not know yet raises its family's class.

import os
import time

from cortexdb import Cortex, V1APIError, V1PolicyDeniedError, V1UnavailableError, errors

with Cortex("http://localhost:3141", bearer=os.environ["CORTEX_API_KEY"]) as c:
    try:
        c.experience("ws:demo", text="important data", wait="indexed")
    except errors.WaitTimeoutError as e:        # captured; the barrier ran out
        print(f"still {e.waiting_on} after {e.waited_ms} ms")
    except V1UnavailableError as e:              # any 503
        if e.retriable:
            time.sleep(e.retry_after or 1)
    except V1PolicyDeniedError as e:             # any 403
        print(f"refused: {e.error_code} {e.details}")
    except V1APIError as e:                      # anything else the server refused
        print(f"API error [{e.status} {e.error_code}]: {e.message}")

V1ConnectionError and V1TimeoutError (the client's own timeout) are the transport's; V1InvalidRequestError is a request the SDK refused before sending it, with the code the server would answer. A value a route always refuses (an extra_scopes list past 40, a malformed break-glass header, a mint the server never accepts) raises ValueError before anything is sent.

What's new in 0.13.0 for CortexDB 0.14.0

cortexdbai 0.13.0 covers the whole v1 API of CortexDB 0.14.0 (COVERAGE.md: every route and the method that calls it), on both clients under the same names.

• Every route has a method, on Cortex and AsyncCortex alike: the access API (access_check, access_explain, the lookups, access_limits), AuthZEN 1.0, change sets (changeset_apply and its steps), the access watch and notices, applications, the directory, SCIM (client.scim), restores, source containers (rehome, watermarks, account resolution), rollups and publish, person, party and tenant erasures with receipts and proofs, the read-free forget, the operator's access routes, and the routes older releases left out (lifecycle, vocabularies, imports, export, keyed blobs, the cohort lanes, audit, retention, the code plane, the native memory host's).
• Typed models for every request and response (cortexdb.v1.models, TypedDicts generated from the server's OpenAPI contract).
• A typed exception per error code (cortexdb.errors), under one class per HTTP status family, with retriable, details (named properties for each documented key) and retry_after.
• Consistency tokens: consistency=True on writes and reads, consistency=<token> or at_least=<token> to demand one, and ConsistencyTokens to carry them per root.
• The new read and write fields: extra_scopes (up to 40), coverage, beliefs="merged", mounts="default", purpose; provenance, intent, subject and the envelope's other fields; streams as StreamEvent iterators; a declared belief's idempotency_key.
• Paging and by-id reads: every listing pages with pages() (events newest first by their le1. cursor, the scope list, the layers); scope= on the by-id layer reads (a scope that does not hold the row is 404); include_vectors=True returns embeddings; the timeline takes as_of.
• Auth: the actor follows the token, mint_token's chain fields with mint_agent_token and mint_app_token, break-glass (client.break_glass) and the subject-token shadow (with_options(subject_token=...)).

Its breaking changes are below. Each feature has a section under Shared memory with an example the SDK's live tests run against a 0.14.0 server.

Breaking changes in 0.13.0

Upgrading from cortexdbai 0.12: every call keeps its name, its positional parameters and every keyword it took; these behave differently.

• Errors are typed per code. Every refusal raises the class of its error_code (cortexdb.errors), a subclass of its HTTP status family's class. A code this SDK does not know raises its family's class (a 503 is a V1UnavailableError), no longer a bare V1APIError: an except V1APIError keeps catching everything, but a type(err) is V1APIError check no longer matches. V1NotConfiguredError is now a V1UnavailableError, and V1InvalidRequestError a V1BadRequestError. For a body without retriable, retriable is the code's documented default instead of False.
• The actor follows the token. A client built with a token bearer and no actor sends the token's sub as X-Cortex-Actor (0.12 sent user:default, which the server refuses with 401 ACTOR_MISMATCH for any other subject), and no header for a token whose subject an issuer maps. The operator key and no bearer still send user:default; pass actor to choose.
• view is the server's. recall, recall_stream, answer and compose send view only when you pass it (0.12 sent "holistic"). The server reads holistic as before, but granular at an org tree's shared scope (team, org, group, compartment), where holistic is 422 VIEW_NOT_ALLOWED_AT_SCOPE.
• Local refusals. A request the server always refuses fails before it is sent, with ValueError: an extra_scopes list past 40, a malformed break-glass header, a mint whose chain fields break mint_chain_v1's rules, more than 1,024 provenance ids.
• Server 0.14.0 refuses what 0.13.x served (its own release notes list them): forget and erasure selectors are strict (422 INVALID_BODY, an unknown selector field or one the cascade would not apply, field_not_applied), every mint names its tenant root (scope, tenant or scopes; only the trusted operator mints deployment-wide, and only for a service: or system: subject), mints are held to their lifetime ceilings, a token that carries scopes must confine every capability it uses, and a tenant in deny refuses the operator key's data reads without break-glass. A header that does not parse is 400 INVALID_HEADER (a malformed break-glass header was BREAK_GLASS_INVALID, which has no class any more).

Shared memory (CortexDB 0.14.0)

CortexDB 0.14.0 adds org trees with real access control, and the routes that run them: change sets to register an org's records, an access API that says who may read what and why, AuthZEN for policy enforcement points, a watch of every access change, notices, consistency tokens, applications, a directory of people, SCIM, rollups, publish, and person and tenant erasures. Every method below exists on Cortex (V1Client) and, with await, on AsyncCortex (AsyncV1Client), under the same name.

The examples run, in order, against a fresh server (the SDK's live tests run each of them). Start one with the operator key, the minter, and demo as the deployment's tenant, so a token minted without tenant belongs to org:demo:

CORTEX_API_KEY=demo-operator-key CORTEX_V1_MINTER_ENABLE=1 \
CORTEX_V1_AUDIENCE=cortexdb:tenant:demo ./cortexdb

import os

from cortexdb import Cortex

URL = os.environ.get("CORTEX_URL", "http://localhost:3141")
ROOT = os.environ.get("CORTEX_DEMO_ROOT", "org:demo")
ENG = f"{ROOT}/team:eng"

# The deployment's operator key. `actor=None` sends no X-Cortex-Actor: the key
# then acts as itself.
operator = Cortex(URL, bearer=os.environ.get("CORTEX_API_KEY", "demo-operator-key"), actor=None)

Change sets: registering an org tree

An org's records (its org, teams, sets, people's private regions, sources …) are written in change sets: staged, previewed (who gains and loses what), then activated in one step. A compiler stages them under its own principal; here the operator key acts as service:compiler. scope_record, member and set_member build the version-2 records; a record holds no names (those live in the directory, below).

from cortexdb.v1 import changeset_part, member, put_record, scope_record, set_member

compiler = operator.with_options(actor="service:compiler")
owner = member("service:compiler", "owner")
records = [
    scope_record(ROOT, members=[owner, member("user:ada", "owner"), member("service:scim-okta", "owner")],
                 kind="org"),
    scope_record(f"{ROOT}/set:eng", members=[owner], kind="set", parent=ROOT,
                 set={"entries": [{"actor": "user:alice"}, {"actor": "user:bob"}]}),
    scope_record(ENG, members=[owner, set_member(f"{ROOT}/set:eng", "writer")],
                 kind="team", parent=ROOT, audience="members"),
    scope_record(f"{ROOT}/user:alice", members=[member("user:alice", "owner"), owner],
                 kind="person", parent=ENG, private=True, self_principal="user:alice"),
]
# Stage, preview and activate in one call ...
applied = compiler.changeset_apply(ROOT, "org-compile-1", records, reason="first compile")
print(applied["activated"]["state"])                 # activated

# ... or step by step, reading the preview before approving it.
sales = scope_record(f"{ROOT}/team:sales", members=[owner, member("user:carol", "writer")],
                     kind="team", parent=ROOT)
compiler.changeset_stage(ROOT, "sales-1", [changeset_part(0, [put_record(sales["path"], sales)])])
preview = compiler.changeset_preview("sales-1", ROOT, people=True)
print(preview["summary"]["people_gaining"], preview["widens"])
status = compiler.changeset_activate("sales-1", ROOT, preview["preview_hash"])
print(status["state"], status["consistency_token"][:4])  # activated ct1.

A refusal raises its typed error: PREVIEW_CHANGED (the registry moved since the preview), ACCESS_ASSERTION_FAILED (a widening breaks one of the tenant's policy cases), CHANGESET_STALE (a record changed after staging), ACCESS_LIMIT_REACHED. changeset_get and changeset_discard read and drop a staged change set.

Under a tenant whose operator mode is deny, the operator key's activation of a change set that registers a record, names a service: principal or widens who reads a private region is an in-band path. Its preview says so (in_band, and in_band_parts names each record and why), and the activation takes a break-glass client, else 403 BREAK_GLASS_REQUIRED: compiler.break_glass("support", "SUP-1142").changeset_activate(id, ROOT, preview["preview_hash"]). Each private region it widens is told first (503 NOTICE_UNAVAILABLE, nothing activated, when that notice cannot commit), and a registry change in between is 503 REGISTRY_STALE, which a retry decides again. Staging, the preview and a discard need no header.

The access API

access_check answers whether a principal chain may read, write or administer a scope, or read one memory, with the deciding path of every link; access_explain says why, atom by atom. Without chain, it decides for the caller's own token; another chain may be asked about by an Owner of the scope or the trusted operator. access_lookup_scopes lists what a chain may read under a prefix, access_lookup_principals who may read a scope, and access_limits the limits the deployment enforces (read them; never hard-code them).

check = operator.access_check(ENG, role="writer", chain={"sub": "user:alice"})
print(check["allowed"], check["delivered_role"])     # True writer
denied = operator.access_check(ENG, chain={"sub": "user:mallory"})
print(denied["denied_by"], denied["links"]["person"]["reason"])  # person not_a_member

why = operator.access_explain(ENG, chain={"sub": "user:alice"})
print(why["home"]["links"]["person"]["via"][0]["record"])        # org:demo/team:eng

for page in operator.pages(operator.access_lookup_scopes, ROOT, chain={"sub": "user:alice"}):
    for item in page["items"]:
        print(item["scope"], item["role"])           # org:demo/team:eng writer
readers = operator.access_lookup_principals(ENG)
print(sorted(i.get("principal") or i.get("set") for i in readers["items"]))

limits = {limit["name"]: limit["value"] for limit in operator.access_limits(ROOT)["limits"]}
print(limits["apps"], limits["scope_path_segments"])  # 8 32

AuthZEN 1.0

A third-party policy enforcement point asks the same questions through AuthZEN 1.0: authzen_evaluation (actions summary, read, write, own) and authzen_search_resource, with the PDP's metadata at authzen_configuration().

print(operator.authzen_evaluation("user:alice", "write", ENG))     # {'decision': True}
print(operator.authzen_evaluation("user:mallory", "read", ENG))
# {'context': {'denied_by': 'person'}, 'decision': False}
found = operator.authzen_search_resource("user:alice", "read", ROOT, limit=10)
print([r["id"] for r in found["results"]])
print(operator.authzen_configuration()["access_evaluation_endpoint"])

Watch and notices

access_watch follows a root's access changes (registry changes with their consistency tokens, change-set events, notices, assertion breaks and restores) from a position, within a 7-day window; wait_ms long-polls. A person reads the notices that name them (who read what of theirs, under a grant or a break-glass read) with access_notices and acknowledges them with access_notices_ack.

start = operator.access_watch(ROOT)                  # the current position, no history
ops = scope_record(f"{ROOT}/team:ops", members=[owner], kind="team", parent=ROOT)
compiler.changeset_apply(ROOT, "ops-1", [ops])
page = operator.access_watch(ROOT, after=start["position"], wait_ms=2000)
print(sorted({item["type"] for item in page["items"]}))  # ['changeset', 'registry']

alice = Cortex(URL, bearer=operator.mint_token(subject="user:alice", scope=ROOT)["token"])
notices = alice.access_notices(ROOT)
if notices["items"]:
    alice.access_notices_ack(ROOT, seqs=[n["seq"] for n in notices["items"]])

Consistency tokens

Every write and read can carry a consistency token: a lower bound, so a later read that demands it sees at least what that response saw or wrote, on any node. Pass consistency=True to a write (or a read) for a token, then consistency=<token> to a read, or at_least=<token> to a GET read. A ConsistencyTokens tracker does it for you, per scope root.

from cortexdb.v1 import ConsistencyTokens

receipt = alice.experience(ENG, text="Shipped payments v3.1 today", consistency=True)
pack = alice.recall(ENG, query="what shipped?", consistency=receipt["consistency_token"])
events = alice.events(ENG, at_least=pack["consistency_token"])

tokens = ConsistencyTokens()
session = alice.with_options(consistency=tokens)
session.experience(ENG, text="Rolled back auth-gateway", wait="captured")
pack = session.recall(ENG, query="what was rolled back?")  # waits for that write
print(sorted(tokens.as_dict()))                      # ['org:demo']

A demand the node cannot reach in time answers 408 WAIT_TIMEOUT (details.waiting_on: registry, indexer or access_view); a token of another root, or one this deployment never issued, 422 CONSISTENCY_TOKEN_INVALID.

Applications

An application record (<root>/app:<id>) names what an application writes and what its chains read by default: mounts, which mounts="default" on a read adds (each admitted by an apps entry of the record it mounts). A tenant holds at most limits.apps of them (access_limits). The record holds no name: app_put refuses display_name, and the application's name is the directory's name entry of its path (directory_put(f"{ROOT}/app:notes", ROOT, display_name="Notes")).

ada = Cortex(URL, bearer=operator.mint_token(subject="user:ada", scope=ROOT)["token"])
app = ada.app_put("notes", ROOT, kinds=["note", "decision"])
print(app["path"], app["home"])     # org:demo/app:notes org:demo/app:notes/user:{person}
print(ada.app_get("notes", ROOT)["kinds"])

The directory

Each tenant's people, with their names, emails and source accounts, and the name entries of its scopes and sets. Records hold no names: a scope's display name is its entry here. Writes can be conditional on the version last read.

operator.directory_put("user:alice", ROOT, display_name="Alice Moreau", emails=["Alice@Demo.Example"],
                       accounts=[{"system": "slack:T1", "account": "UALICE"}])
operator.directory_put(ENG, ROOT, display_name="Engineering")    # a scope's name entry
entry = operator.directory_get("user:alice", ROOT)
print(entry["emails"], entry["version"])                         # ['alice@demo.example'] 1
# A put replaces the whole entry; if_version makes it conditional on the version read.
operator.directory_put("user:alice", ROOT, display_name="Alice M.", emails=entry["emails"],
                       accounts=entry["accounts"], if_version=entry["version"])
print([e["id"] for e in operator.directory_search(ROOT, "ali")["items"]])
resolved = operator.access_resolve_accounts(ROOT, [("slack:T1", "UALICE"), ("slack:T1", "UNKNOWN")])
print([a["id"] for a in resolved["accounts"]])                    # ['user:alice', None]

A stale version is 412 DIRECTORY_VERSION_CHANGED; an account another principal holds 409 DIRECTORY_ACCOUNT_TAKEN. A deactivated entry (status="deactivated") refuses every credential of the person, 401 SUBJECT_UNMAPPED, until it is active again.

SCIM

An identity provider provisions the tenant's people and groups through SCIM 2.0, with a bearer for service:scim-<client> that is an Owner of the tenant root. client.scim sends its requests the way a provider does, without X-Cortex-Actor; scim_user, scim_group and scim_op build the resources.

from cortexdb.v1 import scim_group, scim_op, scim_user

idp = Cortex(URL, bearer=operator.mint_token(subject="service:scim-okta", scope=ROOT,
                                              ttl_seconds=30 * 86400)["token"])
faye = idp.scim.create_user(scim_user("faye@demo.example", external_id="00u-faye", display_name="Faye Ng"))
group = idp.scim.create_group(scim_group("Platform", [faye["id"]]))
print(idp.scim.list_users(filter='userName eq "faye@demo.example"')["totalResults"])  # 1
print(idp.scim.resource_type("User")["endpoint"])                # /Users
idp.scim.patch_user(faye["id"], [scim_op("replace", value={"active": False})])  # her tokens stop
idp.scim.patch_user(faye["id"], [scim_op("replace", value={"active": True})])   # and work again
idp.scim.delete_group(group["id"])
idp.scim.delete_user(faye["id"])

SCIM's error bodies raise the SDK's typed errors, with the CortexDB code as error_code and SCIM's scimType as scim_type.

Restores

A copy restored with its access state lost (cortexdb restore-check --access-state lost) holds every root it restored pending: requests that do not repair it answer 503 ACCESS_RESTORE_PENDING (not retriable), until each of the root's access sources re-sends its state (change sets staged with resync=True, the last one resync=True and activated) and acknowledges with access_restore_ack, or the trusted operator overrides with admin_access_restore. access_records pages a root's registry records, what a source diffs its own state against, with the root's consistency token (at_least= waits for one). access_texts_erase(root, path) erases the free texts a record names (reasons, a hold's matter, a policy case's notes) from every voter's files; the record keeps its references.

from cortexdb.v1 import V1APIError

records = operator.access_records(ROOT)
print(len(records["items"]) > 0, records["revision_changed"])     # True False
print(operator.access_records(ROOT, at_least=records["consistency_token"])["items"] != [])  # True
print(compiler.access_texts_erase(ROOT, ENG)["erased"] >= 0)     # True
try:
    operator.access_restore_ack(ROOT)
except V1APIError as error:
    print(error.error_code)                                       # RESTORE_NOT_PENDING

Source containers

A connector writes each item into its permission unit's container (<tenant>/source:<system>/container:<key>), registered through change sets. When an item's sharing changes it moves the item's live events with access_rehome (nothing is erased); after each sync it stores the source's watermark (access_watermark_put), which reads report in coverage.

from datetime import datetime, timedelta, timezone

slack_owner = member("service:slack", "owner")
source = f"{ROOT}/source:slack"
general, launch = f"{source}/container:general", f"{source}/container:launch"
compiler.changeset_apply(ROOT, "slack-1", [
    scope_record(source, members=[slack_owner], kind="source", parent=ROOT),
    scope_record(f"{ROOT}/set:slack-members", members=[slack_owner], kind="set", parent=ROOT,
                 set={"entries": [{"actor": "user:alice"}]}),
    scope_record(general, members=[slack_owner, set_member(f"{ROOT}/set:slack-members", "reader")],
                 kind="container", parent=source),
    scope_record(launch, members=[slack_owner, set_member(f"{ROOT}/set:slack-members", "reader")],
                 kind="container", parent=source, private=True),
])
slack = Cortex(URL, bearer=operator.mint_token(subject="service:slack", scope=ROOT)["token"])
synced = (datetime.now(timezone.utc) - timedelta(seconds=5)).strftime("%Y-%m-%dT%H:%M:%SZ")
print(slack.access_watermark_put(source, synced)["changed"])      # True
message = slack.experience(general, text="Launch moves to Friday", wait="indexed")
print(slack.access_rehome(launch, [message["event_id"]])["moved"])  # 1
print([w["source"] for w in alice.access_watermarks(launch)["watermarks"]])

Rollups and publish

A unit with an aud:roll compartment gets rollups: typed totals per month and synthesis items, built by the shard leader and served to a reader only when it reads every input (dominance), so its head reads them and others do not. A person widens a memory only by publishing a copy of it into a home they write; erasing the original erases the copy.

import time

roll = f"{ENG}/aud:roll"
compiler.changeset_apply(ROOT, "eng-rollup-1", [
    scope_record(roll, members=[owner], kind="compartment", parent=ENG, audience="rollup"),
    scope_record(f"{ENG}/attrs:unit", kind="attrs", parent=ENG, attrs={"heads": [{"actor": "user:ada"}]}),
])
alice.experience(ENG, text="Decided: ship the new pricing page", wait="indexed")
for _ in range(60):                       # a unit is built the first time it is read
    rolled = ada.rollups(roll)
    if rolled["items"]:
        break
    time.sleep(1)
print(rolled["items"][0]["kind"], ada.rollup_totals(roll)["totals"]["events"] > 0)

note = alice.experience(f"{ROOT}/user:alice", text="Proposal: weekly design reviews", wait="indexed")
copy = alice.publish(ENG, f"{ROOT}/user:alice", event_id=note["event_id"], wait="indexed")
print(copy["published_from"]["event_id"] == note["event_id"])     # True

Only a person's own chain publishes, through an application the person drives too (app with client: "person"): an agent's act, a run, a service, an app without client and the operator key are 403 PUBLISH_PERSON_REQUIRED. consistency=True returns a token at the target's root. Publishing a norm (intent="procedure", rule, …) needs a sign-in within 10 minutes (401 FRESH_AUTH_REQUIRED, its RFC 9470 challenge in the error's headers).

Reading across scopes

A read names its home scope and up to 40 extra_scopes under the same tenant root, and searches them all as one read set; coverage=True reports what it searched, what of its read set the caller may read, and each gap with its reason (a scope left out by a bound, a deadline, a source's watermark, a scope the caller may not read …), so a response never reads as complete when it is not. beliefs="merged" rebuilds the beliefs of every scope read into one set, and mounts="default" adds an application chain's mounts (a token minted with app). purpose declares what the read is for ([a-z0-9_-]{1,64}), which an access path's when: {purpose} condition reads; a read refused only for want of the purpose its one path names raises errors.PolicyDeniedError with reason == "purpose_required" and purpose the one it needs. recall, recall_stream, answer and compose take the same fields, and access_check and access_explain take purpose.

pack = alice.recall(ENG, query="when does the launch move?", extra_scopes=[general, launch],
                    coverage=True)
print(sorted(pack["coverage"]["searched"]["scopes"]))
# ['org:demo/source:slack/container:general', 'org:demo/source:slack/container:launch',
#  'org:demo/team:eng']
print({event["scope"] for event in pack["layers"]["events"]} >= {launch})   # True
answer = alice.answer(ENG, "When does the launch move?", extra_scopes=[general, launch],
                      beliefs="merged", coverage=True, purpose="planning")
print([citation["id"] for citation in answer["citations"]][:1])

A scope outside the root, or a list past the bound, is 422 INVALID_EXTRA_SCOPES; mounts without an application in the token 422 INVALID_MOUNTS (details.reason no_app).

Streams

recall_stream and answer_stream yield StreamEvents (event, data, id) as the server sends them: a recall's plan, each layer, its context_block and done (with the consistency token); an answer's tokens, citations and done. lifecycle_stream follows the write lifecycle (captured, indexed …) of a scope, an event or a job. An error event raises its typed error. On AsyncCortex they are async iterators.

names = [event.event for event in alice.recall_stream(ENG, query="what shipped?")]
print(names[0], names[-1])                                        # plan done
text = "".join(event.data["text"] for event in alice.answer_stream(ENG, "What shipped?")
               if event.event == "token")
print(bool(text))                                                 # True

Writes that cite their sources

A write that summarizes or derives from other memories names them in provenance (at most 1,024 event ids in the same root): the stored event keeps them, and reads and erasures follow them. intent, subject, observed_actor, directives and context are the envelope's other fields.

notes = alice.experience(ENG, text="Raw notes: the gateway timed out twice", wait="indexed")
summary = alice.experience(ENG, text="Summary: gateway timeouts after the deploy", wait="indexed",
                           provenance=[notes["event_id"]], intent="decision")
stored = alice.event(summary["event_id"], scope=ENG)
print(stored["provenance"] == [notes["event_id"]], stored["context"]["intent"])  # True decision

An id that names no event of the root is 422 PROVENANCE_UNKNOWN (details.cited), an erased one 409 PROVENANCE_ERASED, one in another root 422 CROSS_ROOT_PROVENANCE.

The read-free forget

A deletion worker forgets by keys it already holds, so the server reads nothing first: idempotency_keys (the keys the writes were sent with; for another principal's write, (key, caller)), and capture_keys (a label the writer put on every event of one capture). forget_preview counts the same selector. It takes Owner at the scope; the default derived_only cascade removes the selected events with what was derived from them. What a legal hold keeps is matched, not deleted: matched counts it (at most requested), and events_under_legal_hold and derived_under_legal_hold count it apart, in neither deleted nor estimated_deleted.

home = f"{ROOT}/user:alice"
alice.experience(home, text="A draft I will take back", idempotency_key="draft-1", wait="indexed")
preview = alice.forget_preview(home, idempotency_keys=["draft-1"])
print(preview["requested"], preview["matched"], preview["events_under_legal_hold"])  # 1 1 0
gone = alice.forget(home, idempotency_keys=["draft-1"], from_preview_id=preview["preview_id"])
print(gone["deleted"]["events"])                                  # 1

for part in ("first", "second"):
    alice.experience(home, text=f"call 42, {part} passage", labels=["cap:call-42"], wait="indexed")
print(alice.forget(home, capture_keys=["cap:call-42"])["deleted"]["events"])   # 2

Erasures

erase deletes the rows themselves (not only what was derived), frees the writes' idempotency keys and leaves a durable receipt that erasure_verify proves store by store; erasure_wait polls a job (with residual=True, until its bytes left the log on every voter). The trusted operator also erases an outside party by address (subject_email: the events writers labelled party:<address>, in every scope of the tenant), a person (erase_person: their region, their application and capture homes, their own messages in the tenant's groups, then their directory entry) and a whole tenant (erase_tenant, a dry run unless dry_run=False and confirm=True). A legal hold keeps what it names.

slack.experience(general, text="Pat from Initech asked for a quote", wait="indexed",
                 labels=["party:pat@initech.example"])
preview = operator.erasure_preview(ROOT, subject_email="pat@initech.example")
print(preview["estimated_affected"]["events"])                    # 1
job = operator.erase(ROOT, subject_email="pat@initech.example", confirm=True,
                     from_preview_id=preview["preview_id"])
receipt = operator.erasure_receipt(job["erasure_id"])
print(receipt["receipt"]["kind"], receipt["sha256_verified"])     # party True
print(operator.erasure_verify(job["erasure_id"])["checks_ok"])    # True

leaver = operator.erase_person(ROOT, "user:bob", confirm=True, reason="leaver")
print(operator.erasure_wait(leaver["erasure_id"])["status"])      # completed
plan = operator.erase_tenant(ROOT)                                # counts only
print(plan["dry_run"], plan["counts"]["records"] > 0)             # True True

Operator routes

The trusted operator reads and sets each tenant's operator mode (operator_data_reads: allow, audit, which tells the people whose memories an operator read touched, or deny, which takes break-glass) beside the deployment's own (deployment_mode, what a tenant that clears its mode falls back to), reads the access report's counters, rebinds a root's records from the static key's tenant, and reads the node's health, readiness and metrics.

settings = operator.admin_tenant_settings(ROOT)
print(settings["operator_data_reads"], settings["deployment_mode"])
# {'mode': 'allow', 'source': 'deployment'} allow
operator.admin_set_tenant_settings(ROOT, operator_data_reads="audit")
report = operator.admin_access_report(ROOT)
print(report["scope"], len(report["checks"]) > 0)                 # org:demo True
operator.admin_set_tenant_settings(ROOT, operator_data_reads=None)  # the deployment's mode again
print(operator.ready()["ready"], "disk" in operator.admin_metrics())  # True True

Tokens and principal chains

A client acts as its token's subject. mint_token mints a PASETO token for a principal; its chain fields say who is acting for whom: act (an agent acting for the subject; it writes only with a run), app and client (an application, and whether a person or an agent drives it), run and run_start, and taint tags. mint_agent_token and mint_app_token mint the two common chains. A token with a chain field lives at most a day (CHAIN_TOKEN_MAX_TTL_SECONDS), and an agent chain token names its caps with scopes that confine each; the SDK refuses before sending what the server always refuses (check_mint, a ValueError). Other tokens live at most lifetime_ceiling(subject): a week for a person or an agent, 30 days for a service. with_token derives a client that uses another token over the same connections.

from cortexdb.v1 import lifetime_ceiling

erin = Cortex(URL, bearer=operator.mint_token(subject="user:erin", scope=ROOT)["token"])
print(erin.actor, lifetime_ceiling("user:erin"))                 # user:erin 604800

home = f"{ROOT}/user:alice"
planner = alice.with_token(alice.mint_agent_token(
    "user:alice", "agent:planner", run="weekly-plan-1", ttl_seconds=600,
    caps=["scope.write", "scope.read.local"], scopes=[f"scope.write:{home}!", f"scope.read.local:{home}!"]))
planner.experience(home, text="Drafted the weekly plan", wait="indexed")
chain = planner.whoami()["chain"]
print(chain["act"], chain["run"])                                # agent:planner weekly-plan-1

# The application's own principal mints a person's token for it.
notes = alice.with_token(operator.with_options(actor="service:notes").mint_app_token(
    "user:alice", "service:notes", scope=ROOT))
print(notes.whoami()["chain"]["app"])                            # service:notes

act naming an AI employee is 422 INVALID_BODY (act_not_delegable: an AI employee's token has the employee as sub and no act); a token over its ceiling 422 INVALID_BODY ttl_too_long at the mint and 401 INVALID_TOKEN_CLAIM (lifetime_over_ceiling) once presented. Each caps entry is one capability: a pattern such as scope.* is 422 INVALID_BODY with field == "caps", entry the entry and reason == "wildcard_not_allowed" ("invalid" for an entry that does not parse).

Break-glass and the subject-token shadow

A tenant's operator mode (admin_set_tenant_settings) decides what the operator key may do with the tenant's data: allow, audit (served and counted), or deny, where reads of stored rows, private regions and the in-band paths answer 403 BREAK_GLASS_REQUIRED unless the request carries X-Cortex-Break-Glass. break_glass(reason, ticket, note) derives a client that sends it: every request is audited with the reason and ticket, and a read of a private region tells its person first. The header is checked here as the server checks it. with_options(subject_token=...) sends the person's own token beside the key (X-Cortex-Subject-Token): the server evaluates each decision for that token too and counts the differences in the access report, without changing any response.

from cortexdb import errors

operator.admin_set_tenant_settings(ROOT, operator_data_reads="deny")
try:
    operator.events(f"{ROOT}/user:alice")
except errors.BreakGlassRequiredError as refused:
    print(refused.error_code, refused.mode)                      # BREAK_GLASS_REQUIRED deny
glass = operator.break_glass("support", "SUP-1142", "alice asked for help with her notes")
print(len(glass.events(f"{ROOT}/user:alice")["items"]) > 0)       # True: served and audited
print(alice.access_notices(ROOT)["items"][-1]["cause"])          # break_glass
glass.admin_set_tenant_settings(ROOT, operator_data_reads=None)  # lowering from deny takes it too

shadowed = operator.with_options(subject_token=alice.bearer)
shadowed.recall(ENG, query="what shipped?")
checks = {c["check"]: c["requests"] for c in operator.admin_access_report(ROOT)["checks"]}
print(checks["subject_shadow"] > 0)                              # True

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

with_options(actor=, headers=, timeout=, consistency=, bearer=, break_glass=, subject_token=) derives a client that shares the connection pool; break_glass(reason, ticket, note) and with_token(token) are shorthands.

Links

• Documentation
• Support

License

Apache-2.0