Metadata-Version: 2.4
Name: prova-sdk
Version: 0.4.0
Summary: Agent-side SDK for the Prova AI control plane (ingest, gateway-check, register).
Project-URL: Homepage, https://prova.cobound.dev/docs/sdk
Project-URL: Documentation, https://prova.cobound.dev/docs/sdk
License: MIT
Keywords: agents,ai,audit,compliance,langgraph,llm,observability
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: cryptography>=42.0
Requires-Dist: httpx>=0.27
Provides-Extra: langgraph
Requires-Dist: langchain-core>=0.2; extra == 'langgraph'
Description-Content-Type: text/markdown

# 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

```sh
pip install prova-sdk
```

Requires Python 3.10+.

## Quick start

```python
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.

```sh
pip install "prova-sdk[langgraph]"
```

```python
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:

```python
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:

```python
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:

```python
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.

```python
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.

```python
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.

```python
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.

```sh
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:

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

Programmatic:

```python
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

```python
from prova_cp import verify_receipt

verify_receipt(receipt, public_key_pem=PUBLIC_KEY_PEM)
```

Or fetch the public key from the deployment automatically:

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