Metadata-Version: 2.4
Name: chokepoint
Version: 0.3.0
Summary: Deterministic, code-defined guardrails for AI agent tool calls
Keywords: ai-agents,guardrails,llm,policy,safety,security
Author: Pedro Muracchini
Author-email: Pedro Muracchini <bonadioar@gmail.com>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Environment :: Console
Classifier: Framework :: Pytest
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: pydantic>=2.0
Requires-Dist: chokepoint[otel,langgraph,openai-agents,mcp] ; extra == 'all'
Requires-Dist: langchain-core>=0.3 ; extra == 'langgraph'
Requires-Dist: langgraph>=0.2 ; extra == 'langgraph'
Requires-Dist: mcp>=1.2 ; extra == 'mcp'
Requires-Dist: openai-agents>=0.1 ; extra == 'openai-agents'
Requires-Dist: opentelemetry-api>=1.43.0 ; extra == 'otel'
Requires-Dist: opentelemetry-sdk>=1.43.0 ; extra == 'otel'
Requires-Python: >=3.11
Project-URL: Homepage, https://github.com/pedromuracchini/chokepoint
Project-URL: Repository, https://github.com/pedromuracchini/chokepoint
Project-URL: Issues, https://github.com/pedromuracchini/chokepoint/issues
Project-URL: Changelog, https://github.com/pedromuracchini/chokepoint/blob/main/CHANGELOG.md
Provides-Extra: all
Provides-Extra: langgraph
Provides-Extra: mcp
Provides-Extra: openai-agents
Provides-Extra: otel
Description-Content-Type: text/markdown

# Chokepoint

[![CI](https://github.com/pedromuracchini/chokepoint/actions/workflows/ci.yml/badge.svg)](https://github.com/pedromuracchini/chokepoint/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/chokepoint.svg)](https://pypi.org/project/chokepoint/)
[![Python](https://img.shields.io/pypi/pyversions/chokepoint.svg)](https://pypi.org/project/chokepoint/)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](LICENSE)
[![Docs](https://img.shields.io/badge/docs-pedromuracchini.github.io-teal.svg)](https://pedromuracchini.github.io/chokepoint/)

Chokepoint expresses AI agent authorization policies as deterministic, code-defined
predicates evaluated by the runtime — never as natural-language instructions in a
prompt. Every tool call passes through the chokepoint (pre-hook and post-hook)
before and after it executes. The LLM never sees, interprets, or reasons its way
around a policy: policies are plain Python, evaluated outside the model's context.

Framework-agnostic — it's a middleware layer between an agent and its tools, not a
full agent harness. The
[architecture guide](https://pedromuracchini.github.io/chokepoint/architecture/)
explains the design rationale; the
[API reference](https://pedromuracchini.github.io/chokepoint/reference/) has the
signature-level detail.

## Install

```bash
uv add chokepoint                        # or: pip install chokepoint
uv add "chokepoint[otel]"                # with OpenTelemetry spans/metrics
uv add "chokepoint[langgraph]"           # to wrap LangChain/LangGraph tools
uv add "chokepoint[openai-agents]"       # to wrap OpenAI Agents SDK tools
uv add "chokepoint[mcp]"                 # to guard MCP tools/call, client or server side
uv add "chokepoint[all]"                 # every optional integration at once
```

## Batteries included

`chokepoint.policies` ships the rules nearly every agent needs, so you don't
start from a blank lambda. Each returns an ordinary `PolicySet` and composes
with `&` / `|` / `~` like anything you'd write by hand:

```python
from chokepoint import ChokepointInterceptor
from chokepoint.policies import (
    budget_policy, domain_allowlist, no_destructive_shell,
    no_secrets_in_args, path_within, rate_limit_policy, token_budget_policy,
)

interceptor = ChokepointInterceptor(policies=[
    no_secrets_in_args(),                                   # API keys, JWTs, PEM blocks
    no_destructive_shell(tool_names=("run_shell",)),        # rm -rf, mkfs, dd of=/dev/...
    path_within(["/srv/workspace"], tool_names=("write_file",)),
    domain_allowlist([".internal.corp"], tool_names=("http_get",)),
    rate_limit_policy(50),                                  # calls per session
    budget_policy(1000.0, lambda ctx: ctx.args["amount"], tool_name="transfer"),
    token_budget_policy(5.00, input_price=3.00, output_price=15.00,
                        tool_name="call_llm"),              # dollars per session
])
```

### Rate limits, budgets and token cost

Three shapes, because a limit is knowable at three different moments:

| Policy | Bounds | When the cost is known |
|---|---|---|
| `rate_limit_policy(n)` | number of calls per session | before the call |
| `budget_policy(max, amount_from=...)` | any figure derivable from the **arguments** | before the call — the cap is *never exceeded* |
| `token_budget_policy(max, input_price=…, output_price=…)` | dollars spent on model calls | only from the **response** — so the cap is *stop once spent* |
| `token_limit_policy(n)` | raw tokens per session | same as above |

An LLM call cannot be priced before it is made, so the call that crosses the
threshold completes and the *next* one is refused. That overshoot is inherent,
not a shortcut — `budget_policy(max, amount_from=..., actual_from=...)` bounds
it by charging an estimate up front and reconciling with the real figure
afterwards. Prices are per **million** tokens, the way providers quote them,
and no pricing table ships with the library: rates change, and a stale constant
baked into a security library would silently mis-bill.

`token_budget_policy` reads `usage` off the response and understands the
Anthropic (`input_tokens`), OpenAI (`prompt_tokens`) and Google
(`promptTokenCount`) shapes, as dicts or as SDK objects. A response with no
usage block contributes zero rather than failing closed. `on_fail=ESCALATE` is
often the better fit here — a human can decide whether this particular run is
worth more money.

Scope them with `tool_names` — an unscoped policy applies to every tool through
the same interceptor. The pattern matchers are seatbelts against agent
mistakes, not a sandbox: an adversary in full control of the input can evade
them.

## Quickstart

```python
from chokepoint import guard, BLOCK, ESCALATE

@guard(
    pre=lambda ctx: ctx.state_checksum_matches(),
    on_fail=BLOCK,
    reason="state changed since planning",
)
def delete_record(record_id: str) -> dict:
    ...

@guard(
    pre=lambda ctx: ctx.args["amount"] < 500,
    on_fail=ESCALATE,
    escalate_to="slack://finance-approvals",
    timeout_s=300,
)
def transfer_funds(amount: float, to: str) -> dict:
    ...
```

A blocked or denied-escalation call raises `chokepoint.GuardBlocked` instead of
running. Guarded functions keep their original signature — call them
positionally or by keyword as usual — and `ctx.args` is always the flat
name → value mapping, mirroring how agent frameworks pass tool-call arguments
as a JSON object.

Every error Chokepoint raises on purpose derives from `chokepoint.ChokepointError`,
so one `except` clause catches the library's failures without swallowing your
own bugs:

```python
try:
    transfer_funds(amount=1000, to="alice")
except chokepoint.GuardBlocked as exc:
    print(exc.decision.reason, exc.decision.policy_name)
except chokepoint.ChokepointError:
    ...  # misconfiguration, escalation transport failure, ledger problem
```

Each one also subclasses the stdlib exception it replaced
(`ConfigurationError` is a `ValueError`, `LedgerEventNotFound` a `KeyError`),
so existing handlers keep working.

With nothing registered for the `"slack"` scheme, that `ESCALATE` above
always denies (fail-safe) — register a real handler to actually resolve it:

```python
import os

from chokepoint import SlackEscalationHandler, register_handler

register_handler(
    "slack",
    SlackEscalationHandler(
        bot_token=os.environ["SLACK_BOT_TOKEN"],
        channel="C_FINANCE_APPROVALS",
        approvers={"U_CFO", "U_CONTROLLER"},  # required — see SECURITY.md
    ),
)
```

`WebhookEscalationHandler` (one synchronous POST, expects
`{"approved": bool}` back) and `CLIEscalationHandler` (local
`input()`-based human-in-the-loop) are also built in — all three use only
`urllib.request`/stdlib, no new dependency. Run
`uv run python examples/real_escalation_handlers.py` for all three,
end to end.

## Multi-agent: the same tool, different outcomes per caller

```python
from chokepoint import ChokepointRegistry, ChokepointInterceptor, AgentScopedPolicy, BLOCK

registry = ChokepointRegistry()
registry.register("clinical_agent", role="licensed_physician")
registry.register("support_agent", role="support_staff")

delete_policy = AgentScopedPolicy(
    name="only_physician_can_delete_patient",
    allowed_roles=["licensed_physician"],
    applies_to=lambda ctx: ctx.tool_name == "delete_patient",
    on_fail=BLOCK,
    reason="deleting a patient record is restricted to physicians",
)

clinical = ChokepointInterceptor(registry=registry, agent_id="clinical_agent", policies=[delete_policy])
support = ChokepointInterceptor(registry=registry, agent_id="support_agent", policies=[delete_policy])

clinical.call("delete_patient", delete_patient, id=1)  # allowed
support.call("delete_patient", delete_patient, id=1)   # GuardBlocked
```

`call()` takes the tool's arguments as keywords for brevity, but `session_id`
and `domain` are its own parameters. If your tool declares an argument by
either name, pass the arguments explicitly instead — Chokepoint warns rather than
silently dropping one:

```python
interceptor.call("send", send, args={"domain": "example.com", "body": "hi"})
```

Tools wrapped with `interceptor.use(agent)` or `wrap_tool()` are immune: every
keyword they receive is forwarded as a tool argument.

Run the full worked example (registry, `ReversibleAction`, escalation, two
interceptors) with:

```bash
uv run python examples/clinical.py
```

## Async tools

`@guard` and `ChokepointInterceptor.acall()` auto-detect an `async def` tool
function — no separate decorator or interceptor class needed:

```python
@guard(pre=lambda ctx: ctx.args["amount"] < 500, on_fail=BLOCK, reason="too large")
async def transfer_funds(amount: float, to: str) -> dict:
    return await payments_api.transfer(amount, to)

await transfer_funds(amount=100, to="alice")  # await it, same as the undecorated function
```

Predicates stay synchronous (cheap, deterministic checks); `ReversibleAction.do_fn`/
`undo_fn` and a custom `EscalationHandler.escalate` may be `def` or `async def`.
Run the full worked example with `uv run python examples/async_tool.py`.

## Framework integration

`interceptor.use(agent)` (or the module-level `chokepoint.wrap(agent, interceptor)`)
auto-detects LangChain/LangGraph `BaseTool` objects and OpenAI Agents SDK
`FunctionTool` objects — no manual adapter registration needed:

```python
# LangGraph / LangChain — wrap the tool list before building the graph
wrapped_tools = interceptor.use([transfer_funds, search_web])
create_react_agent(model, tools=wrapped_tools)

# OpenAI Agents SDK — wrap an Agent's .tools in place
agent = Agent(name="finance_agent", tools=[transfer_funds, search_web])
interceptor.use(agent)
```

MCP is auto-detected too, on either side of the protocol:

```python
# Client side — police what your agent asks any server to do. A denial raises.
chokepoint.wrap(client_session, interceptor)

# Server side — the policy holds whoever connects. A denial returns
# CallToolResult(isError=True), because raising would kill the connection.
chokepoint.wrap(mcp_server, interceptor)
```

Works with mcp 1.x and 2.x alike — the adapter detects which registration API
the installed package exposes, so `MCPServer` (2.x), `FastMCP` (1.x) and a
low-level `Server` are all accepted, as is 2.x's `Client` facade on the client
side.

Requires the matching extra (`chokepoint[langgraph]` / `chokepoint[openai-agents]` /
`chokepoint[mcp]`). Run `uv run python examples/langgraph_integration.py`,
`examples/openai_agents_integration.py`, or `examples/mcp_integration.py`.

## Reversible actions

```python
from chokepoint import ReversibleAction

delete_s3_bucket = ReversibleAction(
    do_fn=lambda args: s3.delete_bucket(args["bucket"]),
    undo_fn=lambda args, snapshot: s3.restore_from_snapshot(snapshot),
    name="delete_s3_bucket",
    irreversibility_level="high",   # auto-escalates before every execution
    escalate_to="slack://infra-approvals",
    pre_snapshot=lambda args: s3.snapshot(args["bucket"]),
)
```

`irreversibility_level`: `"low"` runs normally; `"medium"` requires `undo_fn` (raised
at construction if missing); `"high"` auto-escalates before every call; `"permanent"`
is an unconditional block — the action never runs.

A `"high"` action needs an `escalate_to` target to differ from `"permanent"`:
with none, its escalation resolves to the fail-safe handler, which denies, so
every call is blocked. `chokepoint lint` warns about this.

## Examples

Every file under `examples/` is runnable directly (`uv run python examples/<name>.py`):

| File | Demonstrates |
|---|---|
| `quickstart.py` | The smallest possible setup — two `@guard`-decorated functions, BLOCK and ESCALATE. |
| `policy_composition.py` | `&` / `\|` / `~` — combining `Policy` objects into a deploy-pipeline gate. |
| `dry_run_rollout.py` | Rolling out a new policy safely: `mode="dry_run"` logs what *would* block, then flip to `enforce`. |
| `custom_escalation_handler.py` | A minimal, hand-rolled `EscalationHandler` (toy in-process approve/deny logic). |
| `real_escalation_handlers.py` | The three built-in real handlers — `SlackEscalationHandler`, `WebhookEscalationHandler`, `CLIEscalationHandler` — each demonstrated end to end. |
| `reversible_levels.py` | `ReversibleAction`'s four `irreversibility_level`s side by side. |
| `delegation_chain.py` | Confused-deputy prevention via delegation-chain depth (`max_delegation_depth_policy`), not just role. |
| `clinical.py` | The full multi-agent story: `ChokepointRegistry`, `AgentScopedPolicy`, `ReversibleAction`, escalation, two interceptors sharing one tool. |
| `multi_agent_orchestrator.py` | A centralized registry with 4 agents, several tools, role + trust-level policies together, and an exported delegation graph. |
| `async_tool.py` | `@guard` and `ChokepointInterceptor.acall()` on `async def` tools. |
| `langgraph_integration.py` | Wrapping real `langchain_core.tools.BaseTool` objects for LangGraph (requires `chokepoint[langgraph]`). |
| `openai_agents_integration.py` | Wrapping real `agents.FunctionTool` objects for the OpenAI Agents SDK (requires `chokepoint[openai-agents]`). |
| `mcp_integration.py` | Guarding MCP `tools/call` from both the client and the server side (requires `chokepoint[mcp]`). |
| `builtin_policies.py` | Every policy in `chokepoint.policies` — secrets, destructive SQL/shell, path confinement, domain allowlist, rate limit, budget. |
| `redaction.py` | Keeping secrets and PII out of the ledger, the JSONL sink, the exports and the Slack message. |
| `audit_and_reporting.py` | `ActionLedger`'s compliance/graph/narrative/pytest-fixture export methods. |

## Audit trail

Every decision is recorded to a process-wide `ActionLedger`. In memory it's a
bounded ring buffer (10,000 events by default) — a memory bound, not a
durability story. For a full lossless history, point it at a JSONL file once at
startup, before the first guarded call:

```python
import chokepoint

chokepoint.configure_ledger(sink_path="/var/log/chokepoint/decisions.jsonl")
```

Every event is mirrored to that file regardless of the in-memory cap, and
`chokepoint report --ledger /var/log/chokepoint/decisions.jsonl` reads it back. A
sink write that fails is logged and counted in `ActionLedger.sink_error_count`,
never raised: recording happens after the tool has already run, so a full disk
must not turn a call the policies allowed into an exception.

For several tenants in one process, give each interceptor its own ledger
instead of configuring the global one:

```python
interceptor = ChokepointInterceptor(policies=[...], ledger=ActionLedger(sink_path=...))
```

### Redaction

Tool arguments reach the ledger, that JSONL file, the JSON/CSV exports and the
escalation message posted to Slack. **Credential-shaped values are scrubbed
out of all of them by default**, along with values under names like
`password` / `api_key` / `authorization`:

```python
interceptor.call("call_api", call_api, authorization="Bearer AKIAIOSFODNN7EXAMPLE")
# ledger: {"authorization": "[REDACTED]"}
```

Policies still evaluate against the **real** arguments — redaction happens at
record time, so a policy that inspects a credential can still see it. PII
patterns (email, SSN, IBAN, Luhn-checked card numbers) are opt-in, because an
email address is often the point of the call:

```python
chokepoint.configure_redaction(include_pii=True, keys=["mrn", "dob"])
chokepoint.configure_redaction(enabled=False)          # off entirely
chokepoint.configure_redaction(redactor=MyDLPClient())  # your own scrubber
```

One tradeoff: `chokepoint.replay()` reconstructs its context from the stored
event, so replaying a redacted call feeds placeholders to the predicates.
`ReplayResult.redacted` flags it, and `export --format fixtures` skips those
events instead of emitting tests that cannot pass. Run
`uv run python examples/redaction.py` for the whole picture.

## CLI

```bash
uv run chokepoint --version
uv run chokepoint report --agent my_agent.py                        # policy inventory + coverage
uv run chokepoint report --agent my_agent.py --format mermaid       # coverage graph
uv run chokepoint report --agent my_agent.py --delegation           # delegation graph
uv run chokepoint report --agent my_agent.py --fail-under 0.8       # CI gate on tool coverage
uv run chokepoint lint --agent my_agent.py                          # static policy checks
uv run chokepoint replay evt_a3f9b2                                 # replay a ledger event
uv run chokepoint repl --agent my_agent.py                          # synthetic-context REPL
uv run chokepoint export --format narrative --ledger audit.jsonl    # plain-English audit summary
uv run chokepoint export --format fixtures -o test_policies.py      # pytest tests from real decisions
```

`report --fail-under` and `lint` both exit non-zero on failure, so they work as
CI gates. `export` accepts `json`, `csv`, `narrative` and `fixtures`.

The CLI loads `my_agent.py` as a plain module and expects a module-level
`POLICIES: list[Policy]` (and, optionally, `REGISTRY` / `TOOL_NAMES` /
`ACTIONS`) — see
`examples/clinical.py`.

## Development

```bash
uv sync --extra all
uv run pytest -q
uv run ruff check .
uv run ruff format --check .
uv run mypy src/chokepoint
```

Commits follow [Conventional Commits](https://www.conventionalcommits.org/) and
CI enforces it — see `CONTRIBUTING.md`.

Docs are published at
[pedromuracchini.github.io/chokepoint](https://pedromuracchini.github.io/chokepoint/),
assembled from this repository's markdown plus an API reference generated from
the package's own docstrings:

```bash
uv run --group docs python scripts/build_docs.py
uv run --group docs mkdocs serve
```

The [architecture guide](docs/architecture.md) covers the design rationale and
what is deliberately out of scope.
