Python SDK

Use the Python SDK from trusted server-side Python code that can safely hold runtime bearer tokens.

The SDK is a typed HTTP client. It does not run vector search, generate embeddings, cache context, or create freshness proof locally.

Install

python -m pip install --upgrade pip
python -m pip install kyrodb==1.0.3

For development inside the SDK repository:

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

Configure

Get runtime credentials from Console -> Runtime -> Backend environment after the runtime has a ready endpoint. Ready means runtime health is healthy or degraded. Click Issue env command and run the generated command from your backend project; it writes .env.kyrodb with KYRODB_BASE_URL, KYRODB_DATA_PLANE_TOKEN, KYRODB_OBSERVABILITY_TOKEN, and non-secret KYRODB_EMBEDDING_DIMENSIONS metadata.

KyroDBClient.from_env() reads required KYRODB_BASE_URL and KYRODB_DATA_PLANE_TOKEN, plus optional KYRODB_OBSERVABILITY_TOKEN, KYRODB_SHADOW_SESSION_ID, and KYRODB_ALLOW_INSECURE_HTTP. SDKs read environment variables from the process, not from .env.kyrodb directly. For local development, load the generated values into the process environment before calling from_env().

set -a
. ./.env.kyrodb
set +a
python worker.py

In deployed environments, copy the generated values into your server, worker, or managed secrets configuration. Use constructor arguments if you prefer explicit configuration:

Run the Runtime page smoke test once after creating .env.kyrodb. It uses tenant_id="demo", namespace="default", and a synthetic vector to verify connectivity, auth, embedding dimension, and trace output. Replace those demo values with your application scope and embedding pipeline before production traffic.

import os
from kyrodb import KyroDBClient
 
client = KyroDBClient(
    base_url=os.environ["KYRODB_BASE_URL"],
    data_plane_token=os.environ["KYRODB_DATA_PLANE_TOKEN"],
    observability_token=os.environ.get("KYRODB_OBSERVABILITY_TOKEN"),
)

Security defaults:

• HTTPS is required outside loopback unless allow_insecure_http=True is explicit.
• Redirects are disabled.
• Data-plane and observability tokens are separate values in the SDK configuration.
• Tokens are redacted from reprs, SDK errors, and transport messages.
• Side-effecting calls are not retried automatically.

Retrieve

from kyrodb import KyroDBClient, RuntimeFilter, Scope
 
client = KyroDBClient.from_env()
 
question = "How do refunds work for annual plans?"  # example user input
request_id = "retrieve-req-2026-05-01-001"  # unique per retryable request
embedding = embed_user_question(question)
 
packet = client.retrieve(
    query_embedding=embedding,
    scope=Scope(tenant_id="acme", namespace="kb"),
    filters=RuntimeFilter.exact("category", "billing"),
    top_k=3,
    freshness_mode="strict",
    include_content=True,
    idempotency_key=request_id,
)
 
print(packet.status)
print(packet.trace_id)

embedding must come from your embedding pipeline and must match the configured runtime connector dimension. The SDK does not generate embeddings.

Handle packet status explicitly:

if packet.status == "stale_blocked":
    raise RuntimeError("KyroDB refused to serve stale context")
 
if packet.warnings:
    print("KyroDB warnings:", packet.warnings)

Idempotency:

• retrieve, invalidate, upsert_document, and delete_document accept idempotency_key=....
• ingest_change_event does not accept an idempotency key; it uses required source_event_id for retry deduplication.
• record_feedback is intentionally signal-only and does not accept an idempotency key.

upsert_document and delete_document require an explicitly certified and provisioned write-through runtime. Default managed pgvector runtimes use event-feed freshness instead.

Change events

When source knowledge changes outside KyroDB, send a scoped change event after your application commits the source write. This is the default managed pgvector pattern: write Postgres directly, commit, then send KyroDB the scoped event. Use a stable source_event_id from your database outbox, CMS event, or write transaction; it is the event-feed idempotency key for identical retries.

from datetime import datetime, timezone
 
from kyrodb import ChangeEvent, ChangeScope, ChangeTarget, ChangeTargetType, ChangeType
 
event_timestamp = datetime(2026, 5, 1, tzinfo=timezone.utc)  # use the committed source event timestamp
 
event = ChangeEvent(
    target=ChangeTarget(type=ChangeTargetType.NAMESPACE, namespace="kb"),
    change_type=ChangeType.CONTENT_UPDATED,
    scope=ChangeScope(tenant_id="acme", namespace="kb"),
    source_event_id="cms-evt-2026-05-01-001",
    timestamp=event_timestamp,
)
 
client.ingest_change_event(event)

Use the trace id from retrieval responses in Trace lookup after the next agent request to confirm the runtime observed the freshness boundary you expected.

Observe

Observability and admin evidence methods require KYRODB_OBSERVABILITY_TOKEN or an explicit observability_token.

trace = client.observability.get_trace(packet.trace_id)
diagnosis = client.observability.diagnose_trace(packet.trace_id)
feedback = client.observability.get_feedback(packet.trace_id)
proof = client.observability.context_proof_report(recent=100)
 
print(trace.status)
print(diagnosis.kind, diagnosis.summary)
print(feedback)
print(proof)

Record feedback when an answer was useful, stale, irrelevant, or wrong-scope:

client.record_feedback(
    trace_id=packet.trace_id,
    signal="stale",
    comment="Agent answered from the old refund policy.",
)

Admin evidence

Admin evidence routes should stay in trusted backend jobs, CLI flows, or internal operations. The Python SDK uses the configured observability token for this surface.

health = client.admin.health()
capture = client.admin.export_replay_capture(recent=50)
bundle = client.admin.build_proof_bundle(recent=50)
root_cause = client.admin.build_root_cause_report(recent=50)
shadow = client.admin.create_shadow_session(
    schema_version=1,
    replay_primer=capture.replay_primer,
)

Use admin workflows to build partner proof, root-cause reports, replay exports, and shadow sessions. Keep data-plane and observability tokens in backend or internal operational code.