viktron-sdk 0.2.2

Production-grade observability and governance SDK for AI agents

Viktron SDK

Framework-agnostic governance SDK for AI agents — policy enforcement, telemetry, and real-time observability for production AI systems.

Install

pip install viktron-sdk
# With optional provider extras:
pip install "viktron-sdk[openai]"
pip install "viktron-sdk[anthropic]"
pip install "viktron-sdk[langchain]"
pip install "viktron-sdk[all]"

Quick Start

One-line decorator instrumentation

import viktron_sdk

# Initialise once at startup
viktron_sdk.init(
    api_key="vk_live_...",
    agent_id="my-research-agent",
    display_name="Research Agent v2",
    framework="crewai",
    llm_provider="anthropic",
    llm_model="claude-3-sonnet",
)

# Works on both sync and async functions
@viktron_sdk.observe(as_type="tool", skip_args=["api_key"])
async def search_web(query: str, api_key: str):
    ...

@viktron_sdk.observe(as_type="llm")
def call_model(prompt: str):
    ...

# Flush before exit (atexit handler covers normal exits automatically)
viktron_sdk.force_flush()
# Or fully shut down the SDK:
viktron_sdk.stop_tracing()

observe shorthand types: "llm" → llm_call, "tool" → tool_call, "guardrail" → guardrail_check, "agent" → agent_response.

CLI

# Check configuration and API connectivity
viktron doctor

# Refresh cached pricing data
viktron pricing

Telemetry (send agent events to your Viktron dashboard)

from viktron_sdk import ViktronTelemetry

tel = ViktronTelemetry(api_key="vk_live_...", agent_slug="my-agent")

# Record a task
tel.record_task("task-123", status="completed", duration_ms=4200, cost_usd=0.003)

# Use a context-managed span
with tel.span("run_campaign") as span:
    span.set_attribute("platform", "meta")
    result = run_campaign()
    span.set_output(result)

tel.close()  # flush remaining events

Policy Guard (intercept LLM calls with governance rules)

import openai
from viktron_sdk import ViktronGuard

client = ViktronGuard.wrap(
    openai.OpenAI(),
    api_key="vk_live_...",
    agent_id="sales-agent-prod",
)

# All create() calls are now policy-checked
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Works with OpenAI (sync & async), Anthropic, Cohere, and any client with .create() / .generate() call patterns.

Framework Integrations

# LangChain
from viktron_sdk.integrations.langchain import ViktronCallbackHandler
handler = ViktronCallbackHandler(api_key="vk_live_...", agent_id="my-chain")

# CrewAI
from viktron_sdk.integrations.crewai import ViktronCrewAIObserver
observer = ViktronCrewAIObserver(api_key="vk_live_...", agent_id="my-crew")

# AutoGen
from viktron_sdk.integrations.autogen import ViktronAutoGenHook
hook = ViktronAutoGenHook(api_key="vk_live_...", agent_id="my-autogen")

API Keys

Generate your API key at app.viktron.ai → Settings → API Keys.

Keys prefixed vk_live_ are production; vk_test_ are for testing.

Links

• Dashboard
• Documentation
• GitHub