velenai 0.8.1

Agent economics SDK — track ROI, KPIs, and cost for AI agents in 2 lines of code

VelenAI SDK

Agent economics in 2 lines of code. Track ROI, KPIs, and cost for your AI agents.

Langfuse is the debugger. VelenAI is the CFO.

Install

pip install velenai

With provider wrappers and framework adapters:

pip install velenai[anthropic]  # Anthropic auto-capture
pip install velenai[openai]     # OpenAI auto-capture
pip install velenai[crewai]     # CrewAI adapter
pip install velenai[langgraph]  # LangGraph adapter
pip install velenai[autogen]    # AutoGen adapter

Quick Start — Agent Capture

Three lines to instrument your entire application:

import velenai

velenai.install(api_key="vai_...", api_secret="vas_...")

@velenai.agent(id="research", description="Research agent")
def run_research(query: str):
    client = OpenAI()  # automatically captured
    return client.chat.completions.create(model="gpt-4o", messages=[...])

install() patches provider classes process-wide. Every OpenAI() or Anthropic() client created after install() is automatically tracked — no wrapping needed.

Async works too

@velenai.agent(id="writer", description="Content writer", tier="production")
async def write_content(brief: str):
    client = AsyncOpenAI()
    return await client.chat.completions.create(model="gpt-4o", messages=[...])

Imperative registration

For agents created from config or at runtime:

import velenai

for agent_def in load_agents_from_yaml():
    velenai.register_agent(
        id=agent_def["name"],
        description=agent_def["role"],
        division=agent_def["division"],
        tier="production",
        owner="[e-mail]",
        tags={"llm_server": agent_def["llm_server"]},
    )

Context manager

For code that can't use decorators (callbacks, event handlers):

from velenai import agent_context

def on_message(event):
    with agent_context("chat-handler"):
        # LLM calls inside this block are attributed to "chat-handler"
        response = client.chat.completions.create(...)

agent_context() sets identity but does not register the agent. Call register_agent() first if you need metadata on the backend.

Identity resolution

When an LLM call happens, the SDK resolves agent identity in this order:

1. Explicit agent_id on wrap_openai(client, vai, agent_id="...") — highest priority
2. ContextVar from @agent() decorator or agent_context()
3. default_agent_id from install(default_agent_id="fallback")
4. Module inference — derives an ID from the calling module's __name__

Undecorated calls still get tracked; they just use the fallback identity.

Coexistence with explicit wrap

If you have existing wrap_openai() calls, they continue working alongside install(). The explicit agent_id on a wrapped client always wins over the ContextVar:

import velenai
from velenai.providers import wrap_openai

velenai.install(api_key="vai_...", api_secret="vas_...")

# This client uses agent_id from the decorator/context
auto_client = OpenAI()

# This client always reports as "legacy-agent" regardless of context
legacy_client = wrap_openai(OpenAI(), vai, agent_id="legacy-agent")

Healthcheck with status()

import velenai

info = velenai.status()
# {
#   "installed": True,
#   "enabled": True,
#   "patched_providers": ["openai", "anthropic"],
#   "registered_agents": ["research", "writer"],
#   "pending_telemetry_count": 3,
#   "last_flush_at": "2026-04-09T12:00:00Z",
#   ...
# }

Debugging with is_wrapped()

from velenai import is_wrapped

client = OpenAI()
assert is_wrapped(client)  # True after install()

ThreadPoolExecutor (Python < 3.12)

On Python < 3.12, ThreadPoolExecutor.submit() does not propagate contextvars. Use the helper:

from concurrent.futures import ThreadPoolExecutor
from velenai import submit_with_context

executor = ThreadPoolExecutor(max_workers=4)

@velenai.agent(id="parallel-worker")
def do_work(item):
    return client.chat.completions.create(...)

# Propagates agent context to the worker thread
future = submit_with_context(executor, do_work, item)

Known limitation: module-level clients

install() patches provider classes. Clients constructed before install() runs are not patched:

from openai import OpenAI

client = OpenAI()  # created at import time — NOT captured

import velenai
velenai.install(...)  # too late for the client above

Workaround: Call install() before any provider imports, or wrap pre-existing clients explicitly with wrap_openai(client, vai).

Environment variables

Instead of passing credentials to install():

export VELENAI_API_KEY=vai_...
export VELENAI_API_SECRET=vas_...
export VELENAI_HOST=http://localhost:8000

import velenai
velenai.install()  # picks up from env

Other env vars: VELENAI_DISABLED=true, VELENAI_DEFAULT_AGENT_ID=fallback, VELENAI_DETECT_PREEXISTING=false.

Advanced — Explicit Wrap API

For fine-grained control, wrap individual clients manually:

from velenai import VelenAI
from velenai.providers import wrap_openai

vai = VelenAI(api_key="your-key", api_secret="your-secret", host="http://localhost:8000")
client = wrap_openai(OpenAI(), vai, agent_id="my-agent")
response = client.chat.completions.create(model="gpt-4o", ...)

Decorator (explicit)

@vai.track("my-agent")
def run_agent(query: str):
    result = llm.chat(query)
    return result

Context Manager (explicit)

with vai.trace("my-agent") as t:
    result = do_work()
    t.success = True
    t.tokens = 150
    t.cost_usd = 0.003
    t.metadata["model"] = "gpt-4"

Manual emit

vai.emit(
    agent_id="my-agent",
    success=True,
    latency_ms=1200,
    tokens=150,
    cost_usd=0.003,
    metadata={"model": "gpt-4"},
)

Provider wrappers

from anthropic import Anthropic
from velenai.providers import wrap_anthropic

client = wrap_anthropic(Anthropic(), vai=vai)
response = client.messages.create(model="claude-sonnet-4-20250514", ...)

from openai import OpenAI
from velenai.providers import wrap_openai

client = wrap_openai(OpenAI(), vai=vai)
response = client.chat.completions.create(model="gpt-4o", ...)

Framework Adapters

CrewAI

from velenai_sdk import VelenAI
from velenai_sdk.adapters.crewai import VelenAICrewCallback

vai = VelenAI()
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    callbacks=[VelenAICrewCallback(vai)]
)

LangGraph

from velenai_sdk import VelenAI
from velenai_sdk.adapters.langgraph import instrument_graph

vai = VelenAI()
graph = StateGraph(...).compile()
graph = instrument_graph(vai, graph, agent_id="my-workflow")
result = graph.invoke(input)

Or as a LangChain callback:

from velenai_sdk.adapters.langgraph import VelenAILangGraphCallback

result = graph.invoke(input, config={"callbacks": [VelenAILangGraphCallback(vai)]})

AutoGen

from velenai_sdk import VelenAI
from velenai_sdk.adapters.autogen import instrument_agent

vai = VelenAI()
assistant = AssistantAgent("analyst", llm_config=llm_config)
assistant = instrument_agent(vai, assistant)

Features

• Sync and async support
• Background batching (events queued and flushed every 5s)
• HMAC-signed requests (replay-protected)
• Fire-and-forget (never blocks your agent)
• Agent Capture: process-wide auto-instrumentation with install()
• Framework adapters: CrewAI, LangGraph, AutoGen
• Zero dependencies beyond httpx
• Self-hosted

License

MIT