Metadata-Version: 2.4
Name: veefore-py
Version: 1.7.0
Summary: VEEFORE SDK for Python agent runtimes.
Author: VEEFORE
License: Apache-2.0
Project-URL: Homepage, https://veefore.com
Project-URL: Documentation, https://docs.veefore.com/build/python-sdk
Project-URL: Repository, https://github.com/veefore/veefore
Keywords: agent,identity,policy,authorization,enforcement
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Security
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: cryptography>=41.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.4; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"

# veefore-py

Identity, policy and enforcement for AI agents. Every protected tool call is
decided by the Veefore gateway before your code runs it.

```bash
pip install veefore-py
```

The package is `veefore-py`; the module is `veefore`. The quickest way to
integrate is the CLI, which installs this package with pip, uv or poetry:

```bash
npm install -g veefore
veefore              # sign in, scan your project, set it up, wrap your tools
```

Or by hand:

```python
from veefore import guard

@tool
@guard("refund_customer")
def refund_customer(amount: int) -> str:
    ...
```

`@guard` asks Veefore first and runs the function only if the call is
permitted. A refusal raises `VeeforeBlockedError` (or
`VeeforeApprovalRequiredError` when a human must decide). Put it directly above
`def`, below your framework's decorator.

## Chatbots: don't wait for an approval, defer it

By default a guarded tool waits for a human to approve. That is right for a
batch job and wrong for a chatbot: a chat turn times out in seconds, and an
approval can take a quarter of an hour. Waiting hangs the turn, and by the time
the reviewer clicks approve there is nothing left listening.

Use `on_approval="defer"`. The tool returns immediately with a dict the model
can say out loud, and the action is finished on a later turn by
`resume_approvals()`.

```python
from veefore import guard, resume_approvals, with_approval_session

@tool
@guard("refund_order", on_approval="defer")
def refund_order(order_id: str, amount_minor: int) -> dict:
    return payments.refund(order_id, amount_minor)


def on_message(chat_id: str, text: str) -> str:
    with with_approval_session(chat_id):
        # Always first: finishes anything a reviewer decided since last turn.
        resolved = resume_approvals()

        messages = [*history(chat_id), {"role": "user", "content": text}]
        for r in resolved:
            messages.append(
                {"role": "system", "content": f"{r['message']} {r.get('result', '')}"}
            )
        return run_model(messages, tools=[refund_order])
```

The conversation that produces:

> **user** — refund order 1234, it was ₹5,000
> **bot** — This one needs a quick sign-off from someone on our team before I
> can run it. I've sent it over, and I'll confirm as soon as they respond.
>
> *…eleven minutes later, a reviewer approves it in the console…*
>
> **user** — any update?
> **bot** — Good news: that was approved and has now gone through.

What each piece is for:

- **`resume_approvals()`** — call it at the top of every turn, before the model
  runs. It costs nothing when nothing is pending, and it is the only thing that
  has to happen for a deferred action to complete. Each resolution carries a
  `message` written for the model and a `say` sentence for the end user. Use
  `await resume_approvals_async()` if your tools are `async`.
- **`with_approval_session(chat_id)`** — required once you serve more than one
  user. Without it every pending approval sits in one pool and whoever speaks
  next completes everybody's actions. It is a `contextvars` variable, so it
  follows `asyncio` tasks correctly.
- **`pending_approval_tool()`** — optional. A ready-made
  `check_pending_approvals` tool definition, for when the user asks "any
  update?" and you would rather the model had something to call than guessed.
- **`grace_seconds`** — optional. Holds the turn briefly first, so an approval
  signed off in three seconds finishes in-band instead of becoming a
  "please wait".

The action runs exactly once, guarded three deep: Veefore consumes an approval
on first use and refuses it thereafter, the pending record is claimed under a
lock before it is replayed, and the replay's idempotency key is derived from the
approval so two concurrent turns collapse into one recorded action. A denial, an
expiry or an unreachable gateway all mean the tool did not run — nothing in this
path can turn an unknown outcome into permission.

## Letting the CLI set it up

```bash
veefore init --approval-mode defer
veefore integrate              # writes on_approval="defer" into each @guard
veefore doctor                 # fails while resume_approvals() is missing
```

`veefore integrate` writes the keyword argument into every tool it decorates,
and prints the `resume_approvals()` snippet for your framework. It does not
insert that call: a turn loop is framework specific and often nowhere near a
tool, and guessing there would be guessing in the one place where being wrong
is invisible. So `veefore doctor` fails until the call exists — deferring
without resuming drops approved actions silently, which is worse than the hang
it replaced.

## The three ways an approval completes

Which one applies depends on how the tool is protected, and a project usually
has more than one. `veefore integrate` and `veefore doctor` both say which.

| Protection | On approval required | How it completes |
| --- | --- | --- |
| `@guard("key")` | returns a deferred dict | `resume_approvals()` runs the tool |
| `guard(expr, …)(args)` at a dispatch point | returns a deferred dict | `resume_approvals()` runs it |
| `checkpoint("key", args)` | **raises** `VeeforeApprovalDeferredError` | the agent reaches the same checkpoint again |
| `mcp_call` / `a2a_invoke` | raises `VeeforeApprovalRequiredError` | does not defer; handle it yourself |

`checkpoint` raises rather than returning, and that is deliberate: it returns
`None` and the statements after it run unconditionally, so a returned status
could be ignored and the action would happen without approval. Raising is the
only outcome that reliably stops the block.

**Which means you need one line where you already catch.** An agent loop almost
always sits inside a `try/except` that turns any exception into "sorry,
something went wrong" — and that will swallow the wait message, showing your
user a failure for an action that is sitting in the approval queue working
perfectly:

```python
from veefore import approval_notice

try:
    return run_agent_turn(message)
except Exception as error:
    notice = approval_notice(error)
    if notice:
        return notice["say"]   # "needs a quick sign-off…" / "a reviewer denied this"
    raise                      # a real error, handled as before
```

`approval_notice()` returns `None` for anything that is not an approval, so
adding it cannot hide a genuine fault. `veefore doctor` fails while it is
missing.

It completes differently too. The SDK never holds those statements, so it
cannot run them for you. Instead the checkpoint recognises its own earlier
deferral when the agent arrives again:

```python
if call.name == "forget_memory":
    checkpoint("forget_memory", call.args)   # turn 1: raises, block stops
    forget(call.args)                        # turn 3: runs, once approved
```

The exception's message tells the model exactly that — relay the wait, and call
the same tool again with the same arguments later. `resume_approvals()` reports
such an approval as `approved_retry` rather than `completed`, because it did not
run it. Same arguments matter: Veefore binds an approval to the request that
created it, so a checkpoint reached with different arguments needs its own.

## Configuration

Read from the environment variables `veefore init` writes to `.env.local`.
Requires Python 3.9 or newer.
