CueCruxVaultCrux

SDKs · 2. Python SDK

corecrux-client is a hand-written Python client for the Crux Daemon's HTTP API, with a synchronous and an asynchronous class exposing the identical method set. It has one runtime dependency (httpx), requires Python 3.10, and is the only Python option in the portfolio. This chapter is reference. For the choice between clients, see chapter 0.

Everything below was verified on 2026-07-27 against the wheel published on PyPI, not against the repository, the two agree.

2.1 Install and version

pip install corecrux-client

This resolves to 0.1.0, uploaded 2026-06-12, which is the same version the repository carries (pyproject.toml:7). Unlike the TypeScript SDK, the published Python package is current.

FactValueSource
Distribution namecorecrux-clientpyproject.toml:6
Import namecorecrux_clientpyproject.toml:20
LicenceApache-2.0: open sourcepyproject.toml:10
Python requirement>=3.10pyproject.toml:11
Dependencieshttpx>=0.27, and nothing elsepyproject.toml:12
Build backendhatchling==1.31.0, pinned exactlypyproject.toml:2
Wheel contentsFour modules: __init__, client, errors, typesverified from the published wheel

To install from a checkout instead:

cd sdks/python
pip install -e .

2.2 The public API surface

Eleven names are exported (__init__.py:20):

from corecrux_client import (
    AsyncCoreCruxClient,
    CoreCruxClient,
    CoreCruxError,
    Fact,
    FactQueryResult,
    SessionState,
    StoreFact,
    TextSearchCoverage,
    TextSearchHit,
    TextSearchMeta,
    TextSearchResult,
)

TextSearchGap has no Python equivalent, gaps arrive as raw dictionaries inside TextSearchCoverage.gaps. There is no exported ProblemDetails, no event types and no options dataclasses; the query methods take keyword arguments rather than an options object.

2.3 Constructing a client

CoreCruxClient(
    base_url: str = "http://localhost:14800",
    token: str | None = None,
    *,
    timeout: float = 30.0,
)

AsyncCoreCruxClient(
    base_url: str = "http://localhost:14800",
    token: str | None = None,
    *,
    timeout: float = 30.0,
)

Both signatures are identical (client.py:149, client.py:405). Unlike the TypeScript client, base_url has a default, and that default is localhost, not 127.0.0.1.

Both are context managers and both expose close():

with CoreCruxClient("http://localhost:14800", token="...") as client:
    ...

async with AsyncCoreCruxClient("http://localhost:14800", token="...") as client:
    ...

__exit__ calls close(), which closes the underlying httpx connection pool (client.py:170); the async form awaits aclose() (client.py:426). If you construct a client without the with block, close it yourself or you leak sockets.

Headers are fixed at construction: Content-Type: application/json, plus Authorization: Bearer <token> when a token is given (client.py:30). No Accept header is set, the daemon returns JSON regardless.

Four things the constructor does not offer:

  1. No environment variables are read. The token must be passed explicitly.
  2. No transport injection. The httpx client is built internally, so there is no supported way to substitute an httpx.MockTransport for tests or to configure a proxy, retry policy, or client certificate. Patching client._client is your only route, and _client is private.
  3. No per-call timeout. timeout is set once for the whole client and applies flat, 30 seconds by default, connect and read alike.
  4. No retries. See chapter 4.

2.4 Method inventory

Sixteen API methods plus close(). The async class has exactly the same names; every one is awaited.

MethodHTTPReturnsNotes
healthz()GET /healthzdict[str, Any]Untyped
readyz()GET /readyzdict[str, Any]Raises on 503 rather than returning the failure body
version()GET /v1/versiondict[str, Any]Untyped
store_fact(fact)PUT /v1/factsFact
store_facts(facts)PUT /v1/facts/bulklist[Fact]Unwraps the facts key
get_fact(fact_id)GET /v1/facts/{id}Fact or NoneNone on 404
delete_fact(fact_id)DELETE /v1/facts/{id}boolReads deleted from the body; False on 404
get_facts_by_entity(entity)GET /v1/facts/entity/{e}list[Fact]Unwraps the facts key
query_facts(...)GET /v1/factsFactQueryResult
export_facts(...)GET /v1/facts/exportdict[str, Any]Untyped
put_session(session_id, state)PUT /v1/sessions/{id}/stateSessionState
get_session(session_id)GET /v1/sessions/{id}/stateSessionState or NoneNone on 404
text_search(...)POST /v1/query/text-searchTextSearchResult
text_search_expand(...)POST /v1/query/text-search/expanddict[str, Any]Untyped
graph_expand(...)POST /v1/query/graph-expanddict[str, Any]Untyped
time_range(...)POST /v1/query/time-rangedict[str, Any]Untyped
close()-NoneAsync form is awaited

Full signatures, as reported by inspect.signature on the published wheel:

healthz(self) -> dict[str, Any]
readyz(self) -> dict[str, Any]
version(self) -> dict[str, Any]

store_fact(self, fact: StoreFact) -> Fact
store_facts(self, facts: list[StoreFact]) -> list[Fact]
get_fact(self, fact_id: str) -> Fact | None
delete_fact(self, fact_id: str) -> bool
get_facts_by_entity(self, entity: str) -> list[Fact]

query_facts(
    self,
    query: str | None = None,
    *,
    entity: str | None = None,
    entity_prefix: str | None = None,
    top_k: int | None = None,
    token_budget: int | None = None,
) -> FactQueryResult

export_facts(
    self, *, since: str | None = None, cursor: str | None = None, limit: int | None = None
) -> dict[str, Any]

put_session(self, session_id: str, state: dict[str, Any]) -> SessionState
get_session(self, session_id: str) -> SessionState | None

text_search(
    self,
    tenant_id: str,
    query: str,
    *,
    limit: int = 10,
    token_budget: int | None = None,
    min_score: float | None = None,
    mode: str | None = None,
) -> TextSearchResult

text_search_expand(self, tenant_id: str, result_ids: list[dict[str, int]]) -> dict[str, Any]

graph_expand(
    self,
    tenant_id: str,
    seed_artifact_ids: list[int],
    *,
    edge_types: list[str] | None = None,
    max_hops: int = 2,
    budget: int = 50,
    min_confidence: float = 0.0,
    include_state: bool = False,
) -> dict[str, Any]

time_range(
    self,
    tenant_id: str,
    start_micros: int,
    end_micros: int,
    *,
    artifact_ids: list[int] | None = None,
    include_relations: bool = False,
    limit: int = 100,
) -> dict[str, Any]

close(self) -> None

Three behaviours the signatures do not show

  1. Path parameters are not URL-escaped. fact_id, entity and session_id are interpolated into f-strings directly (client.py:213, client.py:235). Colons in entity names, execplan:sdk-docs, are safe in a path segment. A /, ? or # in an entity name is not: the request will silently address a different route. Escape with urllib.parse.quote(entity, safe="") before you call. The TypeScript client does this for you.
  2. graph_expand and time_range always send their defaults. max_hops, budget, min_confidence, include_state, include_relations and limit are written into the body unconditionally (client.py:356, client.py:379). You cannot "leave a field out" to inherit a daemon-side default; the SDK's defaults win.
  3. A response body missing a required key raises KeyError, not CoreCruxError. _to_fact indexes fact_id, entity, key, value, confidence, stored_at, tokens and deleted directly (client.py:59). A 200 with an unexpected shape surfaces as a KeyError, which will not be caught by except CoreCruxError.

2.5 Dataclasses

All eight are plain @dataclass definitions with no validation (types.py:13):

@dataclass
class Fact:
    fact_id: str
    entity: str
    key: str
    value: str
    confidence: float
    stored_at: str
    tokens: int
    deleted: bool
    version: int
    source_receipt: str | None = None
    supersedes: str | None = None
    private: bool = False

@dataclass
class StoreFact:
    entity: str
    key: str
    value: str
    confidence: float = 1.0
    private: bool = False
    source_receipt: str | None = None

@dataclass
class TextSearchHit:
    segment_index: int
    doc_id: int
    score: float
    frame_offset: int
    token_count: int

@dataclass
class TextSearchCoverage:
    score: float
    gaps: list[dict[str, Any]] = field(default_factory=list)
    below_floor: int = 0

@dataclass
class TextSearchMeta:
    backend: str
    took_ms: int
    segments_searched: int
    total_docs: int
    total_candidates: int = 0

@dataclass
class TextSearchResult:
    results: list[TextSearchHit]
    coverage: TextSearchCoverage
    meta: TextSearchMeta
    tokens_used: int | None = None
    tokens_available: int | None = None
    results_omitted: int | None = None
    scan_mode: bool = False

@dataclass
class FactQueryResult:
    facts: list[Fact]
    total_tokens: int

@dataclass
class SessionState:
    session_id: str
    state: dict[str, Any]
    updated_at: str
    total_tokens: int
    expires_at: str | None = None

StoreFact is serialised by an explicit mapper, not dataclasses.asdict (client.py:122). Three consequences:

confidence and private are always sent, even when left at their defaults. A StoreFact with no explicit confidence transmits "confidence": 1.0 and "private": false.

private=True is rejected by the daemon. It answers 400 Bad Request with the detail private facts require MCP agent identity; HTTP /v1/facts does not support private=true (facts.rs:229), and the bulk route the same way (facts.rs:492). The field exists because it appears on responses.

There is no horizon_class and no actor. The daemon accepts both; this SDK cannot send either. Freshness-decay class and durable authorship must be set through another surface.

SessionState.state is annotated dict[str, Any] but is assigned straight from the response (client.py:76), so a session stored with a list or a string comes back as that type, contradicting the annotation. Nothing enforces it.

2.6 The error model

class CoreCruxError(Exception):
    status_code: int
    detail: str
    type: str          # RFC 7807 problem type URI; "" when absent

    def __init__(self, status_code: int, detail: str, type: str = "") -> None: ...

Source: errors.py:8. str(err) renders as CoreCrux error {status_code}: {detail}.

Raised for every response with status ≥ 400 (client.py:37). The body is parsed only when content-type starts with application/json or application/problem+json; otherwise the parsed body is an empty dict and detail falls back to the raw resp.text. type falls back to "".

There is no exception hierarchy. No NotFound, no RateLimited, no AuthError, one class, and you branch on status_code. The daemon's flattened problem extensions (code, missingScopes) are discarded: only detail and type are lifted off the body. If you need missingScopes you must re-read the response, which the SDK does not hand you.

from corecrux_client import CoreCruxClient, CoreCruxError

with CoreCruxClient("http://localhost:14800", token="...") as client:
    try:
        client.text_search("my-tenant", "deployment guide")
    except CoreCruxError as exc:
        if exc.status_code == 403:
            raise SystemExit(f"missing a scope: {exc.detail}")
        if exc.status_code == 503:
            raise SystemExit(f"daemon not ready: {exc.detail}")
        raise

Which statuses mean what is chapter 3; which are safe to retry is chapter 4.

2.7 Worked example, synchronous

import os
from urllib.parse import quote

from corecrux_client import CoreCruxClient, CoreCruxError, StoreFact

with CoreCruxClient(
    os.environ.get("CRUX_DAEMON_URL", "http://127.0.0.1:14800"),
    token=os.environ.get("CRUX_AGENT_TOKEN"),
    timeout=10.0,
) as client:
    fact = client.store_fact(
        StoreFact(
            entity="execplan:sdk-docs",
            key="decision:client-choice",
            value="Use corecrux-client for daemon access from Python.",
            confidence=0.9,
        )
    )
    print(fact.fact_id, fact.version)

    result = client.query_facts(
        "client choice",
        entity_prefix="execplan:",
        top_k=5,
        token_budget=500,
    )
    print(len(result.facts), result.total_tokens)

    # Escape the entity yourself, the SDK does not.
    for f in client.get_facts_by_entity(quote("execplan:sdk-docs", safe="")):
        print(f.key, f.value)

    hits = client.text_search("my-tenant", "deployment architecture", limit=10, token_budget=4096)
    print(hits.coverage.score, hits.meta.backend)

    try:
        client.get_fact("does-not-exist")   # returns None, does not raise
    except CoreCruxError as exc:
        print(exc.status_code, exc.detail)

    print(client.delete_fact(fact.fact_id))

2.8 Worked example, asynchronous

The async class mirrors the sync one name for name. Only async with and await change.

import asyncio
import os

from corecrux_client import AsyncCoreCruxClient, StoreFact


async def main() -> None:
    async with AsyncCoreCruxClient(
        os.environ.get("CRUX_DAEMON_URL", "http://127.0.0.1:14800"),
        token=os.environ.get("CRUX_AGENT_TOKEN"),
    ) as client:
        facts = await client.store_facts(
            [
                StoreFact(entity="bench:lme-s", key="metric:recall", value="0.81"),
                StoreFact(entity="bench:lme-s", key="metric:latency_ms", value="430"),
            ]
        )
        print([f.fact_id for f in facts])

        state = await client.put_session("session-1", {"step": 3, "topic": "onboarding"})
        print(state.session_id, state.total_tokens)

        restored = await client.get_session("session-1")
        if restored is not None:
            print(restored.state)


asyncio.run(main())

There is no async iterator over export_facts. Paginate by hand:

cursor = None
while True:
    page = client.export_facts(cursor=cursor, limit=500)
    for raw in page["facts"]:
        ...
    if not page.get("has_more"):
        break
    cursor = page["next_cursor"]

export_facts returns a raw dict, so has_more and next_cursor are dictionary keys, not attributes. Exports include tombstones, deleted facts appear with deleted: true.

2.9 What is not covered

Daemon capabilityState in this SDK
SSE, GET /v1/events/streamNot implemented. The TypeScript client has it; Python does not
Receipt reads, GET /v1/receipts/{id} and its signature and verification sub-routesNot implemented
Typed graph_expand and time_range responsesReturned as dict. The TypeScript client types both
horizon_class, actor on writesNot implemented
Retry, backoff, jitterNot implemented
Custom transport or mock injectionNot exposed

For any of these, call the HTTP API directly with httpx. The daemon's OpenAPI document is at GET /v1/openapi.json.

2.10 The reproducibility gate

This is the strongest trust property in either SDK estate, and it is worth understanding rather than taking on faith.

The Python release workflow builds the distribution twice into two separate directories, with SOURCE_DATE_EPOCH pinned to the commit timestamp, takes sha256sum over each output directory, and runs diff -u over the two sums. If any byte of any artifact differs between the two builds, diff exits non-zero and the workflow fails (sdk-python.yml:43):

export SOURCE_DATE_EPOCH="$(git log -1 --format=%ct)"
python -m build --no-isolation --outdir "$first"
python -m build --no-isolation --outdir "$second"
(cd "$first"  && sha256sum *) >"…/python-sdk-first.sha256"
(cd "$second" && sha256sum *) >"…/python-sdk-second.sha256"
diff -u "…/python-sdk-first.sha256" "…/python-sdk-second.sha256"
cp "$first"/* dist/

The verified directory is then uploaded as a workflow artifact, and the publish job downloads that artifact rather than rebuilding (sdk-python.yml:76). The bytes that reach PyPI are the bytes that were checked.

Three supporting properties make that check meaningful rather than decorative:

PropertyMechanismSource
The build toolchain cannot driftbuild==1.5.0 and hatchling==1.31.0 are pinned exactly, and --no-isolation stops pip resolving anything elsesdk-python.yml:41
A build run cannot publishTop-level permissions: contents: read; only the publish job requests id-token: write, and only on an sdk-python-v* tagsdk-python.yml:16, sdk-python.yml:67
The tag cannot lie about the versionThe tag suffix is compared to pyproject.toml and the run fails on mismatchsdk-python.yml:31
No registry secret exists to stealPyPI Trusted Publishing over OIDC via pypa/gh-action-pypi-publish@v1.14.1; no PYPI_TOKENsdk-python.yml:81

What this does not prove. It proves the build step is deterministic and that the published artifact is the one the workflow inspected. It does not prove the source was reviewed, that the package does what its documentation says, or that the workflow definition itself was not changed in the same commit. It is a supply-chain property, not a correctness property. Verify what you install with pip download --no-deps corecrux-client and read it.

The equivalent TypeScript lane packs once and publishes that exact tarball with --provenance - sound, but with no second build to diff against. The SDKCrux public lane has no reproducibility check at all; see 5.9.

2.11 Versioning and release

The policy is shared with the TypeScript SDK; see 1.12 for the full table. The Python specifics:

RuleDetailSource
Publish triggerPush of sdk-python-vX.Y.Z, matching pyproject.toml exactlysdk-python.yml:31
Manual dispatchBuild-only. workflow_dispatch cannot reach the publish jobsdk-python.yml:67
Daemon tagsv* tags build and package, but never publishsdk-python.yml:4
ImmutabilityDuplicate uploads fail closed. skip-existing is prohibited by policysdk-release-lifecycle.md:76
StabilityPre-1.0: a minor may break, a patch is safesdk-release-lifecycle.md:27

As with TypeScript, the package declares no daemon version it was tested against, though the policy calls for one (sdk-release-lifecycle.md:22). Call version() and compare.

Sources