API Reference

Router

What this does: Routes requests to the best model based on learned outcomes. Works across any modality.

When to use: Create one Router per goal. Reuse for multiple requests in the same thread/async context.

from kalibr import Router

router = Router(
    goal="extract_company",           # required
    paths=["gpt-4o", "claude-sonnet-4-20250514"],  # required
    success_when=lambda x: len(x) > 0,  # optional, bool
    score_when=lambda x: min(1.0, len(x) / 500),  # optional, float 0-1
    auto_register=True,               # optional, default True
)

import { Router } from '@kalibr/sdk';

const router = new Router({
  goal: 'extract_company',           // required
  paths: ['gpt-4o', 'claude-sonnet-4-20250514'], // required
  successWhen: (output) => output.length > 0, // optional, boolean
  scoreWhen: (output) => Math.min(1.0, output.length / 500), // optional, float 0-1
  autoRegister: true,                // optional, default true
});

Required arguments

Optional arguments

Thread Safety

Router is NOT thread-safe. Internal state (trace_id) will be corrupted if used across threads/async contexts.

• Python: Create one Router instance per thread.
• TypeScript: Create one Router instance per async context (request handler, worker, etc.)

Common mistakes

• Sharing one Router across concurrent threads or async contexts. Create a separate instance per thread or request handler.
• Forgetting to call report()
• Using same goal for different tasks
• Using Router across threads/async contexts -- Create separate instances

Router.completion()

What this does: Makes a completion request with intelligent routing.

When to call: Every time your agent needs a model response for this goal.

Example

response = router.completion(
    messages=[
        {"role": "system", "content": "Extract company names."},
        {"role": "user", "content": "Hi, I'm Sarah from Stripe."}
    ],
    max_tokens=100
)
print(response.choices[0].message.content)

const response = await router.completion(
  [
    { role: 'system', content: 'Extract company names.' },
    { role: 'user', content: "Hi, I'm Sarah from Stripe." },
  ],
  { maxTokens: 100 }
);

console.log(response.choices[0].message.content);

Required arguments

Optional arguments

Common mistakes

• Passing model in kwargs (Kalibr picks the model; use force_model/forceModel to override)
• Not handling exceptions (provider errors still raise)

Exceptions

Provider errors propagate to caller:

• Python: openai.OpenAIError, anthropic.AnthropicError
• TypeScript: OpenAI.APIError, Anthropic.APIError

Intelligence service failures do NOT raise -- Router falls back to first path.

HealConfig

What this does: Configures how router.completion(healing=True) retries failed calls. Pass via heal_config=.

When to use: When the default heal budget or gate set does not fit the goal. For example: to enable an LLM-judge quality gate, disable meta-prompt repair, or change the retry count.

from kalibr import Router, HealConfig

config = HealConfig(
    max_retries=2,            # heal attempts before giving up
    gate2_enabled=True,       # LLM-judge quality gate
    meta_prompt_enabled=True, # repair meta prompt before model swap
)

response = router.completion(
    messages=[{"role": "user", "content": "Summarize: ..."}],
    healing=True,
    heal_config=config,
)

Fields

Router.pipeline()

What this does: Runs a sequence of routed, gated, healed steps as a single call. Each step picks its own goal and messages and runs the full self-healing loop independently.

When to use: Multi-step agent workflows (research → outreach, extract → enrich → classify) where you want every intermediate step to be evaluated and healed without writing orchestration code.

result = router.pipeline(
    [
        {"goal": "research", "messages": [...]},
        {"goal": "outreach_generation", "messages": [...], "chain": True},
    ],
    healing=True,
    pipeline_id="sales-outreach-prod",
)

Required arguments

Optional arguments

Returns: Pipeline result with per-step outputs and metadata. If a step fails after exhausting retries, the pipeline returns a partial result with the failure attached.

as_langchain()

Returns a LangChain-compatible LLM that uses Kalibr for routing. Use this to integrate with frameworks like CrewAI and LangChain.

from kalibr import Router

router = Router(
    goal="my_task",
    paths=["gpt-4o", "claude-sonnet-4-20250514"]
)

# Returns a LangChain BaseChatModel
llm = router.as_langchain()

# Use with any LangChain chain or CrewAI agent
from langchain_core.prompts import ChatPromptTemplate
chain = ChatPromptTemplate.from_template("{input}") | llm

Note: You still need to call router.report() after the chain completes to report the outcome.

Router.execute()

What this does: Routes any HuggingFace task with the same outcome-learning loop as completion(). Works for transcription, image generation, embeddings, classification, and all 17 HuggingFace task types.

When to call: When your agent needs to run a non-chat model task (audio, image, embedding, classification).

Returns: Task-native response (transcription text, PIL image, embedding vector, classification labels, etc.)

Examples

# Transcription
result = router.execute(task="automatic_speech_recognition", input_data=audio_bytes)

# Image generation
result = router.execute(task="text_to_image", input_data="a product photo of a laptop")

# Embeddings
result = router.execute(task="feature_extraction", input_data="search query text")

# Classification
result = router.execute(task="text_classification", input_data="This product is amazing!")

Supported task types: chat_completion, text_generation, summarization, translation, fill_mask, table_question_answering, automatic_speech_recognition, text_to_speech, audio_classification, text_to_image, image_to_text, image_classification, image_segmentation, object_detection, feature_extraction, text_classification, token_classification

Router.synthesize()

What this does: Routes a text-to-speech (TTS) request to the best voice model, with the same outcome-learning loop as completion(). Supports ElevenLabs, OpenAI TTS, and Deepgram Aura.

router = Router(
    goal="narrate",
    paths=["tts-1", "eleven_multilingual_v2"],  # OpenAI TTS or ElevenLabs
)
result = router.synthesize("Hello world", voice="alloy")
# result.audio     , audio bytes
# result.cost_usd  , cost of this call
# result.model     , which model was selected
# result.kalibr_trace_id , for manual report() if needed

Provider detection: tts-* / whisper-* -- OpenAI · eleven_* -- ElevenLabs · nova-* / aura-* -- Deepgram

Required env vars: OPENAI_API_KEY for OpenAI TTS · ELEVENLABS_API_KEY for ElevenLabs · DEEPGRAM_API_KEY for Deepgram

Install: pip install kalibr[voice] (includes ElevenLabs + Deepgram SDK)

Router.transcribe()

What this does: Routes a speech-to-text (STT) request to the best model. Supports OpenAI Whisper and Deepgram Nova.

router = Router(
    goal="transcribe_meeting",
    paths=["whisper-1", "nova-2"],  # OpenAI Whisper or Deepgram Nova
)
result = router.transcribe(audio_bytes, audio_duration_seconds=120.0)
# result.text      , transcribed text
# result.cost_usd  , cost of this call
# result.kalibr_trace_id , for manual report() if needed

Provider detection: whisper-* -- OpenAI · nova-* / enhanced / base -- Deepgram

Router.report()

What this does: Reports outcome for the last completion. This is how Kalibr learns.

When to call: After you know whether the task succeeded or failed.

Example

# Success
router.report(success=True)

# Failure with reason
router.report(success=False, reason="invalid_json")

# Success with quality score
router.report(success=True, score=0.8)

# Failure with structured category
router.report(success=False, failure_category="timeout", reason="Provider timed out after 30s")

// Success
await router.report(true);

// Failure with reason
await router.report(false, 'invalid_json');

// Success with score
await router.report(true, undefined, 0.8);

Required arguments

Optional arguments

Common mistakes

• Calling report() multiple times for one completion (second call is ignored)
• Not calling report() at all (routing never improves)
• Reporting success based on "response exists" instead of "task actually worked"

When to call

• After you know if task succeeded/failed
• Once per completion() call
• For multi-turn, report once at end

Validation

Calling report() without a prior completion() raises an error. Calling report() twice for the same completion logs a warning and ignores the second call.

get_policy()

Get the recommended path for a goal without making a completion call. Useful for inspecting what Kalibr would choose or for custom routing logic.

from kalibr import get_policy

policy = get_policy(goal="book_meeting")

print(policy["recommended_model"])       # "gpt-4o"
print(policy["recommended_tool"])        # "calendar_api"
print(policy["outcome_success_rate"])    # 0.87
print(policy["confidence"])              # 0.92

import { getPolicy } from '@kalibr/sdk';

const policy = await getPolicy({ goal: 'book_meeting' });

console.log(policy.recommendedModel);      // "gpt-4o"
console.log(policy.recommendedTool);       // "calendar_api"
console.log(policy.outcomeSuccessRate);    // 0.87
console.log(policy.confidence);            // 0.92

With Constraints

policy = get_policy(
    goal="book_meeting",
    constraints={
        "max_cost_usd": 0.05,
        "max_latency_ms": 2000,
        "min_quality": 0.8
    }
)

const policy = await getPolicy({
    goal: 'book_meeting',
    constraints: {
        maxCostUsd: 0.05,
        maxLatencyMs: 2000,
        minQuality: 0.8,
    },
});

Kalibr will only recommend paths that meet all constraints. If no paths meet the constraints, the response will indicate no recommendation is available.

Parameters

Returns

decide()

Get a routing decision for a goal. Returns the routing decision for a goal. This is what Router.completion() calls internally, but available for low-level control.

from kalibr import decide

decision = decide(goal="book_meeting", task_risk_level="low")

print(decision["model_id"])    # "gpt-4o"
print(decision["tool_id"])     # "calendar_api" or None
print(decision["params"])      # {"temperature": 0.3} or {}
print(decision["trace_id"])    # "abc123...", pass this to report_outcome
print(decision["confidence"])  # 0.85

Parameters

report_outcome()

Report execution outcome directly (without using Router). The feedback loop that teaches Kalibr what works.

from kalibr import report_outcome

report_outcome(
    trace_id="abc123",
    goal="book_meeting",
    success=True,
    score=0.95,                    # optional quality score 0-1
    failure_category="timeout",    # optional structured category
    model_id="gpt-4o",            # optional
)

Parameters

update_outcome()

Update an existing outcome with a late-arriving signal. Use when the real success signal arrives after the initial report, for example, updating 48 hours later when a customer reopens a ticket that was initially reported as resolved.

Only fields that are explicitly passed (not None) will be updated. Other fields retain their original values.

from kalibr import update_outcome

# Customer reopened ticket 48 hours after "resolution"
result = update_outcome(
    trace_id="abc123",
    goal="resolve_ticket",
    success=False,
    failure_category="user_unsatisfied",
)
print(result["fields_updated"])  # ["success", "failure_category"]

Parameters

Returns 404 if no outcome exists for the given trace_id + goal combination.

get_insights()

Get structured diagnostics about what Kalibr has learned. Returns machine-readable intelligence per goal, designed for coding agents that need to decide what to improve.

Response includes schema_version: "1.0" for forward compatibility.

from kalibr import get_insights

# All goals
insights = get_insights()

# Single goal, custom window
insights = get_insights(goal="resolve_ticket", window_hours=48)

for goal in insights["goals"]:
    print(f"{goal['goal']}: {goal['status']} ({goal['success_rate']:.0%})")
    for signal in goal["actionable_signals"]:
        if signal["severity"] == "critical":
            print(f"  {signal['type']}: {signal['data']}")

Parameters

Response structure (per goal)

Actionable signal types

register_path()

Register a new routing path for a goal.

from kalibr import register_path

result = register_path(
    goal="book_meeting",
    model_id="gpt-4o",
    tool_id="calendar_api",              # optional
    params={"temperature": 0.3},         # optional
    risk_level="low",                    # optional: "low", "medium", "high"
)
print(result["path_id"])

FAILURE_CATEGORIES

Constant containing all valid failure category values. Import and use for client-side validation.

from kalibr import FAILURE_CATEGORIES

print(FAILURE_CATEGORIES)
# ["timeout", "context_exceeded", "tool_error", "rate_limited",
#  "validation_failed", "hallucination_detected", "user_unsatisfied",
#  "empty_response", "malformed_output", "auth_error", "provider_error", "healed", "unknown"]

# Used by report() and report_outcome(), raises ValueError if invalid category passed

Intelligence API (TypeScript)

The TypeScript SDK exports convenience functions for direct access to the Intelligence API.

import {
  KalibrIntelligence,
  getPolicy,
  reportOutcome,
  registerPath,
  decide,
  getRecommendation,
  listPaths,
  disablePath,
  setExplorationConfig,
  getExplorationConfig,
} from '@kalibr/sdk';

// Initialize singleton
KalibrIntelligence.init({
  apiKey: process.env.KALIBR_API_KEY!,
  tenantId: process.env.KALIBR_TENANT_ID!,
});

// Get routing decision
const decision = await decide('extract_company');
console.log(decision.model_id, decision.confidence);

// Report outcome directly
await reportOutcome(traceId, 'extract_company', true, {
  score: 0.95,
  modelId: 'gpt-4o',
});

// List registered paths
const { paths } = await listPaths({ goal: 'extract_company' });

// List registered paths
const allPaths = await listPaths({ goal: 'extract_company' });

Auto-Instrumentation (TypeScript)

Wrap OpenAI or Anthropic clients to automatically trace all LLM calls.

import { createTracedOpenAI, createTracedAnthropic } from '@kalibr/sdk';

// Wrap OpenAI client, all calls auto-traced
const openai = createTracedOpenAI();
const response = await openai.chat.completions.create({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Hello!' }],
});

// Wrap Anthropic client
const anthropic = createTracedAnthropic();
const message = await anthropic.messages.create({
  model: 'claude-sonnet-4-20250514',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Hello!' }],
});

Context Management (TypeScript)

Run code within goal or trace contexts using async local storage.

import { withGoal, withTraceId, traceContext } from '@kalibr/sdk';

// Run code within a goal context
await withGoal('extract_company', async () => {
  // All Kalibr operations inherit this goal
  const response = await openai.chat.completions.create({...});
});

// Run code with a specific trace ID
await withTraceId('my-custom-trace-id', async () => {
  // All operations use this trace ID
});

// Combined trace context
await traceContext({ traceId: 'my-trace', goal: 'summarize' }, async () => {
  // Both trace ID and goal available
});

Environment Variables

REST API Endpoints

Intelligence service: https://kalibr-intelligence.fly.dev

All endpoints except GET /health require two headers: X-API-Key and X-Tenant-ID.

POST /api/v1/routing/decide

Get a routing decision for a goal. Selects the best registered path based on outcome history. Returns a trace_id that must be passed to report-outcome.

curl -X POST https://kalibr-intelligence.fly.dev/api/v1/routing/decide \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant" \
  -H "Content-Type: application/json" \
  -d '{"goal": "extract_company"}'

Request

Response

POST /api/v1/intelligence/report-outcome

Report execution outcome. This is the feedback loop that teaches Kalibr what works. Updates both ClickHouse (durable) and Redis (real-time).

curl -X POST https://kalibr-intelligence.fly.dev/api/v1/intelligence/report-outcome \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant" \
  -H "Content-Type: application/json" \
  -d '{"trace_id": "abc123", "goal": "extract_company", "success": true}'

Request

Response

POST /api/v1/intelligence/update-outcome

Update an existing outcome with a late-arriving signal. Only fields explicitly passed (not null) are updated. Use for async validation, user feedback, or downstream confirmation.

curl -X POST https://kalibr-intelligence.fly.dev/api/v1/intelligence/update-outcome \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant" \
  -H "Content-Type: application/json" \
  -d '{"trace_id": "abc123", "goal": "resolve_ticket", "success": false, "failure_category": "user_unsatisfied"}'

Request

Response

GET /api/v1/intelligence/insights

Get structured diagnostics about what Kalibr has learned. Returns health status, failure mode breakdown, path comparisons, and actionable signals per goal.

curl "https://kalibr-intelligence.fly.dev/api/v1/intelligence/insights?window_hours=168&goal=resolve_ticket" \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant"

Query params

Response shape

Each goals[] entry contains: goal, status ("healthy" / "degrading" / "failing" / "insufficient_data"), success_rate, sample_count, trend, trend_delta, confidence, top_failure_modes, paths, and actionable_signals.

POST /api/v1/intelligence/policy

Get the historically best-performing execution path for a goal. Unlike /decide, this is deterministic -- returns the historically best path with no sampling. Returns 404 if no execution data exists yet.

curl -X POST https://kalibr-intelligence.fly.dev/api/v1/intelligence/policy \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant" \
  -H "Content-Type: application/json" \
  -d '{"goal": "summarize_ticket"}'

Request

Response

POST /api/v1/intelligence/get-alternative

Get the next-best path after a primary model fails. Use for retry logic: call /decide, execute, fail, then call this with the failed models in exclude_models. Returns 404 if all registered paths are exhausted.

curl -X POST https://kalibr-intelligence.fly.dev/api/v1/intelligence/get-alternative \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant" \
  -H "Content-Type: application/json" \
  -d '{"goal": "extract_company", "exclude_models": ["gpt-4o"]}'

Request

Response

POST /api/v1/routing/paths

Register a new execution path for a goal. Kalibr only routes to explicitly registered paths.

curl -X POST https://kalibr-intelligence.fly.dev/api/v1/routing/paths \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant" \
  -H "Content-Type: application/json" \
  -d '{"goal": "extract_company", "model_id": "gpt-4o"}'

Request

GET /api/v1/routing/paths

List registered paths for a goal.

curl "https://kalibr-intelligence.fly.dev/api/v1/routing/paths?goal=extract_company" \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant"

DELETE /api/v1/routing/paths/{path_id}

Disable a path. Soft-deleted -- marked disabled, traffic stops, outcome history preserved.

curl -X DELETE https://kalibr-intelligence.fly.dev/api/v1/routing/paths/path_abc123 \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant"

GET /api/v1/routing/stats

Get goal-level routing statistics including success rates, sample counts, and path performance.

curl "https://kalibr-intelligence.fly.dev/api/v1/routing/stats?goal=extract_company" \
  -H "X-API-Key: your-key" \
  -H "X-Tenant-ID: your-tenant"

GET /api/v1/intelligence/health

Check service health. Does not require authentication.

curl https://kalibr-intelligence.fly.dev/api/v1/intelligence/health

Returns {"status": "healthy", "clickhouse": "connected", "redis": "connected", "last_aggregation": "..."}.

Default Values

Next

• Production Guide. Failure modes, monitoring