Python SDK

1. Install and authenticate

Use Python 3.10 or newer. Install the published contexara package in your application’s virtual environment. The examples target version 0.1.0 and use the synchronous client unless marked async.

python -m venv .venv
# macOS / Linux
source .venv/bin/activate
python -m pip install "contexara==0.1.0"

On Windows PowerShell, activate with .venv\Scripts\Activate.ps1. Set the environment variable in the same terminal that runs your app.

export CONTEXARA_API_KEY="ctx_live_replace_with_your_key"

$env:CONTEXARA_API_KEY = "ctx_live_replace_with_your_key"

Create an API key in the dashboard at /keys. Enable capture permission for writes and recall permission for searches. Keep the key on your server or in your secret manager; never expose it in browser JavaScript, a public repository, or logs. The key determines the authenticated account. A workspace ID selects an authorized destination; it does not grant access to another account.

2. Capture your first turn

Save this as capture.py and run python capture.py. Contexara reads CONTEXARA_API_KEY automatically. A context manager closes the client-owned HTTP transport. This script performs a real write to your account.

from contexara import Contexara

with Contexara() as ctx:
    if not ctx.verify():
        raise RuntimeError("The API key is invalid or inactive")
    receipt = ctx.memory.add(
        "The deployment region is Mumbai.",
        assistant_message="I will use Mumbai for deployment.",
    )
    print(receipt.event_id, receipt.turn_id, receipt.status)

verify() only checks whether the key is active. It does not establish operation permissions or remaining quota. The receipt contains locally generated event_id, session_id, turn_number, and turn_id, plus the server’s accepted or processed status.

3. Recall existing memories

Run this separately after processing, or against memories already in your account. An empty results list is a valid response, not an exception.

from contexara import Contexara

with Contexara() as ctx:
    result = ctx.memory.search(
        "What deployment region did we choose?",
        limit=5,
        scope="general",
    )
    for memory in result.results:
        print(memory.memory_id, memory.text, memory.source_type)
    if not result.results:
        print("No matching memories are available yet.")

Memory records expose text, memory_id, rrf_score, entities, source_type, source_id, modality, content_hash, page_number, slide_number, mime_type, event_time, record_time, and speaker. Provenance fields can be null. rrf_score is a ranking signal, not a probability or confidence percentage. Keep the returned order. timings_ms is an extensible numeric map, not a fixed list of timing keys. Entity objects may have name, text, or type; none is always present.

Capture parameters and sessions

A client creates one default session ID and increments turn numbers locally. For an application handling multiple conversations, pass one stable session ID per conversation. Counters are in memory, not shared between workers or persisted across restarts. For resumed or distributed sessions, allocate turn numbers atomically in your own durable storage and pass them explicitly. Never mix automatic and explicit numbering for the same session in one client. Start a new session when the intended project or workspace changes. Each distinct session consumes one of the client registry’s 100,000 slots; existing sessions continue working at capacity, but new sessions are rejected.

from uuid import uuid4
from contexara import Contexara

session_id = str(uuid4())  # New conversation only; persist for later turns.
with Contexara() as ctx:
    ctx.memory.add("Compare Rust and Go.", session_id=session_id)
    ctx.memory.add("Choose Rust for the parser.", session_id=session_id)

Project and workspace scope

Use general for unassigned memories in the selected workspace, or project with an active project ID. Omit the workspace ID for your personal workspace; pass an authorized workspace ID for a team workspace. A project must belong to that workspace. Do not pass a project ID with general scope. all_active is outside the supported stable contract. Capture and recall destinations are separate: explicitly use the same intended project and workspace in both calls.

Replace the example UUID with an active project from your workspace. This fragment assumes an open ctx client.

result = ctx.memory.search(
    "What deployment region did we choose?",
    scope="project",
    project_id="11111111-1111-4111-8111-111111111111",
    limit=10,
)
# For a team workspace, also pass workspace_id on capture and search.

Build context for your model

Use format_context_block on an existing result to avoid a second network request. It preserves memory IDs and source labels while fitting the output to a token budget.

from contexara import format_context_block

# result is the RetrieveResult returned by memory.search().
context = format_context_block(result, token_budget=500)
print(context)
# Pass context as reference material to your chosen model.

ctx.context_block(query, token_budget=500) is a convenience method that performs its own general search. It does not accept project/workspace options. For scoped context, search with the desired scope first, then format the returned result. Token counts use an approximate character-count heuristic by default. Supply token_counter=lambda text: len(tokenizer.encode(text)) with your model’s tokenizer for model-specific counting. Empty results produce an empty string. Treat retrieved text as reference data, not trusted instructions.

Deep Recall

Deep Recall is opt-in and can fall back to Fast Recall when plan or quota conditions do not allow synthesis. A completed Deep Recall run may still have a null synthesis. Check whether Deep Recall ran separately from whether an answer was produced. Its final result count is independently budgeted: limit constrains the initial Fast Recall pass, not the final synthesized result set. Do not enable deep and force-fast together.

result = ctx.memory.search("Why did we choose Mumbai?", deep=True)
if result.deep_recall_ran and result.synthesis is not None:
    print(result.synthesis)
else:
    for memory in result.results:
        print(memory.text)

Async applications

Use AsyncContexara in async request handlers and workers. Await verify, add, search, send, and context_block; consume upload with async for. Reuse the client within its application lifecycle, and close it with async with.

import asyncio
from contexara import AsyncContexara

async def main():
    async with AsyncContexara() as ctx:
        result = await ctx.memory.search("deployment region", limit=5)
        for memory in result.results:
            print(memory.text)

asyncio.run(main())

Durable capture and replay

prepare_add() performs local preparation without a network call. The following example saves one event before delivery using exclusive file creation. Use a transactional outbox or durable queue with atomic turn allocation in a distributed service. The rehydration type is currently in contexara._capture, not the top-level exports; pin 0.1.0 when using that module path.

import json
from pathlib import Path
from contexara import Contexara
from contexara._capture import PreparedCaptureEvent

path = Path("pending-capture.json")
with Contexara() as ctx:
    if not path.exists():
        event = ctx.memory.prepare_add("The deployment region is Mumbai.")
        with path.open("x", encoding="utf-8") as file:
            json.dump(event.to_dict(), file)
    event = PreparedCaptureEvent.from_dict(
        json.loads(path.read_text(encoding="utf-8"))
    )
    receipt = ctx.memory.send(event)
    print(receipt.event_id, receipt.status)
# A repeated run replays this same event. Mark it delivered in your outbox
# after acknowledgement; use a new record for each new logical write.

Upload documents and images

Pass a path, pathlib.Path, or binary file object. Binary inputs need a filename and MIME type when these cannot be inferred. Upload yields structured events as the server streams them. Create notes.md locally before running this example.

from contexara import Contexara, StreamError

with Contexara() as ctx:
    try:
        for event in ctx.upload("notes.md", caption="Roadmap notes", explain=False):
            if event.attachment_id:
                print("Stored attachment:", event.attachment_id)
            if event.chunk:
                print(event.chunk, end="", flush=True)
    except StreamError as error:
        print("Stream failed; attachment:", error.attachment_id)
        raise

Options include caption, filename, mime_type, explain, session_id, project_id, workspace_id, and group_id. group_id is for grouping image batches. workspace_id is translated to explicit_workspace_id in multipart form data.

Timeouts and retries

Default policy: 5,000 ms timeout, one retry, 200 ms base backoff, 4,000 ms maximum backoff, and no overall deadline. Backoff uses jitter. Verify and Fast Recall retry transient failures within the configured budget; Deep Recall retries at most once because generation may already have incurred cost. Capture retries reuse the unchanged prepared event. Uploads never retry automatically. A usable Retry-After header is honored for 429 responses; a 429 without it is not automatically retried. An overall deadline includes attempts and backoff for ordinary requests; upload has separate connection/read-idle behavior rather than one total stream deadline.

from contexara import Contexara, ResiliencePolicy

policy = ResiliencePolicy(timeout_ms=5000, max_retries=1, deadline_ms=8000)
with Contexara(resilience=policy) as ctx:
    result = ctx.memory.search("deployment region")

The constructor’s timeout argument is in seconds; ResiliencePolicy.timeout_ms and deadline_ms are milliseconds. Providing a resilience policy overrides the constructor’s timeout/max_retries shorthand. Do not wrap capture in a retry that calls add() again: use prepared-event replay for uncertain outcomes.

Handle failures explicitly

All SDK operation errors below share ContexaraError; invalid constructor configuration can raise ValueError. Network failures use ConnectionError_ (with a trailing underscore). RateLimitError.retry_after_seconds can be None.

from contexara import Contexara, ContexaraError, RateLimitError

with Contexara() as ctx:
    try:
        result = ctx.memory.search("deployment region")
    except RateLimitError as error:
        print("Retry-after seconds:", error.retry_after_seconds)
        raise
    except ContexaraError as error:
        print(type(error).__name__, str(error))
        raise

Test without API calls

Use the included mock for deterministic application tests. seed() controls search results for every query; add() records captures but does not extract or learn new memories.

from contexara import MockContexaraClient, Memory

ctx = MockContexaraClient()
ctx.seed([Memory(memory_id="m1", text="Use Mumbai.", rrf_score=0.9)])
result = ctx.memory.search("deployment region")
assert result.results[0].text == "Use Mumbai."
ctx.memory.add("Use Mumbai.")
assert len(ctx.recorded_captures) == 1

For HTTP-level tests, inject httpx.MockTransport through transport=. The caller owns an injected transport and is responsible for closing it. Test your empty results, timeout, permission-denied, and partial-upload paths as well as successful responses.

Hosts and supported surface

base_url and CONTEXARA_BASE_URL are legacy retrieval-host aliases. Different explicit base_url and retrieval_base_url values are rejected. close() closes client-owned resources; with handles this automatically.

The 0.1.0 SDK exposes verify, capture, search, context formatting, and uploads. It does not expose list, delete, chat, or stats methods. Public API keys cannot call Loop’s internal chat endpoints. Capture acknowledgements cannot report processing errors that occur later in the asynchronous worker.