Python SDK Guide

The KLA Python Telemetry SDK (kla-otel-python) provides automatic tracing, budget guardrails, and compliance logs for Python frameworks. It is designed to work with FastAPI, LangChain, and standard HTTP execution patterns.

---

Installation

Install the package via pip or include it in your environment definitions:

pip install kla-otel-python

---

Auto-Instrumentation (FastAPI)

For web APIs routing model inputs, instantiate the KLA Middleware:

# main.py
import os
from fastapi import FastAPI
from kla_otel import KLATelemetryMiddleware

app = FastAPI()

# Mount telemetry collector middleware
app.add_middleware(
    KLATelemetryMiddleware,
    tenant_id=os.getenv("KLA_TENANT_ID"),
    api_key=os.getenv("KLA_API_KEY"),
    endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://api.kla.digital/collector"),
    mask_pii=True
)

@app.get("/")
def read_root():
    return {"status": "governed"}

---

LangChain Integration

If you use LangChain to orchestrate prompts and memory, enable the KLA callback handler to automatically trace agents, tools, and token boundaries:

from langchain_openai import ChatOpenAI
from kla_otel.callbacks import KLACallbackHandler

# Load the KLA callback tracer
kla_callback = KLACallbackHandler(
    agent_id="agt_9f81a7",
    tenant_id="kpmg-reporting"
)

# Attach callback to your model execution
model = ChatOpenAI(
    model_name="gpt-4o",
    temperature=0,
    callbacks=[kla_callback]
)

response = model.invoke("Analyze customer invoice records for Q1.")

---

Dynamic Action Evaluations

For manual safety overrides (such as modifying files or requesting API keys), evaluate action objects directly:

from kla_otel import KLACheckpoint

checkpoint = KLACheckpoint(agent_id="agt_9f81a7")

def delete_user_account(user_id: str):
    # Perform safety policy evaluation
    decision = checkpoint.evaluate_action(
        action="delete_account",
        context={"user_id": user_id}
    )

    if decision.is_approved():
        db.delete(user_id)
        checkpoint.log_success(user_id=user_id)
    else:
        # Action is suspended, checking Decision Desk queue
        checkpoint.log_escalation(
            user_id=user_id,
            reason="Requires tenant owner approval"
        )
        raise PermissionError("Deletion pending human approval desk review")