vantio-agent-sdk · Vantio Optics (Python)

Vantio Optics open-core SDK for Sight Loop observe — shield() telemetry for Python agents. Metadata only; no prompts. Current PyPI release: 3.0.6.

pip install vantio-agent-sdk

Optics: vantio.ai/optics

vantio run python agent.py (CLI 0.3.4+) loads this wrap for the whole interpreter. You can also use shield() when you want a trace id inside the process. Optics records urllib to in-scope LLM hosts. If requests, httpx, or aiohttp are already installed, those are recorded the same way. socket.connect / create_connection to in-scope hosts share host-block and observe. subprocess / os.system / asyncio curl spawns to in-scope hosts share host-block and observe (curl bodies are not rewritten). With a Gate key, the same wrap can block a destination, redact PII, or enforce a spend limit on HTTP bodies. Browsers stay outside this wrap.

Providers: OpenAI (including regional), Anthropic, Google Gemini, Azure OpenAI, Azure AI, Cohere, Mistral, Groq, Together AI, Perplexity, xAI, DeepSeek, Fireworks, OpenRouter, Cerebras, Voyage AI, SambaNova, DeepInfra, Amazon Bedrock, Google Vertex AI, Hugging Face Inference, Replicate, Ollama, hosted NVIDIA NIM.

v3.0.x — Breaking change from v2.x

v3.0.0+ is a complete rewrite. The old VantioSession / VANTIO_PROXY_ENDPOINT API is removed. The new API uses @shield (a decorator or async context manager) with zero dependencies and no proxy — observe runs inside your SDK, not through an external endpoint.

Migrate from v2.x:

# Old (v2.x) — remove this
from vantio.session import VantioSession
with VantioSession(agent_name="my-agent") as session: ...

# New (v3.x) — use this instead
from vantio import shield
@shield
async def run_agent(): ...
# or: async with shield() as ctx: ...

shield() — trace context

from vantio import shield, report_anomaly

# Decorator form
@shield
async def run_agent():
    result = await call_openai(prompt)
    return result

# Context manager form
async with shield() as ctx:
    print(f"Trace ID: {ctx.trace_id}")
    result = await run_agent()

get_current_trace_id()

Returns the active trace ID for the current async context, or None outside a shield() frame.

from vantio import shield, get_current_trace_id

async with shield() as ctx:
    trace_id = get_current_trace_id()  # same as ctx.trace_id

get_current_trace_id()  # None — outside shield() frame

fetch_policy() — policy retrieval

Fetches your tenant's governance policy from GET /api/v1/config. Fails open — if the control plane is unreachable, a permissive default policy is returned so your agent never stalls waiting for governance.

from vantio import fetch_policy, redact_pii
import os

policy = fetch_policy(os.environ["VANTIO_API_KEY"])

# Check enforcement flags
if policy.enforce:
    if "api.openai.com" in policy.blocked_hosts:
        raise RuntimeError("OpenAI is blocked by policy")

# Use with redact_pii for SDK-side PII scrubbing
if policy.pii_redact:
    result = redact_pii(user_input, policy.pii_types)
    prompt = result.text   # PII scrubbed before the LLM call

VantioPolicy fields:

redact_pii() — local PII scrubbing

Pure, side-effect-free local PII redaction using the same patterns as the CLI interceptor and @vantio/agent-sdk. No content leaves the process.

from vantio import redact_pii

result = redact_pii("Contact bob@example.com or call 555-123-4567")
# result.text       → "Contact [VANTIO_REDACTED:EMAIL] or call [VANTIO_REDACTED:PHONE]"
# result.redactions → ["email", "phone"]

Pass a custom pii_types list to control which categories are scanned (values are case-insensitive):

result = redact_pii(text, pii_types=["email", "ssn"])

Built-in categories: ssn, email, credit_card, phone.

RedactionResult fields:

report_anomaly() — cloud ingest

async with shield():
    await run_agent()
    await report_anomaly(
        target_host="api.openai.com",
        bytes_severed=14382,
        # Valid values: "OBSERVED" | "ALLOWED" | "REDACTED" | "BLOCKED_HOST" | "BLOCKED_SIZE" | "BLOCKED_SPEND"
        action_taken="BLOCKED_HOST",
    )

report_anomaly() must be called within a shield() context. It is a no-op unless VANTIO_CLOUD_INGEST=true. Non-fatal — never crashes the agent.

Environment variables

Anonymous telemetry (opt-out)

The SDK sends a single anonymous usage ping per process the first time shield() runs. It contains only aggregate, non-identifying metadata — a random anonymous id, the Python version, the OS string, and an event name. It never sends prompts, completions, API keys, emails, or any content/PII. Fire-and-forget on a daemon thread; never blocks.

export VANTIO_TELEMETRY_DISABLED=1   # or
export DO_NOT_TRACK=1

Zero required dependencies

Core tracing requires only the Python standard library (contextvars, asyncio, hashlib, hmac, re). Cloud ingest and anonymous telemetry use urllib.request and threading. This package does not depend on aiohttp, requests, or httpx.

While the wrap is active (vantio run python or shield()), Optics records urllib calls to in-scope LLM hosts (OpenAI, Anthropic, Bedrock, Vertex AI, Hugging Face Inference, Replicate, Ollama, NVIDIA NIM, and the rest of the catalog). If requests, httpx, or aiohttp are already installed in the agent environment, those are recorded the same way. They are optional; installing this package does not pull them in. socket.connect / create_connection to in-scope hosts share host-block and observe. subprocess curl to in-scope hosts shares host-block and observe; curl bodies are not rewritten. Browsers stay outside this wrap.