prova-sdk 0.3.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 in-process

The handler also 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). That catch is asynchronous: by the time the receipt is signed, the agents have already burned the budget.

Pass break_on_loop=True to run the same detection algorithm in-process and stop the run the moment a persistent loop forms:

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("broke 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 caught. The detection is a faithful port of the canonical server-side detector, so a loop caught 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

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.

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).

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