Contexara Python SDK

Persistent memory for your AI apps, agents, and tools, built directly on CTX's existing capture and retrieval pipeline. This SDK does not create a second storage/processing system; every call goes through the same production backend CTX's own hooks (Claude Code, Cursor, Codex) and Loop already use.

Install

pip install contexara

Requires Python 3.10 or later. The full test suite passes on both 3.10 (the declared floor) and 3.13 (a current release), each verified in a real, separate virtual environment against the built wheel — not just an import check.

api_key can also be set via the CONTEXARA_API_KEY environment variable.

The API key you pass determines everything server-side: which account's data a capture is written to, which account's data a search can read, and which account is billed/usage-metered for the call. There is no separate tenant or workspace selector independent of the key — an api_key scoped to capture-only or recall-only will be rejected (403) for the other operation; see your dashboard for key scopes and usage.

Quick start

from contexara import Contexara

ctx = Contexara(api_key="ctx_live_...")

# Save something
receipt = ctx.memory.add("The deployment region is Mumbai.")
print(receipt.status)  # "accepted": durably queued, not yet processed or searchable (see Eventual consistency below)

# Recall it later
result = ctx.memory.search("What deployment region did we choose?")
for m in result.results:
    print(m.text, m.rrf_score)

# Budget-aware string ready for prompt injection
block = ctx.context_block("deployment region", token_budget=500)

ctx.close()

Or use it as a context manager:

with Contexara(api_key="ctx_live_...") as ctx:
    ctx.memory.add("The deployment region is Mumbai.")

Async

from contexara import AsyncContexara

async def main():
    async with AsyncContexara(api_key="ctx_live_...") as ctx:
        receipt = await ctx.memory.add("The deployment region is Mumbai.")
        result = await ctx.memory.search("deployment region")

What this SDK wraps

Four real, public, x-api-key-authenticated routes, nothing else. See platform/sdk/contracts/PUBLIC_API_CONTRACT.md in this repo for the full, source-derived contract this SDK is built against.

capture_base_url defaults to https://api.contexara.com. retrieval_base_url (alias: base_url, for backward compatibility) defaults to https://retrieval.contexara.com.

memory.add(): capturing text

receipt = ctx.memory.add(
    "The deployment region is Mumbai.",
    assistant_message=None,   # optional: an omitted reply is sent as null, never a fabricated string
    session_id=None,          # optional, see Sessions below
    project_id=None,
    workspace_id=None,
)

print(receipt.event_id, receipt.session_id, receipt.turn_number, receipt.turn_id, receipt.status)

event_id, session_id, turn_number, and turn_id are generated locally by the SDK, before the request is even sent, so they're known immediately, not returned by the server. Only status ("accepted" or "processed", the server's own real vocabulary, see Eventual consistency below) comes back from the API.

Sessions

session_id is your stable identity for one logical conversation, the same concept a Claude Code or Cursor hook already uses (its own IDE's conversation_id, or a fallback constant when the IDE has none). It is not generated fresh per call. If you don't pass one, the SDK generates a single default session_id for the life of the Contexara instance and reuses it across every add() call from that instance, incrementing turn_number each time, mirroring how a real hook tracks one conversation.

ctx = Contexara(api_key="...")
ctx.memory.add("First message.")   # turn_number=1, same auto-generated session_id
ctx.memory.add("Second message.")  # turn_number=2, same session_id

For a real multi-turn integration (e.g. one CrewAI crew run, one LangGraph thread), pass your own stable session_id explicitly:

ctx.memory.add("Researching Rust vs Go.", session_id="crew-run-482")
ctx.memory.add("Comparing build times.", session_id="crew-run-482")  # turn_number=2 for THIS session

Never reuse one session_id across genuinely different intended destinations (different project_id/workspace_id). A session's stored destination is set once and does not silently move; start a new session_id instead of trying to redirect an existing one.

Session capacity and resuming across processes. The SDK tracks turn numbers locally in a bounded, in-memory registry (100,000 distinct session_ids per Contexara instance by default, configurable). A session_id consumes one registry slot the first time it's used, in either allocation mode (automatic or explicit turn_number) — the slot is never freed for the life of the instance, and registering a brand-new session_id past capacity raises ValidationError. An already-registered session_id keeps working past capacity regardless of mode.

This registry is per-process, in-memory, and not shared or persisted. If your application resumes an existing conversation in a new client instance or a new process (a restart, a new worker picking up a queued job, a distributed deployment), the new instance has no memory of that session's prior turns — you must pass turn_number explicitly, tracked in your own durable storage. Letting the SDK auto-allocate for a resumed session_id it has never seen will restart that session's local counter at 1, which can collide with turns the server already has for it.

Durable replay (persistent / distributed applications)

For an application that needs to persist a capture event before sending it (so it can survive a restart and retry the exact same logical write), use prepare_add() + send() instead of add():

event = ctx.memory.prepare_add("The deployment region is Mumbai.", session_id="crew-run-482")
# event.to_dict() is JSON-serializable, persist it to your own durable queue here

receipt = ctx.memory.send(event)
# On an uncertain outcome (timeout, crash before this line runs), reload the
# SAME event from your queue and call ctx.memory.send(event) again. Never
# call prepare_add() again for what is logically the same write, since that
# generates a NEW event_id.

Known, unresolved backend limitation, documented honestly rather than silently: the backend's duplicate-write protection is keyed on event_id, checked before content comparison. Resending the exact same event_id unchanged (a real retry) is safe. Reusing an event_id with genuinely different content (a client bug, never something this SDK does on its own) is not reliably caught by the server today — this is a real, currently-open backend gap, not a fixed design constraint, and may be closed in a future server-side change. This SDK enforces the replay rule by construction (a prepared event's fields are frozen; nothing on the SDK's own retry path ever regenerates them), not by relying on the server to catch a violation. This does not affect the supported exact-replay path above: replaying the same, unchanged event is safe today and will remain safe regardless of how this gap is eventually closed.

memory.search(): recall

result = ctx.memory.search(
    "What deployment region did we choose?",
    limit=10,                  # 1-50
    scope="general",           # "general" | "project"
    project_id=None,           # required when scope="project"
    workspace_id=None,
    deep=False,                # opt into Deep Recall synthesis, if entitled
    force_fast=False,          # short-circuits Deep Recall even if deep=True
)

for m in result.results:
    print(m.text, m.rrf_score, m.source_type)

if result.deep_recall_ran:
    print(result.synthesis)  # may still be None even when Deep Recall ran, see below

scope="all_active" is not part of the stable SDK contract. The backend does not yet enforce real workspace/project scoping for it (confirmed unscoped as of this SDK's build). Only "general" and "project" are supported.

deep=True and force_fast=True together are rejected client-side as contradictory; send only one.

Deep Recall

Requesting deep=True does not guarantee synthesis ran. The server silently falls back to plain Fast Recall (no synthesis/ grounded_in_source_windows in the response at all) when the account's plan or quota doesn't allow Deep Recall. Even when it does run, result.synthesis can legitimately be None, a normal outcome, not an error. Check result.deep_recall_ran to know whether Deep Recall actually executed; check result.synthesis is not None separately to know whether it produced an answer.

Deep Recall's limit only controls the initial Fast Recall pass. The final synthesized result set is independently budget-limited server-side and is not guaranteed to respect your requested limit.

upload(): files

for event in ctx.upload("notes.md", caption="Q3 roadmap notes"):
    if event.chunk:
        print(event.chunk, end="")
    if event.attachment_id:
        attachment_id = event.attachment_id

explain defaults to False (ingestion-oriented use). A non-image file with explain=False produces just the metadata frame and completion, no generated explanation. Images still produce an explanation regardless of this flag today, a real, current backend behavior, not an SDK limitation.

Supported options: caption, explain, session_id, project_id, workspace_id, group_id (only meaningful for a multi-image batch).

Uploads are never automatically retried. A retry after any failure, even after the file was already successfully stored, would create a second, independent attachment record for the same content, since duplicate detection only happens at the blob-content level (event.duplicate), never at the request level.

A stream that raises StreamError may still have stored the file successfully; the exception carries whatever content_hash/duplicate/ attachment_id metadata was already received, so your error handling can tell the difference between "nothing happened" and "the file is stored but the explanation stream failed."

Eventual consistency

A successful add() does not mean the content is immediately searchable. There are two separate, sequential async hops after add() returns:

1. status="accepted" (202, the real production response) means the write is durably queued, not yet processed by the capture worker.
2. Once processed, a second, separate step (per-turn evidence indexing) makes the content searchable. This can overlap with the worker's own remaining work, not strictly happen after it finishes.

If you search() immediately after add(), you may not find what you just added. There is no polling helper built into this SDK v1; if your application needs "wait until searchable," implement a bounded retry loop around search() yourself.

status="processed" (200) is a real, valid response, but only in a self-hosted/dev deployment with no message queue configured. Do not assume every deployment returns 202.

Errors

from contexara import (
    ContexaraError,           # base class, catches everything below
    AuthenticationError,      # 401: invalid/inactive key
    PermissionDeniedError,    # 403: scope denied, capture paused, unauthorized workspace
    QuotaExceededError,       # 402: hard quota lock
    RateLimitError,           # 429: check .retry_after_seconds (may be None)
    ValidationError,          # 400/422, or a request this SDK refused to send at all
    ConflictError,            # 409: synchronous-mode-only turn collision
    ConnectionError_,         # network-level failure
    ContexaraTimeoutError,    # timed out after all configured retries
    StreamError,              # /upload SSE protocol failure, carries partial attachment metadata
    ContexaraAPIError,        # any other non-2xx
)

try:
    ctx.memory.add("...")
except ContexaraError as e:
    print(e)

ctx.verify() returns False for an invalid/inactive key (401). It never returns False for a network failure or timeout, which raise normally. A True result does not confirm capture/recall permission or remaining quota for any specific operation; it only confirms the key itself is valid.

Migration note: all exception types live in contexara.errors (or the top-level contexara package, as shown above). An earlier, pre-release draft of this SDK had a separate contexara.exceptions module with a narrower set of types (ContexaraError, AuthenticationError, ContexaraTimeoutError, ContexaraAPIError); that module was removed and replaced by contexara.errors, which is a superset (same four types, plus PermissionDeniedError, QuotaExceededError, RateLimitError, ValidationError, ConflictError, ConnectionError_, ResponseValidationError, StreamError). This SDK has never been published, so there is no real installed base importing the old path. This note exists for anyone who may have copied source from an early draft directly rather than installing a release.

Resilience

from contexara import Contexara, ResiliencePolicy

ctx = Contexara(api_key="...", resilience=ResiliencePolicy(timeout_ms=5000, max_retries=1))

Retry policy is operation-specific, not one flat setting:

• verify() / search() (Fast Recall): bounded automatic retries on transient failure.
• search(deep=True): conservative, at most one retry, since Deep Recall may incur real generation cost server-side before an uncertain failure.
• memory.add() / send(): retries resend the exact unchanged prepared event only.
• upload(): never automatically retried, by design (see above).

A 429 with a real Retry-After header is honored automatically. A 429 with no Retry-After (e.g. upload's daily/monthly quota limit) is not auto-retried.

Testing without a real namespace

from contexara import MockContexaraClient, Memory

ctx = MockContexaraClient()
ctx.seed([Memory(memory_id="m1", text="user prefers dark mode", rrf_score=0.9)])

result = ctx.memory.search("preferences")  # no network call, returns the seeded data

receipt = ctx.memory.add("a fact")  # recorded in ctx.recorded_captures, no network call

The mock does not simulate real semantic search or extraction. seed() sets exactly what search() returns regardless of query text, and add() never produces a synthetic memory a later search() would need to have "learned." Seed what you expect a test to see; don't expect the fake to derive it.

Injected transport (advanced / test use)

import httpx
from contexara import Contexara

transport = httpx.MockTransport(my_handler)
ctx = Contexara(api_key="...", transport=transport)

When you supply a transport, this client does not own or close it. Calling ctx.close() leaves your transport open, since you're responsible for its lifecycle.

What this SDK does not do

There is no list(), delete(), chat(), or stats() method. No public backend route exists for any of them (chat additionally requires an internal secret, not a public API key, so it isn't reachable by this SDK at all regardless). scope="all_active" is not exposed as a stable value. This SDK does not invent placeholder methods for capability CTX's backend does not have; see platform/sdk/contracts/PUBLIC_API_CONTRACT.md for the complete, current picture.