pylva (Python)

Cost infrastructure for AI agent businesses. Auto-instruments openai and anthropic clients; emits server-priced telemetry to your Pylva backend.

Version 1.0.4 adds flexible provider and model identifiers. Public APIs follow SemVer; the telemetry wire format remains schema 1.6.

pip install pylva-sdk

import pylva

pylva.init(api_key="pv_live_...", endpoint="https://api.pylva.com")
# every subsequent openai / anthropic call emits a telemetry event.

Package name

The PyPI distribution package is pylva-sdk; the runtime import package is pylva:

pip install pylva-sdk

import pylva

Failover (reliability_failover rules)

If you use reliability_failover rules, switch to the explicit-client constructor so Pylva has a handle to the backup provider:

from openai import OpenAI
from anthropic import Anthropic
from pylva import Pylva

Pylva(
    api_key="pv_live_...",
    openai=OpenAI(),
    anthropic=Anthropic(),
    providers={"openrouter": openrouter_client},
)

Use providers={...} for any additional provider id exactly as it appears in your rules or telemetry.

pylva.init(api_key) keeps working for telemetry-only deployments. In v1.0, the SDK detects active cross-provider failover states and surfaces warnings; actual backup-provider dispatch is still beta/internal and is not part of the stable launch promise.

Auto-instrumentation

Importing pylva monkey-patches any openai / anthropic modules already loaded. Patches are isolated per R1 — SDK errors never propagate to your agent.

Calls are captured with: model, provider, tokens_in, tokens_out, latency_ms, status, step_name (if set), customer_id. Cost is computed server-side.

Tracking context

from pylva import track_context

with track_context(customer_id="acme-corp", step="authentication"):
    resp = openai_client.chat.completions.create(...)

track_context is a contextvars-backed context manager; it threads metadata across async + threaded boundaries correctly.

LangGraph / LangChain callback

For LangGraph apps, use the callback handler when you want cost attribution by graph run, node, and customer:

pip install "pylva-sdk[langchain]"

import os
from pylva.langchain import PylvaCallbackHandler

handler = PylvaCallbackHandler(api_key=os.environ["PYLVA_API_KEY"])

graph.invoke(
    {"question": "Where did spend increase?"},
    config={
        "callbacks": [handler],
        "metadata": {"pylva_customer_id": "cust_acme"},
    },
)

The handler reads LangChain usage metadata, preserves LangGraph run ids, uses langgraph_node as the default step name, and never captures prompt, completion, tool input, or tool output text. Metadata step labels from langgraph_node, langgraph_step, and pylva_step must be identifier-like ([A-Za-z0-9_.:/-], max 100 chars). Free-text prompts or messages should never be placed in metadata; unsafe labels are ignored.

Non-LLM costs

from pylva import report_usage

report_usage(
    customer_id="acme-corp",
    tool="translation",
    metric="characters",
    value=4200,
    step="translation",
)

Reactive budget enforcement (B2a)

When a builder configures a budget_limit rule with hard_stop=True, the SDK enforces pre-call:

import pylva
from pylva import PylvaBudgetExceeded, BudgetExceededSource

pylva.init(api_key="pv_live_...")

try:
    resp = openai_client.chat.completions.create(...)
except PylvaBudgetExceeded as err:
    print(f"Budget hit for {err.customer_id}: ${err.accumulated_usd:.2f} / ${err.limit_usd:.2f}")
    print(f"Source: {err.source.value}")  # 'sdk_precall' or 'backend_ingest_flag'
    # graceful degradation

How it works

• Pre-call accumulator (per-process dict + threading.Lock). Keyed on (rule_id, scope_token, period_start). Local enforcement is driven by backend budget flags and /budget/sync reconciliation. If the local view is already over a hard-stop limit, the SDK raises PylvaBudgetExceeded(source=BudgetExceededSource.SDK_PRECALL).

• Backend-authoritative flag. Every ingest response may carry budget_exceeded[]. The SDK bumps local accumulators to limit + 1 on receipt; next pre-call raises source=BACKEND_INGEST_FLAG.

• 5-min sync loop. A threading.Timer re-POSTs accumulator state to /api/v1/budget/sync and overwrites local with the server truth (not additive).

• Passthrough. Cold rules cache or unreachable backend → pre-call is a no-op. Your agent never blocks due to Pylva being degraded (R5).

Rule behavior matrix

Webhook verification

import os
from pylva import verify_webhook, InvalidSignatureFormat

def handle_webhook(body: str, signature: str, timestamp: str):
    try:
        ok = verify_webhook(body, signature, os.environ["WEBHOOK_SECRET"], timestamp)
    except InvalidSignatureFormat:
        return 400
    if not ok:
        return 400
    # ...

verify_webhook accepts both raw hex and the sha256=<hex> GitHub-style prefix (B2a D34 parity with TS SDK). Default timestamp tolerance is 300 s.

Privacy & PII

Pylva does not redact step_name, customer_id, or metadata values. Do not pass raw user message content, email addresses, or phone numbers into these fields. LangGraph metadata step labels are accepted only when they are identifier-like; other validation rejects HTML and most control characters. This does not protect against free-form PII.

SDKs never auto-redact by design: regex redaction is noisy and incomplete, and classifier redaction would add an LLM round-trip to the hot path. PII handling is the builder's responsibility; we provide exportable data and deletion endpoints.

License

MIT.