Skip to content
SDK Documentation

Utilities API Reference

Utilities API Reference

Reference for utility functions, helper classes, and convenience methods available in the Klira SDK.

Tracing Context API

The tracing context API provides functions for managing user and conversation context within Klira traces. Import from klira.sdk.tracing:

from klira.sdk.tracing import (
    UserMessageTraceContext,
    set_klira_context,
    get_klira_user_id,
    get_klira_conversation_id,
    get_klira_message_id,
    is_in_klira_trace,
    propagate_context,
)

UserMessageTraceContext

Context manager for creating a root klira.user.message trace span. Supports both sync and async usage.

# Sync
with UserMessageTraceContext(user_id="user_123", conversation_id="conv_456") as ctx:
    result = process_message("Hello")


# Async
async with UserMessageTraceContext(user_id="user_123", conversation_id="conv_456") as ctx:
    result = await process_message("Hello")

Parameters

ParameterTypeRequiredDescription
user_idstrNoUser identifier
conversation_idstrNoConversation identifier
message_idstrNoMessage identifier

set_klira_context()

Set user, conversation, and message context on the current span.

def set_klira_context(
    user_id: Optional[str] = None,
    conversation_id: Optional[str] = None,
    message_id: Optional[str] = None,
) -> None

Use this when you need to set context inside an existing span (e.g., within a @workflow function) rather than creating a new root span.

from klira.sdk.tracing import set_klira_context


@workflow(name="chat")
async def chat(user_id: str, message: str) -> str:
    set_klira_context(user_id=user_id, conversation_id=f"conv_{user_id}")
    return await process_message(message)

Context Getters

from klira.sdk.tracing import get_klira_user_id, get_klira_conversation_id, get_klira_message_id


user_id = get_klira_user_id()          # Returns Optional[str]
conv_id = get_klira_conversation_id()  # Returns Optional[str]
msg_id = get_klira_message_id()        # Returns Optional[str]

is_in_klira_trace()

Check if the current execution is within a Klira trace.

from klira.sdk.tracing import is_in_klira_trace


if is_in_klira_trace():
    print("Running inside a Klira-traced context")

propagate_context()

Propagate Klira trace context to a new thread or task.

from klira.sdk.tracing import propagate_context
import threading


ctx = propagate_context()


def worker():
    with ctx:
        do_work()  # Klira context is available here


thread = threading.Thread(target=worker)
thread.start()

Framework Detection Utilities

detect_framework()

Automatically detect the LLM framework being used in a function.

from klira.sdk.utils.framework_detection import detect_framework


def detect_framework(func: Callable) -> Optional[str]

Supported Frameworks

  • "agents_sdk" — OpenAI Agents SDK
  • "langchain" — LangChain
  • "crewai" — CrewAI
  • "llama_index" — LlamaIndex
  • "custom" — Custom framework patterns

Example

from klira.sdk.utils.framework_detection import detect_framework


def create_openai_agent():
    return Agent(name="Assistant", instructions="Be helpful")


framework = detect_framework(create_openai_agent)
print(framework)  # Output: "agents_sdk"

is_framework_available()

Check if a specific framework is available in the current environment.

from klira.sdk.utils.framework_detection import is_framework_available


if is_framework_available("langchain"):
    from langchain.agents import create_openai_tools_agent

Registry Utilities

FrameworkRegistry

Centralized registry for framework adapters.

from klira.sdk.utils.framework_registry import FrameworkRegistry


FrameworkRegistry.register("my_framework", MyCustomAdapter)
available = FrameworkRegistry.list_available()

LLMClientRegistry

Registry for LLM client adapters.

from klira.sdk.utils.framework_registry import LLMClientRegistry


llm_adapters = LLMClientRegistry.list_available()

Performance Utilities

@performance_monitor

Decorator to monitor function performance.

from klira.sdk.utils.performance import performance_monitor


@performance_monitor(log_threshold_ms=100)
def expensive_operation(data: str) -> str:
    return process_large_dataset(data)
ParameterTypeDefaultDescription
log_threshold_msint50Log execution time if above this threshold
include_argsboolFalseInclude function arguments in logs
metric_nameOptional[str]NoneCustom metric name for telemetry

Timer

Context manager for timing operations.

from klira.sdk.utils.performance import Timer


with Timer("policy_evaluation") as timer:
    result = await guardrails.process_message(message)


print(f"Policy evaluation took {timer.elapsed_ms:.2f}ms")

Error Handling Utilities

@handle_errors

Decorator for standardized error handling.

from klira.sdk.utils.error_handling import handle_errors


@handle_errors(fail_closed=False, default_return_on_error={"allowed": True})
async def optional_guardrails_check(message: str) -> dict:
    return await guardrails.process_message(message)
ParameterTypeDefaultDescription
fail_closedboolFalseRe-raise exceptions instead of returning default
default_return_on_errorAnyNoneDefault value to return on error
log_errorsboolTrueWhether to log errors
reraise_onOptional[List[Type[Exception]]]NoneException types to always re-raise

KliraErrorReporter

Centralized error reporting and logging.

from klira.sdk.utils.error_handling import KliraErrorReporter


try:
    result = risky_operation()
except Exception as e:
    KliraErrorReporter.report_error(
        error=e,
        context={"operation": "policy_evaluation"},
        severity="high"
    )

Validation Utilities

validate_api_key()

from klira.sdk.utils.validation import validate_api_key


if validate_api_key("klira_live_abc123"):
    print("Valid API key format")

validate_url()

from klira.sdk.utils.validation import validate_url


if validate_url("https://api.getklira.com"):
    print("Valid URL")

Logging Utilities

get_logger()

from klira.sdk.utils.logging import get_logger


logger = get_logger(__name__)
logger.info("Starting operation")

Debugging Utilities

collect_diagnostics()

from klira.sdk.utils.debug import collect_diagnostics


diagnostics = collect_diagnostics()
print(f"SDK version: {diagnostics['version']}")
print(f"Active adapters: {diagnostics['adapters']}")

Context Management Best Practice

Use decorator params for context when possible, set_klira_context() for complex cases:

from klira.sdk.decorators import workflow
from klira.sdk.tracing import set_klira_context


# Simple — use decorator params
@workflow(name="chat", user_id="user_123", conversation_id="conv_456")
async def simple_chat(message: str) -> str:
    return await process_message(message)


# Complex — dynamic context
@workflow(name="chat")
async def dynamic_chat(user_id: str, message: str) -> str:
    set_klira_context(user_id=user_id, conversation_id=f"conv_{user_id}")
    return await process_message(message)