Agent-side SDKs.

Run it on your own agent, no account

Wrap your real LangGraph or LangChain agent, run it, and Prova grades the run's health and tells you what it found: secrets or PII sent to a model, prompt-injection phrasing, where your spend went, runaway steps, and coordination loops. No API key, no network. Nothing leaves your machine.

pip install prova-sdk

from prova_cp import LocalCaptureHandler

handler = LocalCaptureHandler(app_id="my-agent")
graph.invoke(inputs, config={"callbacks": [handler]})   # your LangGraph / LangChain agent
print(handler.summary())

Run health: C  (70/100).  Flagged. Clearly broken.
  signals: severe_finding

Prova found 2 issues in this run (nothing left your machine).

  ! CRITICAL an OpenAI-style API key was sent in payload.prompt  [sk-...cd]
  - MEDIUM   possible PII (an email address) in payload.completion  [a@b...om]

  spend      $2.10 (local estimate); top driver gpt-4o (78%)
  steps      37   model calls 9

No LangChain? RunGuard gives the same findings-first report from two calls in any runtime (see "Any runtime" below). Already have a trace on disk? Pipe newline-delimited JSON through the CLI:

prova-local --file trace.ndjson   # add --fail-on-loop to gate CI

Already on LangSmith, Langfuse, or OpenAI? You do not need to instrument anything. Point prova-local at a raw export and it maps the logs to receipts and scores them offline, still no account, still nothing uploaded.

prova-local --source langsmith --file langsmith-export.ndjson

These figures are local estimates and the loop algorithm is identical to the server's, so what you see here is what a signed receipt would report. When you want the signed receipt and the dashboard, add a free API key and send the same events (next).

1. Send your first receipt

import { ProvaClient } from '@cobound/prova-sdk';

const prova = new ProvaClient({ apiKey: process.env.PROVA_API_KEY! });

await prova.ingest({
  kind: 'model_call',
  source: { org_id: 'YOUR_ORG', framework: 'langgraph', app_id: 'claims-orchestrator' },
  model: { provider: 'openai', name: 'gpt-4o' },
  payload: { messages, response },
});

from prova_cp import ProvaClient

prova = ProvaClient(api_key=os.environ["PROVA_API_KEY"])
prova.ingest({
    "kind": "model_call",
    "source": {"org_id": "YOUR_ORG", "framework": "langgraph", "app_id": "claims-orchestrator"},
    "model": {"provider": "openai", "name": "gpt-4o"},
    "payload": {"messages": messages, "response": response},
})

Both clients retry transient 429 / 5xx responses with exponential backoff (4 attempts by default) and respect the per-request Idempotency-Key header. Pass verifyReceipts: true (Node) or verify_receipts=True (Python) to verify the Ed25519 signature of every returned receipt before the call resolves.

2. Check before you act

For pre-execution enforcement, call gatewayCheck / gateway_check with the prospective payload. The response carries an explicit action.

const { action, findings } = await prova.gatewayCheck({
  kind: 'model_call',
  payload: { messages },
});
if (action === 'block') {
  throw new Error('blocked: ' + findings.map(f => f.detector).join(', '));
}

3. Catch coordination loops as they form

The callback handler (Node and Python) auto-instruments every node, LLM call, and tool call. It accumulates the agent trace, emits one agent_run receipt per run so the server-side coordination_loop detector fires, and by default logs a warning the moment a persistent loop forms. You see the loop in real time, not just later in the dashboard.

The default warns, it does not stop the run. A structural loop is also what a healthy planner and executor iteration looks like: the planner writes a plan, the executor reads it and writes a result, the planner reads the result and writes the next plan. Stopping every cycle would break agents that are working correctly. When you want the run halted, pass break_on_loop=True (Python) or breakOnLoop: true (Node) and catch the error. The run stops at the step the loop became persistent instead of running to your framework's recursion limit. The signed agent_run receipt is flushed before the exception propagates, and with cost attribution on (next section) that receipt carries exactly what the stopped run cost.

from prova_cp import ProvaCallbackHandler, CoordinationLoopError

# Default: warns the moment a loop forms, never stops the run.
handler = ProvaCallbackHandler(prova, app_id='claims-orchestrator')

# Opt in to stopping the run instead:
handler = ProvaCallbackHandler(prova, app_id='claims-orchestrator', break_on_loop=True)
try:
    graph.invoke(inputs, config={'callbacks': [handler]})
except CoordinationLoopError as e:
    log.error('stopped a loop across %s after %s steps',
              e.match['agents'], e.match['persistence_steps'])

import { ProvaCallbackHandler, CoordinationLoopError } from '@cobound/prova-sdk';

// Default warns; pass breakOnLoop to stop the run instead.
const handler = new ProvaCallbackHandler(prova, {
  appId: 'claims-orchestrator',
  breakOnLoop: true,
});
try {
  await graph.invoke(inputs, { callbacks: [handler] });
} catch (e) {
  if (e instanceof CoordinationLoopError) {
    console.error('stopped a loop across', e.match.agents);
  }
}

Scope: the detector fires on loops that keep revisiting the same state. A loop whose output drifts every round (a real model rephrasing the same non-progress) can slip past the inline check; catching those needs a semantic signal and is on the roadmap. Runnable end-to-end examples live in the repo under sdks/control-plane/examples/: langgraph_loop_demo.py (no LLM, free) and real_agents.py (real Claude calls, plus tool and RAG capture).

4. Cap runaway spend with a budget

Pass budget_usd to the handler and the run stops the moment its estimated spend crosses the cap. A budget is a hard limit you set, so unlike loop detection it stops the run by default. Combine it with max_steps and break_on_loop and you have a circuit breaker for runaway agents in one constructor.

from prova_cp import ProvaCallbackHandler, BoundaryViolationError

handler = ProvaCallbackHandler(
    prova,
    app_id='claims-orchestrator',
    budget_usd=0.50,     # stop if the run's estimated spend exceeds $0.50
    max_steps=40,        # stop after 40 agent steps
    break_on_loop=True,  # stop on a coordination loop
)
try:
    graph.invoke(inputs, config={'callbacks': [handler]})
except BoundaryViolationError as e:
    log.error('circuit breaker tripped: %s', e.match['dimension'])

const handler = new ProvaCallbackHandler(prova, {
  appId: 'claims-orchestrator',
  budgetUsd: 0.50,
  maxSteps: 40,
  breakOnLoop: true,
});

The local estimate uses a built-in price catalog (override with set_model_price / setModelPrice). The signed receipt carries a separate, canonical figure: the server computes payload.cost_usd from the model name and a maintained catalog and signs it, so the number Finance sees is the number an auditor verifies. The handler attaches normalized token_usage to every model_call receipt so that figure is always present. Set a monthly cap in the dashboard and the monthly_budget_cap policy blocks at the gateway too. View aggregated spend at /dashboard/spend.

5. Declare autonomy boundaries

Boundaries are a declarative contract for an agent run: which tools it may call, how many steps, how much it may spend, which data scopes it may touch. The SDK enforces in-process and raises the moment a dimension is crossed. The server-side boundary_violation policy enforces the same contract on every ingested event and produces a signed audit trail of any breach. The two implementations are pinned to one schema in lib/boundaries/schema.ts and verified by a contract test.

from prova_cp import ProvaBoundaries, ProvaCallbackHandler, BoundaryViolationError

boundaries = ProvaBoundaries(
    allowed_tools=['search', 'send_email'],
    max_steps=20,
    budget_usd_per_run=0.50,
    data_scopes=['customer_id'],
)
handler = ProvaCallbackHandler(prova, boundaries=boundaries, break_on_violation=True)
try:
    graph.invoke(inputs, config={'callbacks': [handler]})
except BoundaryViolationError as e:
    log.error('boundary breached: %s', e.match['dimension'])

6. Register an integration up front

Declare an integration before its first receipt so the AI Inventory can flag "wired up but never exercised in prod" gaps.

await prova.register({
  app_id: 'claims-orchestrator',
  provider: 'openai',
  model_name: 'gpt-4o',
  environment: 'production',
  framework: 'langgraph',
});

7. Verify a receipt offline

The verifier recomputes the canonical hash and checks the Ed25519 signature against the public key. Pass a PEM directly or let the SDK fetch it from /api/v1/keys/{keyId}.

import { verifyReceipt } from '@cobound/prova-sdk';
await verifyReceipt(receipt, { publicKeyPem: PUBLIC_KEY_PEM });

from prova_cp import verify_receipt
verify_receipt(receipt, public_key_pem=PUBLIC_KEY_PEM)

8. Migrate existing logs in one shot

A typed mapper bulk-imports existing observability data. Supported sources today: LangSmith run exports, Langfuse observation exports, and raw OpenAI ChatCompletion responses. Each row becomes a signed AIDecisionEvent with payload._migrated_from set to the source; idempotency keys derive from the source row id so reruns are safe.

PROVA_API_KEY=prv_... npx prova-migrate \
  --source langsmith \
  --file runs.ndjson

PROVA_API_KEY=prv_... prova-migrate \
  --source langfuse \
  --file observations.ndjson

Or call migrate() directly with an async iterable of parsed rows when the source isn't NDJSON on disk. The ingest endpoint accepts up to 1000 events per batch; the SDK defaults to 200 and surfaces per-batch progress through onProgress.

Any runtime: AutoGen, custom loops

No LangChain callbacks (AutoGen, a hand-rolled orchestrator)? RunGuard gives the same loop, budget, and step protection from two calls you place wherever your runtime advances. It imports no framework and makes no network calls.

from prova_cp import RunGuard, CoordinationLoopError, BoundaryViolationError

guard = RunGuard(budget_usd=0.50, max_steps=40, break_on_loop=True)
try:
    for turn in run_agent():            # AutoGen reply hook, or your own loop
        guard.observe_step(turn.agent, reads=turn.state_in, writes=turn.state_out)
        guard.observe_model_call(turn.model, turn.token_usage)
except (CoordinationLoopError, BoundaryViolationError) as e:
    log.error('circuit breaker tripped: %s', e)

import { RunGuard } from '@cobound/prova-sdk';

const guard = new RunGuard({ budgetUsd: 0.5, maxSteps: 40, breakOnLoop: true });
for (const turn of runAgent()) {
  guard.observeStep(turn.agent, turn.stateIn, turn.stateOut);
  guard.observeModelCall(turn.model, turn.tokenUsage);
}

The loop algorithm is the same one the server runs. Send the observed events to Prova for signed receipts and the dashboard.

OpenAI Agents SDK (Python)

On the OpenAI Agents SDK? Pass ProvaAgentsHooks as hooks= to Runner.run. Every LLM call, tool call, and agent step is ingested as a signed receipt, with token usage for server-side cost and the same coordination-loop detection the rest of the SDK uses. Use it as a context manager so the agent_run trace is flushed at the end of the run (that is what the server-side detector consumes).

from agents import Agent, Runner
from prova_cp import ProvaClient, ProvaAgentsHooks

prova = ProvaClient(api_key=os.environ["PROVA_API_KEY"])

async with ProvaAgentsHooks(prova, app_id="support-agent") as hooks:
    result = await Runner.run(agent, "...", hooks=hooks)

# Sync runs:
with ProvaAgentsHooks(prova, app_id="support-agent") as hooks:
    result = Runner.run_sync(agent, "...", hooks=hooks)

Pass break_on_loop=True, budget_usd=, or max_steps= to stop a run that loops or runs away. Every hook is fail-silent, so a Prova outage never breaks the agent.

Vercel AI SDK (Node)

Building on the Vercel AI SDK? Wrap your model with provaMiddleware and every generateText / streamText call emits a signed model_call receipt with token usage, so cost is computed and signed server-side. Streaming calls are tapped, not buffered. Works with ai v5 and v6.

import { wrapLanguageModel } from 'ai';
import { openai } from '@ai-sdk/openai';
import { ProvaClient } from '@cobound/prova-sdk';
import { provaMiddleware } from '@cobound/prova-sdk/vercel';

const prova = new ProvaClient({ apiKey: process.env.PROVA_API_KEY! });
const model = wrapLanguageModel({
  model: openai('gpt-4o'),
  middleware: provaMiddleware(prova, { appId: 'my-agent' }),
});
// use `model` with generateText / streamText exactly as before

Telemetry is fire-and-forget: a failed ingest never breaks your model call. The ai package is an optional peer dependency, so the SDK installs cleanly without it.

LlamaIndex (Python)

Register ProvaLlamaHandler on the LlamaIndex dispatcher and every LLM call, tool call, and retrieval becomes a signed receipt. Retrievals land as rag_retrieval receipts whose retrieved context the pii_in_retrieved_context policy screens for PII leaving the vector store.

from llama_index.core.instrumentation import get_dispatcher
from prova_cp import ProvaClient, ProvaLlamaHandler

prova = ProvaClient(api_key=os.environ["PROVA_API_KEY"])
get_dispatcher().add_event_handler(ProvaLlamaHandler(prova, app_id="rag-app"))

LlamaIndex is a query / RAG framework rather than a multi-agent loop runtime, so this handler produces the audit, cost, and retrieval-PII trail; use the LangGraph or OpenAI Agents adapters when you also want coordination-loop detection. Every handler call is fail-silent.

Pydantic AI (Python)

Pydantic AI exposes a run as a result object you inspect afterward, so run your agent as usual, then pass the result to ingest_pydantic_run. It emits one signed model_call receipt per model response (with token usage for server-side cost) and one tool_call receipt per completed tool call.

from prova_cp import ProvaClient, ingest_pydantic_run

prova = ProvaClient(api_key=os.environ["PROVA_API_KEY"])
result = await agent.run("...")
ingest_pydantic_run(prova, result, app_id="support-agent")

This is a post-run audit, not real-time enforcement. It reads the result's messages and usage defensively, so it works across Pydantic AI versions, and never raises out of the call.

Which SDK should I install?

Two packages share the prova-sdk name for legacy reasons. Pick by what you're building.

• Instrumenting agent runs, model calls, tool uses, or want signed receipts in your audit trail: install @cobound/prova-sdk (Node) or prova-sdk (Python 3.10+). This page.
• Verifying a single reasoning chain end-to-end (the legacy verifier product): see the verifier API reference.