@obsvr/sdk Reference

Installation

The SDK is not published to npm yet (coming soon). When it ships it will install from npm as below. The package ships as ESM with full TypeScript types and requires Node 18+.

npm install @obsvr/sdk

Quick Start

Initialize obsvr once at startup, then wrap your LLM client. Every call through the wrapped client is automatically intercepted, policy-checked, and signed into the audit trail.

import OpenAI from 'openai'
import { obsvr } from '@obsvr/sdk'

obsvr.init({
  apiKey: process.env.OBSVR_API_KEY,
  environment: 'production',
})

const client = obsvr.wrap(new OpenAI())

const res = await client.chat.completions.create({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Hello' }],
})

Zero-Code Setup

Prefer not to wrap each client? Govern every client in the process automatically.

• Start Node with the module interceptor, or add import obsvr.register before your provider imports in Python.
• Every OpenAI and Anthropic client is then governed, including those created inside third-party libraries.
• No monkey-patching: prototypes are never modified, so APM and tracing on the same SDKs keep working.
• Use the providers option to narrow coverage.

node --import @obsvr/sdk/register app.js

# or via NODE_OPTIONS (Docker, PaaS, npm scripts)
NODE_OPTIONS="--import @obsvr/sdk/register" npm start

Authentication

Obtain your API key from the obsvr dashboard. The SDK reads from the OBSVR_API_KEY environment variable by default, or you can pass it explicitly to init(). For direct HTTP ingestion, include it as an X-API-Key header.

# .env
OBSVR_API_KEY=sk_live_...

# SDK auto-reads from env
obsvr.init({ environment: 'production' })

obsvr.init(config)

Initialize the SDK once, before any wrap() calls; it configures the ingest endpoint, policy engine, and PII scanner. For zero-code global coverage, start Node with node --import @obsvr/sdk/register app.js (prototypes are never patched, so APM and tracing keep working).

obsvr.wrap(client, options?)

Returns a proxy of the given LLM client that intercepts API calls. The original client is never modified - a new ES Proxy is returned. Supports OpenAI, Anthropic, and Google clients. Double-wrap is automatically prevented.

const client = obsvr.wrap(new OpenAI())

obsvr.flush(timeoutMs?)

Flushes the internal event queue, ensuring all pending audit events are sent to the ingest service. Essential for serverless environments (Lambda, Vercel, Cloudflare Workers) where the process may exit before background sends complete.

export async function handler(event) {
  const client = obsvr.wrap(new OpenAI())
  const res = await client.chat.completions.create({ ... })

  await obsvr.flush(5000) // wait up to 5s
  return { statusCode: 200, body: res.choices[0].message.content }
}

Utility Methods

Helper methods for inspecting SDK state at runtime.

defineConfig()

Type-safe config helper for use in obsvr.config.ts files. Returns the config object unchanged - it exists solely for IDE autocompletion and type checking.

import { defineConfig } from '@obsvr/sdk'

export default defineConfig({
  apiKey: process.env.OBSVR_API_KEY!,
  ingestUrl: 'https://ingest.obsvr.co',
  environment: 'production',
  providers: ['openai', 'anthropic'],
  piiPolicy: {
    default: 'redact',
    rules: { ssn: 'block', email: 'redact' },
  },
})

OpenAI

Wrap the official OpenAI client. Intercepts chat.completions.create (including streaming) and embeddings.create.

import OpenAI from 'openai'
import { obsvr } from '@obsvr/sdk'

obsvr.init({ environment: 'production' })
const client = obsvr.wrap(new OpenAI())

const res = await client.chat.completions.create({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Hello' }],
})

Anthropic

Wrap the Anthropic client. Intercepts messages.create and messages.stream.

import Anthropic from '@anthropic-ai/sdk'
import { obsvr } from '@obsvr/sdk'

obsvr.init({ environment: 'production' })
const client = obsvr.wrap(new Anthropic())

const res = await client.messages.create({
  model: 'claude-sonnet-4-20250514',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Hello' }],
})

AWS Bedrock

Use wrapBedrock to intercept AWS Bedrock InvokeModelCommand and Converse calls. Import from the dedicated subpath.

import { BedrockRuntimeClient } from '@aws-sdk/client-bedrock-runtime'
import { wrapBedrock } from '@obsvr/sdk/bedrock'

const client = wrapBedrock(new BedrockRuntimeClient({
  region: 'us-east-1',
}))

// Supports: InvokeModelCommand, ConverseCommand, ConverseStreamCommand

Azure OpenAI

Use wrapAzureOpenAI for Azure-hosted OpenAI endpoints. Same interface as the OpenAI wrapper but tags events with the azure-openai provider.

import { AzureOpenAI } from 'openai'
import { wrapAzureOpenAI } from '@obsvr/sdk/azure-openai'

const client = wrapAzureOpenAI(new AzureOpenAI({
  endpoint: 'https://my-resource.openai.azure.com',
  apiVersion: '2024-06-01',
}))

Google Vertex AI

Use wrapVertexAI for Google’s Vertex AI Gemini models.

import { VertexAI } from '@google-cloud/vertexai'
import { wrapVertexAI } from '@obsvr/sdk/vertex'

const client = wrapVertexAI(new VertexAI({
  project: 'my-project',
  location: 'us-central1',
}))

Cloudflare Workers AI

Use wrapWorkersAI for Cloudflare Workers AI. Pass ctx.waitUntil so audit events flush before the worker terminates.

import { wrapWorkersAI } from '@obsvr/sdk/cloudflare'

export default {
  async fetch(req, env, ctx) {
    const ai = wrapWorkersAI(env.AI, {
      waitUntil: (p) => ctx.waitUntil(p),
    })

    const res = await ai.run('@cf/meta/llama-3-8b-instruct', {
      messages: [{ role: 'user', content: 'Hello' }],
    })
    return Response.json(res)
  },
}

LangChain JS

Use the ObsvrCallbackHandler as a LangChain callback to audit all LLM calls within a chain or agent.

import { ChatOpenAI } from '@langchain/openai'
import { ObsvrCallbackHandler } from '@obsvr/sdk/langchain'

const handler = new ObsvrCallbackHandler()

const llm = new ChatOpenAI({
  model: 'gpt-4o',
  callbacks: [handler],
})

Vercel AI SDK

Use obsvrMiddleware with the Vercel AI SDK’s wrapLanguageModel to audit all generate/stream calls.

import { openai } from '@ai-sdk/openai'
import { wrapLanguageModel, generateText } from 'ai'
import { obsvrMiddleware } from '@obsvr/sdk/vercel-ai'

const model = wrapLanguageModel({
  model: openai('gpt-4o'),
  middleware: obsvrMiddleware(),
})

const { text } = await generateText({
  model,
  prompt: 'Hello',
})

LlamaIndex

Use the obsvrLlamaIndexHandler to intercept LlamaIndex query engine and chat engine calls.

import { obsvrLlamaIndexHandler } from '@obsvr/sdk/llamaindex'

const handler = obsvrLlamaIndexHandler()
// Attach to your LlamaIndex engine callbacks

OpenAI Agents SDK

Use ObsvrTraceProcessor to capture agent traces from the OpenAI Agents SDK (Python). Every tool call, LLM invocation, and handoff is signed.

from obsvr import ObsvrTraceProcessor

processor = ObsvrTraceProcessor()
# Register as a trace processor with the Agents SDK

Agent tool governance

Wrap any framework's tool with obsvrGovernTool to govern it at the point of execution. One primitive works across Vercel AI, LlamaIndex, OpenAI Agents, and LangChain because it governs the tool's own execute function - not a framework-specific hook that changes between versions.

• Denied tools are blocked before they run (agentPolicy.deniedTools / allowedTools).
• Tool arguments are PII-scanned and redacted in the signed record.
• Every tool call is a signed tool.call event, tied into the same audit chain.

import { obsvr, obsvrGovernTool } from '@obsvr/sdk'

obsvr.init({ agentPolicy: { deniedTools: ['delete_file'] } })

// Wrap your tool once, then pass the governed version to your agent.
const safeCalculator = obsvrGovernTool(calculatorTool, { name: 'calculator' })
// Vercel AI:               tools: { calculator: safeCalculator }
// LlamaIndex / OpenAI Agents: tools: [safeCalculator]

MCP (Model Context Protocol)

Use obsvrGovernMCP to govern every MCP tool call, and mcpToolPolicy to allowlist or denylist tool names. Non-mutating: it returns a governed Client (a construct-trap proxy) - the MCP SDK's prototype is never patched, so it coexists with everything else.

• Allow/deny every tool call by name; denied tools are blocked before they run.
• Defends against tool poisoning: every tool description is scanned at discovery (tools/list) for instruction overrides, hidden directives, cross-tool chains, exfiltration, and concealment.
• Records the exact tool surface shown to the model at every discovery as a signed event - your queryable tool inventory.
• Drift detection: if a third-party server later changes a tool's description, it's flagged as drifted; a clean tool turning poisoned raises a high-severity incident (agent supply-chain defense).
• Set blockPoisonedTools to strip poisoned tools before the model sees them.

import { Client } from '@modelcontextprotocol/sdk/client/index.js'
import { obsvrGovernMCP, getConfig } from '@obsvr/sdk'

// Returns a governed Client - no prototype patching. Use the returned class.
const GovernedClient = obsvrGovernMCP(Client, getConfig())
const client = new GovernedClient({ name: 'app', version: '1.0.0' }, { capabilities: {} })

// Config-level tool policy
obsvr.init({
  mcpToolPolicy: {
    allowedTools: ['read_file', 'search'],
    deniedTools: ['execute_command'],
    // Remove tools with poisoned descriptions at discovery
    blockPoisonedTools: true,
  },
})

Works Behind Any Gateway

obsvr wraps your provider client in-process: no proxy, no method patching, no extra network hop.

• Composes cleanly with your stack: API gateways, LLM routers, caching layers, and egress proxies all keep working.
• Governance runs before the request leaves your process, and audit delivery rides its own async channel.
• Enable the OTel mirror and every governed call also lands in your existing tracing backend, with GenAI semantic-convention attributes plus obsvr's verdicts.

PII Scanner

A built-in scanner runs bounded, ReDoS-guarded detectors for PII, secrets, and prompt injection before every call (cards are Luhn-validated). Enable it with piiPolicy; each type has a default action you can override, and NLP types (names, addresses, medical, national IDs) need a connected Presidio instance.

obsvr.init({
  piiPolicy: {
    default: 'redact',
    rules: {
      ssn: 'block',
      credit_card: 'block',
      email: 'redact',
      phone: 'detect_only',
    },
  },
})

Policy Hooks

Use onPreCall and onPostCall hooks for custom policy logic. Pre-call hooks receive the prompt and can block, redact, or allow. Post-call hooks receive the response and can flag or redact it.

obsvr.init({
  onPreCall: async (event) => {
    if (event.prompt?.includes('confidential')) {
      return { decision: 'block', reason: 'Contains confidential data' }
    }
    return { decision: 'allow' }
  },
  hookTimeoutMs: 3000,
  hookTrigger: 'always', // 'always' | 'on_pii' | 'on_block'
})

Policy Rules

Structured rules are evaluated before the custom hook. Block wins over redact, and the first block short-circuits. Beyond text matching (keyword, regex, topics, PII), the engine includes governance rule types for actions, rate/token quotas, model allowlists, tenant isolation, and more.

obsvr.init({
  policyRules: [
    {
      id: 'block-competitor',
      name: 'Block competitor mentions',
      enabled: true,
      action: 'block',
      type: 'keyword',
      conditions: { keywords: ['competitor-name'] },
      applies_to: 'prompt',
    },
    {
      id: 'redact-internal',
      name: 'Redact internal project names',
      enabled: true,
      action: 'redact',
      type: 'regex',
      conditions: { pattern: 'PROJECT-(ALPHA|BETA|GAMMA)' },
      applies_to: 'both',
    },
  ],
})

Remote Policy Management

Manage policy rules from the dashboard instead of in code.

• Rules are served per project; the SDK polls for updates (default 30s, set 0 to disable).
• Incoming rules are schema-validated (including ReDoS checks) before taking effect: a rejected rule is never partially applied and emits an sdk:rule_rejected event.
• The same poll delivers approval grants and the kill switch, so pausing a project blocks all governed calls on that key within one interval and resumes on unpause.
• Each poll self-reports the client's version, supported rule types, and enforced policy hash, so the Fleet tab can flag clients that are behind or degraded.
• Every policy change is recorded in the append-only Change Log with actor and before/after.

obsvr.init({
  apiKey: process.env.OBSVR_API_KEY,
  // Poll for dashboard-managed rule updates every 30s (default).
  // When a poll succeeds, remote rules become the active rule set
  // (they replace code-defined policyRules). Set 0 to disable
  // polling and manage rules purely in code.
  policyRefreshIntervalMs: 30000,
})

Human-in-the-Loop Approvals

Any block rule with require_approval: true becomes a human-gated action.

• When it fires without a grant, the call is blocked with a signed record and a request is filed in the dashboard Approvals queue (deduplicated per rule and user).
• A reviewer grants a time-boxed approval (15 minutes to 24 hours) or denies; grants reach every SDK on the next poll, so an approved retry succeeds in about 30 seconds.
• Grants always expire, can be revoked early, and can be pinned to one end user via user_id.
• Each grant is pinned to the exact rule it was requested under, so editing the rule voids it.
• No permanent grants by design: a permanent exception is a policy change.

obsvr.init({
  policyRules: [
    {
      id: 'wire-transfer-gate',
      name: 'Wire transfers need human sign-off',
      enabled: true,
      action: 'block',
      type: 'keyword',
      conditions: {
        keywords: ['wire transfer'],
        require_approval: true,
      },
      applies_to: 'prompt',
    },
  ],
})

// First call blocks and files a request in the dashboard queue.
// After a reviewer approves, the retry passes within one policy poll.

Shadow Mode (Rehearse Before Enforcing)

The safe way to roll out a strict rule before it can block anything.

• Set mode: 'shadow' and the rule evaluates live traffic but never blocks, redacts, consumes quota, or files approvals.
• Each event instead records a shadow_outcome (which rule matched and what it would have done), shown as a badge in the audit trail.
• Watch the would-have outcomes for a few days, then flip to enforce.
• Conformance-tested: the active decision is byte-identical whether shadow rules are present or not, and flipping shadow/enforce changes the policy hash, so the transition is on the record.

obsvr.init({
  policyRules: [
    {
      id: 'candidate-topic-block',
      name: 'Considering: block pricing discussions',
      enabled: true,
      action: 'block',
      type: 'topic_deny',
      conditions: { topics: ['internal pricing'] },
      mode: 'shadow', // records outcomes, never enforces
    },
  ],
})

// Events now carry:
// shadow_outcome: {
//   rule_id: 'candidate-topic-block',
//   would: 'block',
//   reason: 'Considering: block pricing discussions',
// }

explain(): Check-Only Evaluation

explain() runs the same PII scan and policy evaluation a real call would, and returns the decision, the determining rule, the current policy hash, and any shadow outcome.

• No side effects: no quota consumed, no injection state advanced, no approvals filed, no events emitted.
• Customer hooks are skipped, since they may have side effects of their own.
• Use it in tests, CI, and anywhere you want to know what governance would do before sending a prompt.

import { explain } from '@obsvr/sdk'

const verdict = explain('My key is sk-live-abc123, is it valid?')
// {
//   decision: 'block',
//   rule_id: 'ex-block-secrets',
//   rules_hash: 'e8c53c78786308e7',
//   pii: { detected: true, types: ['api_key'] },
//   shadow_outcome: null,
//   not_evaluated: ['customer_hook', 'multi_turn_injection'],
// }

Multi-Turn Injection Scoring

Catches injection attacks split across several turns, where no single message trips a pattern.

• With multiTurnInjection enabled, the SDK keeps a per-session score (keyed by user_id) that adds weighted weak signals each turn.
• The score decays with a configurable half-life, so normal traffic never accumulates.
• When the decayed score crosses the threshold, the gate blocks or flags with rule id sdk:multi_turn_injection and the per-turn signals in the record.
• Full single-turn matches are still handled by the built-in scanner.

obsvr.init({
  multiTurnInjection: {
    enabled: true,
    threshold: 1.0,      // decayed score that trips the gate
    halfLifeMs: 600000,  // 10 minute half-life
    action: 'block',     // or 'flag'
  },
})

// Attribute turns to a session by passing user_id in metadata:
await client.chat.completions.create({
  model: 'gpt-4o-mini',
  messages,
  metadata: { user_id: 'user-123' },
})

Agent Policy

Run-level policies for agentic frameworks. Applied on top of per-call policies to enforce tool allowlists, step limits, and output restrictions across an entire agent execution.

obsvr.init({
  agentPolicy: {
    allowedTools: ['search', 'read_file', 'write_file'],
    deniedTools: ['execute_command', 'delete_file'],
    maxSteps: 25,
    stepLimitAction: 'escalate',
    allowPiiAccess: false,
    outputPolicy: {
      deniedTopics: ['financial-advice', 'medical-diagnosis'],
    },
  },
})

Signed, Tamper-Evident Trail

Every governed call becomes a signed, tamper-evident audit event.

• Each event is HMAC-signed at the source, hash-chained to the previous one, and countersigned on acceptance with a key that never leaves obsvr, so a client alone cannot forge a valid record and any edit breaks the chain.
• Each event records the exact policy hash the decision ran under (byte-identical across both SDKs), so any verdict can be reproduced against the precise rule set; shadow-mode rules record what they would have done.
• Deliveries are idempotent: a retried send lands on the same record instead of duplicating evidence.
• Content is hashed by default; raw prompt and response text is stored only if you opt into the raw layer (see Storage).
• Verify a chain client-side with obsvr.verifyAuditChain().

import { obsvr } from '@obsvr/sdk'

// Re-verify the SDK signature chain for a sequence of events.
const result = obsvr.verifyAuditChain(events)
// { valid: true, brokenAt: null } - any tampering flips valid to false

Alerts & Signed Webhooks

Each project can configure an HTTPS webhook for governance notifications.

• Fires on incidents (above a minimum severity), kill-switch changes, and new approval requests, so a pending approval is never missed.
• Payloads carry metadata only, never prompt or response text; target URLs are validated against private and cloud-metadata ranges.
• Set a signing secret and every delivery carries X-Obsvr-Signature (HMAC-SHA256 over timestamp and body) plus X-Obsvr-Timestamp, so you can verify origin and reject replays.
• Slack incoming webhooks work out of the box.
• The Compliance panel also exports a coverage report: active policy set and hash, fleet coverage, decision counts, change history, and approvals, with a machine-readable appendix.

// Express receiver verifying a signed obsvr webhook
import { createHmac, timingSafeEqual } from 'node:crypto'

app.post('/obsvr-hook', express.text({ type: '*/*' }), (req, res) => {
  const ts = req.get('X-Obsvr-Timestamp')
  const sig = req.get('X-Obsvr-Signature') // "sha256=<hex>"
  const expected = 'sha256=' + createHmac('sha256', process.env.OBSVR_WEBHOOK_SECRET)
    .update(ts + '.' + req.body)
    .digest('hex')
  const ok = sig && timingSafeEqual(Buffer.from(sig), Buffer.from(expected))
    && Math.abs(Date.now() - Number(ts)) < 5 * 60_000 // reject stale replays
  if (!ok) return res.status(401).end()
  const payload = JSON.parse(req.body)
  // payload.type: "incident" | "kill_switch" | "approval_requested"
  res.status(200).end()
})

Bring Your Own Storage

Route audit records to an S3 bucket in your own AWS account, configured per project in the dashboard Storage tab.

• You create a cross-account IAM role; the dashboard generates the trust and permission policies, pinned to your project via ExternalId.
• obsvr assumes the role with short-lived credentials, so no secret keys are ever entered or stored.
• Records land as date-partitioned JSON (canonical/customer={id}/date={date}/), ready for Athena.
• Choose which layers route to your bucket: hash records, raw content, or both. The raw layer gives full-content tracing under your own retention.
• Writes are best-effort, so a failure in your bucket never blocks ingestion, and a one-click test writes a probe object.

In-VPC Deployment (Data Plane)

Run the entire enforcement and audit pipeline inside your own AWS account with the deploy/customer-vpc Docker Compose stack (ingest, both Presidio containers, and the Merkle ledger).

• Your SDKs point at the VPC-internal endpoint; canonical and raw records stream to S3 buckets you own (Object Lock recommended).
• The countersigning HMAC key is generated and held by you.
• The only cross-boundary traffic is policy, approval, and key sync inbound, and hash-only event summaries outbound to power the dashboard.
• Remote policies, the kill switch, and approvals work identically here, on the same cadence as the managed service.
• Keep the ingest port VPC-internal; requests are still API-key authenticated, but there is no reason to expose it publicly.

cd deploy/customer-vpc
cp .env.example .env   # your buckets, your HMAC key, scoped service account
docker compose up -d
curl -s localhost:8787/health

Error Classification

The SDK classifies provider errors into one of four error_type values (or null on success). Audit-side errors never propagate to your application - they are logged and the LLM call proceeds normally. Connection-level failures (DNS, refused connections) fall through to api_error.

Policy Block Errors

When a pre-call policy returns a block decision, the SDK throws an Error instead of calling the LLM. The message is prefixed with "[obsvr] Request blocked by policy" and notes whether the cause was PII detection or a policy violation.

try {
  const res = await client.chat.completions.create({
    model: 'gpt-4o',
    messages: [{ role: 'user', content: prompt }],
  })
} catch (err) {
  if (err.message.includes('[obsvr] Request blocked by policy')) {
    // Policy blocked this call before it reached the provider
    console.log('Blocked:', err.message)
  }
}

ObsvrClient

For scenarios where the proxy wrapper isn’t suitable, use the manual ObsvrClient to send audit events directly.

• Useful for custom providers, batch pipelines, or non-JS runtimes sending over HTTP.
• Manual events bypass SDK compliance controls (PII scanning, policy rules, HMAC chain signing) and are stamped compliance_bypass by ingest.
• Prefer obsvr.wrap() for governed traffic.

import { ObsvrClient } from '@obsvr/sdk'

const client = new ObsvrClient({
  apiKey: process.env.OBSVR_API_KEY,
  baseUrl: 'https://ingest.obsvr.dev',
})

trackCompletion() / trackBatch()

Send individual or batched audit events manually.

• Each event requires prompt, response, model, and region; source, requestId, and metadata are optional (a UUID is generated when requestId is omitted).
• Manual events carry no token or latency fields and are stamped compliance_bypass by ingest.

await client.trackCompletion({
  model: 'gpt-4o',
  prompt: 'What is the meaning of life?',
  response: '42',
  region: 'us-east-1',
  source: 'batch-job',
  metadata: { pipeline: 'nightly-summaries' },
})