Metadata-Version: 2.4
Name: checkpoint-protocol
Version: 0.1.0
Summary: The control layer for AI agents. Check any agent action before it runs: allow, stop, or escalate.
Author: Checkpoint
License: MIT
Project-URL: Homepage, https://checkpointprotocol.xyz
Project-URL: Documentation, https://checkpointprotocol.xyz/docs
Keywords: ai,agents,control,policy,audit,langchain,crewai,autogen
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"

# Checkpoint Python SDK

The control layer for AI agents. Check any agent action before it runs, evaluate it against your policy, and decide: allow, stop, or escalate. Every decision is written to a tamper-evident audit log.

## Install

```bash
pip install checkpoint-sdk
```

## Quickstart

```python
import checkpoint as cp
from checkpoint import Policy, rule, CheckpointStopped

cp.set_policy(Policy([
    rule("transfer_funds", when="amount > 25000", verdict="escalate", to="cfo"),
    rule("export_records", when="pii_rows > 10000", verdict="stop"),
]))

def process_payment(amount, recipient):
    cp.check("transfer_funds", {"amount": amount, "recipient": recipient})
    transfer_funds(amount, recipient)  # only reached if allowed
```

If the amount is within policy, the check returns and the transfer runs. If it crosses a threshold, the check stops it (raising `CheckpointStopped`) or escalates to a human.

## Reading the verdict

```python
verdict = cp.check("transfer_funds", {"amount": amount}, raise_on_stop=False)
if verdict.allowed:
    transfer_funds(amount, recipient)
else:
    print(verdict.reason)
```

## The three verdicts

- allow: within bounds, proceeds instantly.
- stop: crosses a hard limit, execution stops.
- escalate: needs a human, routes to the right person and waits.

## Audit log

```python
for entry in cp.audit_log.entries():
    print(entry["log_id"], entry["verdict"], entry["action"])

assert cp.audit_log.verify()  # hash chain intact
```

## Framework integrations

Gate your agent's tools so `cp.check()` fires automatically before each one runs. A stopped action raises `CheckpointStopped` and the tool never executes; an escalation returns the verdict without running.

Any tool or function, with the `@guard` decorator:

```python
from checkpoint.integrations import guard

@guard("transfer_funds")
def transfer_funds(amount, to):
    ...

transfer_funds(amount=80000, to="acct_1")  # checked before it runs
```

Positional or keyword arguments are mapped to the payload by parameter name, so a rule on `amount` works either way. Shape the payload yourself with `payload_from` if your rules read different fields.

LangChain:

```python
from checkpoint.integrations import wrap_langchain_tool

guarded = wrap_langchain_tool(my_tool, action="transfer_funds")
# give `guarded` to your agent instead of the raw tool
```

CrewAI:

```python
from checkpoint.integrations import wrap_crewai_tool

guarded = wrap_crewai_tool(my_tool, action="transfer_funds")
```

AutoGen:

```python
from checkpoint.integrations import register_autogen

register_autogen(transfer_funds, action="transfer_funds",
                 agent=assistant, executor=user_proxy, name="transfer_funds")
```

The SDK has no hard dependency on any framework; each is imported only when you call its wrapper.

## Direct LLM tool calls (OpenAI, Anthropic, Gemini)

If you call a model directly instead of through an agent framework, the model proposes tool calls and your code runs them. Checkpoint checks each proposed call before you execute it.

```python
from checkpoint.llm import check_openai_tool_calls

resp = client.chat.completions.create(model="gpt-4", messages=msgs, tools=tools)

for c in check_openai_tool_calls(resp):
    if c.allowed:
        result = run_tool(c.name, c.args)
    else:
        result = c.verdict.reason   # stopped or escalated; tool not run
    # feed result back to the model as the tool result for c.id
```

Same shape for the other providers:

```python
from checkpoint.llm import check_anthropic_tool_calls, check_gemini_function_calls

for c in check_anthropic_tool_calls(message): ...
for c in check_gemini_function_calls(response): ...
```

Each returns a `CheckedCall` per proposed call, with `.allowed`, `.stopped`, `.escalated`, the parsed `.args`, the provider's call `.id` (for returning results), and the full `.verdict`. Nothing executes automatically; you run the allowed calls and stay in control of the loop.

Use `name_map={"wire_money": "transfer_funds"}` if the model's function name differs from your policy action, and `payload_from=...` to reshape arguments into the fields your rules read.

## Run the example and tests

```bash
PYTHONPATH=. python examples/payment_agent.py
pip install -e ".[dev]" && pytest
```

## Hosted mode

By default the SDK runs a local in-memory engine, so you can use it without an account. To evaluate against your team's shared policy and durable audit log, point it at the hosted service:

```python
cp.configure(api_key="cp_live_xxx", base_url="https://your-checkpoint-host")
```

With `base_url` set, `check()` calls `POST /v1/check` on the server. If the service is unreachable it fails closed (stops the action) by default. The public API is identical in both modes.

## License

MIT.
