kyrodb 1.0.2

Python SDK for KyroDB, the freshness-aware context runtime for AI systems.

KyroDB Python SDK

Typed Python client for KyroDB, the freshness-aware context runtime for AI systems.

KyroDB fixes stale agent context by serving retrieval through an explicit freshness, invalidation, trace, proof, and replay contract. This SDK is intentionally only a client for a running KyroDB Runtime. It does not implement vector search, local caching, freshness proofs, or planner logic in Python.

Install

pip install kyrodb

For local development from this repository:

python -m pip install --upgrade pip
python -m pip install -e ".[dev]"

Quickstart

Set runtime credentials in your backend environment. Do not ship these tokens to browser code.

export KYRODB_BASE_URL="https://runtime.example.com"
export KYRODB_DATA_PLANE_TOKEN="data-plane-token"
export KYRODB_OBSERVABILITY_TOKEN="observability-token"

from kyrodb import KyroDBClient, Scope
from your_app.embeddings import embed_query


# KyroDB does not create embeddings. Reuse the same embedding model that writes
# vectors into your knowledge store; the dimension must match the runtime
# connector, for example pgvector vector(1536).
query_embedding = embed_query("How do refunds work?")
scope = Scope(tenant_id="acme", namespace="kb")

with KyroDBClient.from_env() as client:
    packet = client.retrieve(
        query_embedding=query_embedding,
        scope=scope,
        top_k=3,
        freshness_mode="balanced",
    )

    trace = client.observability.get_trace(packet.trace_id)
    diagnosis = client.observability.diagnose_trace(packet.trace_id)

print(
    {
        "status": packet.status.value,
        "trace_id": packet.trace_id,
        "execution_path": trace.execution_path.value,
        "diagnosis": diagnosis.kind.value,
    }
)

from_env() reads KYRODB_BASE_URL, KYRODB_DATA_PLANE_TOKEN, optional KYRODB_OBSERVABILITY_TOKEN, optional KYRODB_SHADOW_SESSION_ID, and optional KYRODB_ALLOW_INSECURE_HTTP.

Security Defaults

• HTTPS is required for non-loopback hosts unless allow_insecure_http=True is explicit.
• Redirects are disabled by default to prevent replaying captured documents or bearer tokens to another origin.
• Data-plane and observability/admin tokens are separate by default.
• Tokens are never included in repr, logs, or SDK exceptions.
• Data-plane-only clients cannot access client.observability or client.admin until an observability token is configured.
• Side-effecting calls are not retried automatically because retrieval and mutations write trace, replay, and freshness evidence.
• Responses are read through a bounded stream to protect callers from unbounded memory use.
• Requests are preflighted against the runtime's default body, top_k, embedding, metadata, filter, replay, and recent limits before they leave the process.
• Shadow-session IDs are validated before being sent as kyro-shadow-session-id.

API Shape

Data-plane methods live on KyroDBClient:

• retrieve
• invalidate
• ingest_change_event
• upsert_document
• delete_document
• record_feedback

Usage-metered calls (retrieve, invalidate, upsert_document, and delete_document) accept idempotency_key="..." for server-side retry identity. Change events use their required source_event_id field as the runtime idempotency key.

For the default pgvector event_feed deployment model, your application keeps writing to its own database and emits a change event after each relevant insert/update/delete:

from datetime import datetime, timezone

from kyrodb import ChangeEvent, ChangeScope, ChangeTarget, ChangeTargetType, ChangeType

client.ingest_change_event(
    ChangeEvent(
        source_event_id="documents:doc_123:v42",
        change_type=ChangeType.CONTENT_UPDATED,
        target=ChangeTarget(type=ChangeTargetType.DOCUMENT, doc_id="doc_123"),
        scope=ChangeScope(tenant_id="acme", namespace="kb"),
        timestamp=datetime.now(timezone.utc),
    )
)

Use a stable source_event_id from your source system or outbox so retries are deduplicated by the runtime.

Observability methods live under client.observability:

• get_feedback
• get_trace
• diagnose_trace
• context_proof_report

Admin/evidence methods live under client.admin:

• health
• build_proof_bundle
• build_root_cause_report
• diff_replay_runs
• run_counterfactual_replay
• create_shadow_session
• export_replay_capture

Async parity is available through AsyncKyroDBClient.

Counterfactual replay and replay diff calls use typed request models at the SDK boundary while keeping replay artifacts as forward-compatible JSON objects:

from kyrodb import CounterfactualIntervention, CounterfactualReplayScenario

capture = client.admin.export_replay_capture(recent=50)
scenario = CounterfactualReplayScenario(
    name="drop retry noise",
    interventions=[CounterfactualIntervention.drop_step(0)],
)
summary = client.admin.run_counterfactual_replay(capture=capture, scenario=scenario)

Shadow replay workflows can pass the server-returned session ID explicitly:

shadow = KyroDBClient(
    base_url="https://candidate-runtime.example.com",
    data_plane_token="candidate-data-token",
    observability_token="candidate-observability-token",
    shadow_session_id="session-id-from-create-shadow-session",
)

Development

python -m pytest
python -m ruff format .
python -m ruff check .
python -m mypy src/kyrodb
python scripts/check_release.py --no-build

Optional live tests should be wired to a running KyroDB Runtime with:

• KYRODB_BASE_URL
• KYRODB_DATA_PLANE_TOKEN
• KYRODB_OBSERVABILITY_TOKEN
• KYRODB_TEST_QUERY_EMBEDDING
• KYRODB_SHADOW_SESSION_ID
• KYRODB_ALLOW_INSECURE_HTTP
• KYRODB_LIVE_TESTS=1

KYRODB_LIVE_TESTS=1 \
KYRODB_BASE_URL=https://runtime.example.com \
KYRODB_DATA_PLANE_TOKEN=... \
KYRODB_OBSERVABILITY_TOKEN=... \
KYRODB_TEST_QUERY_EMBEDDING='[0.0123, -0.0456, 0.0789]' \
python -m pytest -m live

Additional live gates are intentionally explicit:

• KYRODB_LIVE_EVENT_FEED_TESTS=1
• KYRODB_LIVE_WRITE_THROUGH_TESTS=1
• KYRODB_LIVE_REPLAY_TESTS=1

KYRODB_TEST_QUERY_EMBEDDING must come from the same embedding pipeline used by the live runtime connector and must match its configured vector dimension. Live tests default to tenant default and namespace kb; override them with KYRODB_TEST_TENANT_ID and KYRODB_TEST_NAMESPACE when validating another seeded fixture.

Examples

The examples/ directory contains small production-shaped flows:

• retrieve.py: retrieve fresh context and print trace evidence identifiers.
• trace_diagnosis.py: inspect why a trace was stale, degraded, or fresh.
• proof_bundle.py: build a redacted proof bundle from retained replay evidence.
• design_partner_shadow.py: create a candidate shadow session from replay primer state.

examples/retrieve.py intentionally requires a real embedding from your embedding pipeline. The vector below shows the JSON shape only; use the full dimension configured in your runtime connector.

KYRODB_EXAMPLE_QUERY_EMBEDDING='[0.0123, -0.0456, 0.0789]' \
KYRODB_EXAMPLE_TENANT_ID=acme \
KYRODB_EXAMPLE_NAMESPACE=kb \
python examples/retrieve.py

Release

Release candidates must pass the required local gate in RELEASE.md; run the optional dependency audit when validating a release environment with the security extra installed.