prova-sdk 0.4.0

Agent-side SDK for the Prova AI control plane (ingest, gateway-check, register).

prova-sdk (Python)

Agent-side SDK for the Prova AI control plane. Thin wrappers around:

• POST /api/v1/audit/ingest
• POST /api/v1/gateway/check
• POST /api/v1/inventory

Plus an Ed25519 receipt verifier and a one-shot migration tool that bulk-imports existing LangSmith / Langfuse / OpenAI logs into the Audit Vault.

Separate from the legacy prova package (the reasoning-chain verifier). See /docs/sdk for guidance on which one to install.

Install

pip install prova-sdk

Requires Python 3.10+.

Quick start

from prova_cp import ProvaClient

prova = ProvaClient(api_key="prv_...")

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},
})

check = prova.gateway_check({"kind": "model_call", "payload": {"messages": messages}})
if check["action"] == "block":
    raise PolicyBlocked(check["findings"])

Pass verify_receipts=True to make the client verify every returned receipt's Ed25519 signature against the published public key before returning.

LangGraph / LangChain auto-instrumentation

Install the optional extra and drop the callback handler into any graph. Every LLM call, node, and tool call is ingested as a signed receipt automatically. No per-node code changes.

pip install "prova-sdk[langgraph]"

from prova_cp import ProvaClient, ProvaCallbackHandler

prova = ProvaClient(api_key="prv_...")
handler = ProvaCallbackHandler(
    prova,
    app_id="claims-orchestrator",
    environment="production",
    framework="langgraph",
)

# LangGraph
graph.invoke(inputs, config={"callbacks": [handler]})

# LangChain
chain.invoke(inputs, config={"callbacks": [handler]})

The handler is fail-silent: a Prova outage logs at warning level and never breaks the agent. LLM calls become model_call receipts, graph nodes become agent_step, tool calls become tool_call.

Catch the loop as it forms

The handler accumulates the {node, reads, writes} trace and emits one agent_run receipt per run, so the server-side coordination_loop detector fires from auto-instrumentation (it only triggers on agent_run, never on per-step events). By default it also runs the same detection in-process and logs a warning the moment a persistent loop forms, so you see it in real time rather than only later in the dashboard.

The default warns, it does not stop the run. A structural loop is also what a healthy planner/executor iteration looks like (planner writes a plan, executor reads it and writes a result, planner reads the result and writes the next plan), and stopping every cycle would break agents that are working correctly.

Pass break_on_loop=True to upgrade the warning to a stop. It raises CoordinationLoopError the moment the loop becomes persistent, before the run keeps burning budget:

from prova_cp import ProvaCallbackHandler, CoordinationLoopError

handler = ProvaCallbackHandler(prova, app_id="claims-orchestrator", break_on_loop=True)

try:
    graph.invoke(inputs, config={"callbacks": [handler]})
except CoordinationLoopError as e:
    # e.match: {agents, born_at_step, persistence_steps, total_steps, total_agents}
    log.error("stopped a coordination loop across %s", e.match["agents"])

The signed agent_run receipt is flushed before the exception propagates, so the audit trail records the loop you stopped. The detection is a faithful port of the canonical server-side detector, so a loop seen locally is the same loop an auditor sees in the receipt.

For a runtime without LangChain callbacks, drive LoopGuard directly:

from prova_cp import LoopGuard, CoordinationLoopError

guard = LoopGuard()  # raise_on_detect=True by default
for node, reads, writes in run_agent():
    guard.observe(node, reads=reads, writes=writes)  # raises on a persistent loop

Circuit breaker: stop runaway spend

budget_usd and max_steps are hard caps you set, so they stop the run by default (unlike loop detection, which warns). Combine them with break_on_loop and you have one circuit breaker for runaway agents:

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"])

Cost is estimated in-process from token usage using a built-in price catalog (override with set_model_price), so the run stops before the next call fires. That estimate is local only. The signed receipt's cost_usd is the canonical figure: the server computes it from the model name and a maintained catalog and signs it into the receipt.

CrewAI

CrewAI has no LangChain-style callbacks; use its step_callback / task_callback hooks instead.

from prova_cp import ProvaClient, ProvaCrewAI

tap = ProvaCrewAI(ProvaClient(api_key="prv_..."), app_id="research-crew")
crew = Crew(agents=[...], tasks=[...],
            step_callback=tap.step_callback,
            task_callback=tap.task_callback)

Agent steps become agent_step receipts; completed tasks become agent_run.

AutoGen and custom runtimes

AutoGen has no LangChain-style callbacks, and neither does a hand-rolled orchestrator. Use RunGuard: the same loop, budget, and step protection the callback handler gives LangGraph, driven by two calls you place wherever your runtime advances. It imports no framework and makes no network calls, so it works against any AutoGen version.

from prova_cp import RunGuard, CoordinationLoopError, BoundaryViolationError

guard = RunGuard(budget_usd=0.50, max_steps=40, break_on_loop=True)

def on_reply(recipient, messages, sender, config):
    # AutoGen ConversableAgent.register_reply hook (signature varies by version).
    guard.observe_step(sender.name, reads={"messages": len(messages)})
    return False, None  # let AutoGen continue

agent.register_reply([autogen.Agent, None], on_reply)

try:
    user.initiate_chat(agent, message=task)
except (CoordinationLoopError, BoundaryViolationError) as e:
    log.error("circuit breaker tripped: %s", e)

print(guard.report())  # {steps, estimated_cost_usd, loop}

The loop algorithm is the same one the server runs, so a loop caught here is the loop a signed receipt would report. Send the observed events to Prova when you want signed receipts.

Raw OpenAI / Anthropic clients (no framework)

Wrap the vendor client once. Every completion is mirrored to a signed receipt. The vendor response is returned unchanged and a Prova failure never raises. Synchronous, async (AsyncOpenAI/AsyncAnthropic), and streamed (stream=True) calls are all captured; for streams the receipt fires once after the stream is fully consumed, with the chunk text reassembled.

from openai import OpenAI
from prova_cp import ProvaClient, wrap_openai

client = wrap_openai(OpenAI(), ProvaClient(api_key="prv_..."), app_id="support-bot")
client.chat.completions.create(model="gpt-4o", messages=[...])  # auto-ingested

wrap_anthropic is identical for the Anthropic SDK (messages.create).

Run it locally, no account

Loop detection and cost estimation run entirely on your machine. No API key, no network, nothing leaves the process. Useful before you wire Prova into production, in CI as a gate, or air-gapped.

pip install prova-sdk
prova-local --file trace.ndjson          # human-readable report
prova-local --file trace.ndjson --json   # machine-readable
prova-local --file trace.ndjson --fail-on-loop   # exit 3 if a loop is found

The trace is newline-delimited JSON, one event per line in the shape the SDK emits. The loop algorithm is the same one the server runs, so a loop seen locally is the loop a signed receipt would report. Call analyze_events(events) to get the report programmatically. Send the same events to Prova for the signed receipt and dashboard.

Migrate existing logs

CLI:

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

Programmatic:

from prova_cp import ProvaClient, migrate
from prova_cp.migrate import read_ndjson

with ProvaClient(api_key="prv_...") as client, open("observations.ndjson") as f:
    result = migrate(client, "langfuse", read_ndjson(f))
    print(result)

Supported sources: langsmith, langfuse, openai. Idempotency keys are derived from the source row id, so re-running the migration is safe.

Verify a receipt offline

from prova_cp import verify_receipt

verify_receipt(receipt, public_key_pem=PUBLIC_KEY_PEM)

Or fetch the public key from the deployment automatically:

verify_receipt(receipt, base_url="https://api.prova.cobound.dev")