korasafe-sdk 0.2.0

KoraSafe Python SDK — inline guardian scans + chain-of-thought trace capture for Python agents

KoraSafe Python SDK

Two surfaces in one package:

• KoraSafeClient — inline guardian inspection (scan, gate, submit findings). Metadata-only transport.
• kora_trace — capture agent chain-of-thought (plan, LLM calls, tool calls, reasoning, human approvals) and ship to the KoraSafe audit log.

pip install korasafe-sdk
export KORASAFE_API_KEY=ks_live_...

kora_trace — chain-of-thought capture

10-line FastAPI example:

from fastapi import FastAPI
from korasafe import init_trace, kora_trace

init_trace()
app = FastAPI()

@app.post("/classify")
def classify(text: str) -> dict[str, str]:
    with kora_trace.run("classify_claim"):
        kora_trace.plan(["look up policy", "score risk", "route"])
        kora_trace.llm_call(provider="openai", model="gpt-4o", input=text, output="tier=gold", input_tokens=120, output_tokens=30)
        kora_trace.tool_call(name="policy_lookup", parameters={"id": "pol-7"}, response={"tier": "gold"}, status="ok")
        kora_trace.human_approval(reviewer_id="user-42", decision="approved")
        return {"tier": "gold"}

Events appear in the KoraSafe audit log within 5 seconds.

API

Method	Purpose
kora_trace.run(task_name)	Context manager. Opens a trace; run_start + run_end events bracket the block. Nested method calls auto-attach via contextvars.
@kora_trace.trace(task_name)	Decorator equivalent of run. Works on sync and async functions.
kora_trace.plan(steps, reasoning=None)	Log initial plan. steps can be strings or {step, description} dicts.
kora_trace.llm_call(provider, model, input, output, input_tokens=, output_tokens=, total_tokens=, cost_usd=, duration_ms=, status=, error=)	Log an LLM invocation. Tokens auto-sum if total_tokens omitted.
kora_trace.tool_call(name, parameters=, response=, status=, duration_ms=, error=)	Log a tool or external API call.
kora_trace.human_approval(reviewer_id, decision, notes=, approval_chain=)	Log a HITL decision.
kora_trace.flush()	Force-flush buffered events.
kora_trace.close()	Drain buffer + close HTTP client.

init_trace(**kwargs) reinitializes the singleton with overrides (api_key, endpoint, batch_size=10, flush_interval_s=5, timeout_s=10, max_retries=3, disabled=False, logger=, http_client=).

Calling plan / llm_call / tool_call / human_approval outside a run() context raises RuntimeError — wrap your agent loop first.

KoraSafeClient — guardian inspection

from korasafe import KoraSafeClient, withKoraSafeScan

client = KoraSafeClient()


@withKoraSafeScan(client=client, context={"system_id": "claims-agent"})
def answer_claim(prompt: str) -> str:
    return "approved"

result = client.scan("Does this contain PII?", {"system_id": "claims-agent"})
gate = client.gate({"action": "payment_approval", "risk_tier": "high"})
finding = client.submit_finding({"guardian_id": "pii", "title": "PII found", "severity": "high"})

Raw strings passed to scan() or the decorator are hashed locally. The SDK sends content hash, byte length, direction, surface, labels, and caller metadata rather than prompt or response bodies.

Frameworks

LangChain:

from korasafe import KoraSafeCallback, KoraSafeClient

callbacks = [KoraSafeCallback(client=KoraSafeClient(), context={"system_id": "claims-agent"})]

LlamaIndex:

from korasafe import KoraSafeClient, KoraSafeLlamaIndexMiddleware

query_engine = KoraSafeLlamaIndexMiddleware(KoraSafeClient()).wrap_query_engine(query_engine)

Development

cd packages/sdk-python
python -m pip install -e ".[dev]"
ruff check .
mypy .
coverage run -m pytest
coverage report
python -m build

Publishing uses GitHub OIDC trusted publishing to PyPI from sdk-python-v* tags.