Python and Node SDKs.

Overview

The SDKs are thin clients of the Runtime API: before your driver runs a statement, they send it to POST /api/v1/proxy/query and apply the decision. A blocked statement never reaches the database, a flagged one runs and is reported, and one that reads masked columns runs as the rewrite Vericto returns. Evaluation happens in your workspace: the SDK evaluates nothing locally.

• Python: vericto (Python 3.10+, one dependency, urllib3). Wrappers for snowflake-connector-python, sqlite3, SQLAlchemy and any DB-API driver; integrations for LangChain, LlamaIndex, dbt and Airflow.
• Node.js: @vericto/sdk (Node.js 18+, no runtime dependencies, ESM and CommonJS, TypeScript types). Wrappers for snowflake-sdk and better-sqlite3, a LangChain.js integration, and a Runtime API client for any other driver.

For PostgreSQL and MySQL there is also the TCP proxy: it needs no code changes and protects all the SQL on the connection, including what an ORM generates. The SDK protects the SQL that goes through it (see Enforcement boundary).

Install

# Python 3.10+
pip install vericto                  # core, sqlite3 and any DB-API driver
pip install "vericto[sqlalchemy]"    # extras: snowflake, sqlalchemy, langchain, llamaindex, airflow, dbt

# Node.js 18+
npm install @vericto/sdk             # entry points: /snowflake, /better-sqlite3, /langchain

Drivers and frameworks are extras (Python) or optional peer dependencies (Node): importing vericto or @vericto/sdk never loads a driver.

Configuration

Create an API key with the proxy:evaluate scope in Dashboard → API keys (one per service, so its events identify it) and register the database you are protecting. Both SDKs read the same environment variables:

Constructor options take precedence over the variables: Vericto(api_key=..., database_id=..., fail_mode=..., timeout=...) in Python (the CA bundle goes in RuntimeApiEvaluator(ca_bundle=...)) and new Vericto({ apiKey, databaseId, failMode, timeoutMs }) in Node, where the timeout is in milliseconds. The key never appears in logs, error messages, repr() or toString().

The API

Share one Vericto instance per process: it is thread-safe and reuses its HTTP connections. The wrappers call enforce for every statement, and you can call it yourself around any other driver:

• check(sql, dialect): returns the verdict, whatever it is. It does not raise on BLOCKED and does not apply the fail mode: use it to ask.
• enforce(sql, dialect): evaluates and applies the decision. It raises the block error, logs FLAGGED and MONITORED, applies the fail mode when Vericto cannot answer, and returns the verdict.
• prepare(sql, dialect): enforce, returning the SQL you must run: the masked rewrite when there is one, otherwise your SQL.
• enforce_all (Python) and enforceAll (Node): a batch. A block in any statement raises before you run the first one.

import psycopg
import vericto
from vericto import Vericto, VerictoBlocked

guard = Vericto()  # one per process; reads the VERICTO_* variables
conn = vericto.wrap(psycopg.connect("dbname=app"), dialect="postgres", guard=guard)

try:
    conn.cursor().execute("DELETE FROM customers")  # no WHERE: blocked, never reaches the database
except VerictoBlocked as exc:
    print(exc.verdict.rule_code, exc.suggested_fix)

sql = guard.prepare("SELECT id, email FROM customers LIMIT 10", "postgres")  # the masked rewrite, if any

import { Vericto, VerictoBlockedError, sqlToExecute } from '@vericto/sdk';

const vericto = new Vericto(); // one per process; reads the VERICTO_* variables

try {
  const verdict = await vericto.enforce(sql, 'postgres');
  await pool.query(sqlToExecute(verdict)); // the masked rewrite, if any
} catch (err) {
  if (!(err instanceof VerictoBlockedError)) throw err;
  console.log(err.verdict.ruleCode, err.suggestedFix);
}

The verdict carries decision, the rule code, the severity, the suggested fix, the masked rewrite, the findings, the sensitive columns it reads, what the agent allowlist denied, and the Audit Trail event id (in Python as rule_code, suggested_fix, rewritten_query, event_id; in Node as ruleCode, suggestedFix, rewrittenQuery, eventId). Parameter values are never sent: the SQL goes as your application wrote it, placeholders included, and the driver binds them afterwards.

What each decision does

Log lines carry the rule code, the severity and the event_id, never the SQL text (its literals can hold personal data). Every evaluation is recorded in the Audit Trail and counts against the monthly query quota.

Fail closed, timeouts and retries

When Vericto cannot answer (network error, timeout, HTTP 5xx, the per-minute limit after retries, or the monthly quota used up), with closed, the default, the SDK raises VerictoUnavailable (Python) or VerictoUnavailableError (Node) and the statement does not run. With open, it runs it unevaluated, emits VerictoFailOpenWarning and passes an UNEVALUATED verdict to on_verdict / onVerdict.

Configuration errors are never let through, in either mode: a missing or revoked key (401), a key without the scope or without access to that database (403 without decision), an invalid request (400) or a dialect the server does not evaluate raise VerictoConfigError. Letting them through would switch the firewall off without anyone noticing.

Network errors, timeouts, 5xx and 429 RATE_LIMIT_EXCEEDED are retried twice by default (max_retries / maxRetries), with exponential backoff and jitter of up to 2 s that honours Retry-After. QUOTA_EXCEEDED is not retried. The timeout is per attempt: with the defaults, a statement can wait three 5 s attempts plus the backoff (and, on the first call, the request for the server's dialects) before it fails closed. Lower it on latency-sensitive paths.

Errors

All of them extend VerictoError.

Sync and async

Python is synchronous. Every evaluation is a blocking HTTP call, including the one vericto.sqlalchemy.instrument() makes on an AsyncEngine. In asyncio code (FastAPI, OpenAI Agents SDK, Pydantic AI), evaluate in a thread so the event loop is not blocked:

import asyncio

sql = await asyncio.to_thread(guard.prepare, query, "postgres")

Node is asynchronous. Every method returns a promise and there is no synchronous variant. That is why, with better-sqlite3, prepare and exec become async, and with snowflake-sdk, execute() resolves the statement.

Frameworks

An official integration is SDK code (a module or an entry point) that already does the evaluation. A pattern is a documented way to call the SDK from your own code with that framework: the SDK ships no adapter for it. Each guide has a minimal example and the versions it was tested with.

The pattern is the same in any agent framework: the tool evaluates the query before it runs it, runs the SQL prepare() returns and, when Vericto blocks it, returns the reason and the suggested fix as text so the model can correct the query.

import psycopg
from vericto import Vericto, VerictoBlocked

guard = Vericto()  # one per process
conn = psycopg.connect("dbname=app")  # a plain connection: the tool does the check

def run_sql(query: str) -> str:
    """The function you register as the agent's SQL tool."""
    try:
        sql = guard.prepare(query, "postgres")  # the masked rewrite when there is one
    except VerictoBlocked as exc:
        return f"Vericto blocked this query ({exc.verdict.rule_code}); it was not run. Suggested fix: {exc.suggested_fix}"
    with conn.cursor() as cur:  # VerictoUnavailable and VerictoConfigError propagate
        cur.execute(sql)
        return repr(cur.fetchall())

import pg from 'pg';
import { Vericto, VerictoBlockedError } from '@vericto/sdk';

const vericto = new Vericto(); // one per process
const pool = new pg.Pool(); // a plain pool: the tool does the check

// The function you register as the agent's SQL tool.
export async function runSql(query: string): Promise<string> {
  let sql: string;
  try {
    sql = await vericto.prepare(query, 'postgres'); // the masked rewrite when there is one
  } catch (err) {
    if (!(err instanceof VerictoBlockedError)) throw err; // unavailable or misconfigured: never hand it to the model
    return `Vericto blocked this query (${err.verdict.ruleCode}); it was not run. Suggested fix: ${err.suggestedFix}`;
  }
  const { rows } = await pool.query(sql);
  return JSON.stringify(rows);
}

Let VerictoUnavailable and VerictoConfigError propagate: turning them into text for the model would switch fail-closed off without anyone noticing. The official integrations already work this way. Some frameworks turn any tool exception into a message for the model (OpenAI Agents SDK, Vercel AI SDK, Vanna): each guide shows how to avoid it.

Enforcement boundary

The SDK protects the SQL that goes through it. It does not evaluate SQL run on another connection, through the driver's original object (cursor.wrapped, db.raw, rawDataSource) or from another process with the same credentials, nor queries an ORM generates without exposing their SQL before they run. Pair the SDK with database-side least privilege and, on PostgreSQL and MySQL, with the TCP proxy when the guarantee matters.

See also

• Runtime API
• Set up the SDK with AI
• SQLite
• Sensitive columns
• Agent access
• AI agents (MCP)