kyrodb 1.0.3

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

Requires Python 3.10 or newer.

pip install kyrodb

For local development from this repository:

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

Quickstart

Use the managed runtime handoff from the KyroDB console:

1. Open Console -> Runtime -> Backend environment.
2. Click Create env command.
3. Run the generated one-time command from your backend project. It writes .env.kyrodb.
4. Load .env.kyrodb only in server-side code before calling KyroDBClient.from_env().

For local development, use your normal server env loader, such as python-dotenv, direnv, your process manager, or your deployment platform's secret store. Runtime bearer tokens must never be placed in browser JavaScript.

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 public body, top_k, embedding, metadata, filter, replay, and recent safety 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.