Python SDK

Auto-instrumentation and manual tracing for OpenAI, Anthropic, and Google LLM calls.

Package: kalibr
Version: 1.3.0
Python: >=3.9

Installation

pip install kalibr

With framework integrations:

# LangChain
pip install kalibr[langchain]

# CrewAI  
pip install kalibr[crewai]

# OpenAI Agents SDK
pip install kalibr[openai-agents]

# All integrations
pip install kalibr[integrations]

Auto-Instrumentation

Just import kalibr — it automatically instruments OpenAI, Anthropic, and Google SDKs on import:

import kalibr  # Auto-instruments on import
from openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)
# ✅ Trace automatically captured and sent

Auto-instrumentation happens when KALIBR_AUTO_INSTRUMENT=true (the default). To disable:

export KALIBR_AUTO_INSTRUMENT=false

Manual Instrumentation

For explicit control:

from kalibr import auto_instrument

# Instrument specific providers
auto_instrument(["openai", "anthropic", "google"])

# Check what's instrumented
from kalibr import get_instrumented_providers
print(get_instrumented_providers())  # ['openai', 'anthropic']

@trace Decorator

For custom functions or when you need more control:

from kalibr import trace

@trace(operation="summarize", provider="openai", model="gpt-4o")
def summarize_text(text: str) -> str:
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Summarize: {text}"}]
    )
    return response.choices[0].message.content

Decorator Parameters

The decorator captures duration, calculates cost, and handles errors automatically.

Intelligence API

Query Kalibr for optimal model, tool, and parameter recommendations based on historical outcomes:

from kalibr import get_policy, report_outcome, get_trace_id

# 1. Get best model, tool, and params for your goal
policy = get_policy(
    goal="book_meeting",
    include_tools=True,
    include_params=["temperature"],
)
model = policy["recommended_model"]        # e.g., "gpt-4o"
provider = policy["recommended_provider"]  # e.g., "openai"
tool = policy.get("recommended_tool")      # e.g., "calendar_api"
params = policy.get("recommended_params")  # e.g., {"temperature": "0.3"}

# 2. Execute with recommended configuration
response = openai.chat.completions.create(model=model, ...)

# 3. Report outcome with full context
report_outcome(
    trace_id=get_trace_id(),
    goal="book_meeting",
    success=True,
    tool_id=tool,
    execution_params=params,
)

get_policy() UPDATED

policy = get_policy(
    goal="book_meeting",           # Required
    task_type="scheduling",        # Optional filter
    constraints={                  # Optional
        "max_cost_usd": 0.05,
        "max_latency_ms": 3000,
        "min_confidence": 0.7,
    },
    window_hours=168,              # Default: 1 week
    include_tools=True,            # Get tool recommendations
    include_params=["temperature"], # Get param recommendations
)

# Response includes model + tool + param recommendations
{
    "recommended_model": "gpt-4o",
    "recommended_provider": "openai",
    "outcome_success_rate": 0.87,
    "recommended_tool": "calendar_api",
    "tool_success_rate": 0.91,
    "recommended_params": {"temperature": "0.3"},
    ...
}

report_outcome() UPDATED

report_outcome(
    trace_id="abc123",             # Required
    goal="book_meeting",           # Required
    success=True,                  # Required
    score=0.9,                     # Optional: quality 0-1
    failure_reason="timeout",      # Optional: if failed
    tool_id="calendar_api",        # Optional: tool used
    execution_params={             # Optional: params used
        "temperature": "0.3",
        "timeout": "30"
    },
    metadata={"key": "value"},     # Optional
)

get_recommendation()

from kalibr import get_recommendation

rec = get_recommendation(
    task_type="summarization",
    optimize_for="balanced",  # cost, quality, latency, balanced, cost_efficiency
)

Context Functions

Access trace context for linking spans:

from kalibr import get_trace_id, get_parent_span_id, new_trace_id

# Get current trace ID (from active span)
trace_id = get_trace_id()

# Get parent span ID (for nesting)
parent_id = get_parent_span_id()

# Generate a new trace ID
new_id = new_trace_id()

Environment Variables

CLI

Run Python scripts with auto-tracing:

# Run with tracing
kalibr run my_script.py

# Show version
kalibr version

All Exports

OpenTelemetry

The SDK uses OpenTelemetry for tracing. Configure the collector:

from kalibr import setup_collector

setup_collector(
    service_name="my-service",
    otlp_endpoint="http://localhost:4317",  # Optional OTLP collector
    file_export=True,    # Write to /tmp/kalibr_otel_spans.jsonl
    console_export=False # Print to console (debugging)
)

Next Steps

• Framework Integrations — LangChain, CrewAI, OpenAI Agents
• Intelligence API — Full execution routing (model + tool + params)
• TypeScript SDK — For Node.js applications