Metadata-Version: 2.4
Name: sieve-layer
Version: 0.1.0
Summary: Drop-in middleware that intercepts every tool call your AI agents make, blocks unsafe actions, and logs a tamper-proof audit trail.
License: MIT
License-File: LICENSE
Keywords: ai-safety,audit,langchain,mcp,middleware,policy
Classifier: Development Status :: 3 - Alpha
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
Requires-Python: >=3.10
Requires-Dist: pyyaml>=6.0
Provides-Extra: dev
Requires-Dist: fastmcp>=2.0; extra == 'dev'
Requires-Dist: langchain-community>=0.3; extra == 'dev'
Requires-Dist: langchain-core>=0.3; extra == 'dev'
Requires-Dist: pip-audit>=2.7; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.3; extra == 'langchain'
Provides-Extra: mcp
Requires-Dist: fastmcp>=2.0; extra == 'mcp'
Provides-Extra: similarity
Requires-Dist: fastembed<1,>=0.8; extra == 'similarity'
Description-Content-Type: text/markdown

# Sieve

**Middleware for AI agent tool calls with policy checks, approval prompts, loop detection, and tamper-evident audit logs.**

---

## Quickstart

```bash
python -m pip install sieve-layer[langchain,similarity]
cp examples/policy.yaml policy.yaml
python examples/toy_agent.py
sieve verify examples/demo_audit.db
```

For contributors working from this repository, use the editable install:

```bash
python -m pip install -e ".[langchain,similarity]"
cp examples/policy.yaml policy.yaml
python examples/toy_agent.py
sieve verify examples/demo_audit.db
```

Use the same Python interpreter for installation and demo commands. If you use
a different interpreter, install the package and extras there first.

When the demo prompts for the DELETE query, type `n`. You should see the call
blocked, then a clean audit verification at the end.

### 1. Install the repo

```bash
python -m pip install -e ".[langchain,similarity]"
```

### 2. Copy the example policy

```bash
cp examples/policy.yaml policy.yaml
```

If you want to inspect or customize it first, this is the shipped example:

```yaml
# policy.yaml
version: 1
default: deny      # fail-closed: anything not explicitly allowed is blocked

rules:
  - name: allow-selects
    match:
      tool: "postgres_query"       # fnmatch glob on the tool name
      args:
        query: "(?i)^\\s*select"   # regex matched against the argument value
    action: allow

  - name: guard-mutations
    match:
      tool: "postgres_query"
      args:
        query: "(?i)\\b(delete|drop|update|truncate|insert)\\b"
    action: require_approval       # blocks and prompts operator at the terminal

circuit_breaker:
  - tool: "*"
    max_similar_calls: 3
    window_seconds: 60
    similarity_threshold: 0.92
    action: deny
```

### 3. Configure Sieve at startup

```python
import sieve

sieve.configure(
    policy_path="policy.yaml",
    db_path="audit.db",          # SQLite file for the tamper-evident audit log
)
```

### 4. Guard plain functions or tools

#### Plain functions

```python
@sieve.guard()
def postgres_query(query: str) -> str:
    # only runs if policy allows or operator approves
    return run_sql(query)

result = postgres_query("SELECT * FROM users")   # allowed
postgres_query("DELETE FROM users")              # blocked or prompts for approval
```

#### LangChain

```python
from langchain_core.tools import BaseTool
from sieve.integrations.langchain import wrap_tool

class MyPostgresTool(BaseTool):
    name = "postgres_query"
    description = "Execute SQL"

    def _run(self, query: str) -> str:
        return run_sql(query)

safe_tool = wrap_tool(MyPostgresTool())
# safe_tool.invoke({"query": "DELETE ..."}) will trigger Sieve before execution
```

#### FastMCP

```python
from fastmcp import FastMCP
from sieve.integrations.mcp import SieveMiddleware

mcp = FastMCP("my-server")
mcp.add_middleware(SieveMiddleware())

@mcp.tool()
def postgres_query(sql: str) -> str:
    return run_sql(sql)
```

### 5. Run the demos

```bash
python examples/toy_agent.py
python examples/toy_agent.py --loop-demo
```

This runs a toy LangChain agent against a local SQLite database. A `SELECT` passes through immediately; a `DELETE` triggers the approval prompt:

```
============================================================
[SIEVE] Approval required for tool call:
  Tool : postgres_query
  Args : {
  "query": "DELETE FROM users WHERE id = 1"
}
============================================================
Approve this call? [y/N]:
```

Type `n` to block it. The `PolicyViolation` is raised and the DELETE never executes.

The loop demo runs a deterministic toy search agent that retries the same query
with tiny whitespace changes. Sieve allows the first two calls, then emits a
`CIRCUIT_BREAK` audit entry and halts the 3rd jittered repeat.

### 6. Inspect the audit log

```bash
sieve tail examples/demo_audit.db    # print recent entries
sieve verify examples/demo_audit.db  # verify the hash chain is intact
```

---

## Policy schema

See [docs/POLICY_SCHEMA.md](docs/POLICY_SCHEMA.md) for the full field-by-field
schema reference.

| Field | Type | Description |
|-------|------|-------------|
| `version` | int | Must be `1` |
| `default` | string | Action when no rule matches: `allow`, `deny`, or `require_approval` |
| `rules` | list | Ordered list of rules; first match wins |
| `rules[].name` | string | Human-readable identifier used in audit log |
| `rules[].match.tool` | string | fnmatch glob matched against the tool name |
| `rules[].match.args` | mapping | `arg_name: regex`; all patterns must match |
| `rules[].action` | string | `allow`, `deny`, or `require_approval` |
| `circuit_breaker[]` | list | Optional embedding-similarity repeat detectors |
| `circuit_breaker[].tool` | string | fnmatch glob for tools covered by this breaker |
| `circuit_breaker[].max_similar_calls` | int | Number of similar calls, including current call, before firing |
| `circuit_breaker[].window_seconds` | number | Recent-call window to inspect |
| `circuit_breaker[].similarity_threshold` | number | Cosine threshold for argument embeddings |
| `circuit_breaker[].action` | string | `deny` or `require_approval` |
| `max_cost_per_task` | number | Optional cumulative LLM cost limit for one configured runtime |

**Matching semantics:**
- Tool name uses `fnmatch` glob: `postgres*` matches `postgres_query`, `postgres_write`, etc.
- Argument patterns are full Python `re.search()` regexes applied to `str(value)`.
- All `args` patterns must match for a rule to fire (logical AND).
- Rules are evaluated top-to-bottom; the first match wins.
- Circuit breakers compare embedded, canonicalized args only within the same tool name.
- Cost tracking requires attaching a Sieve callback to the LLM/agent and providing model prices.

---

## Tamper-evident audit chain

Every tool call decision is written to a SQLite database as a hash-chained log:

```
entry_hash = sha256(prev_hash + canonical_json(timestamp, tool_name, tool_args, outcome, ...))
```

The first entry uses `prev_hash = "0" * 64` (genesis). Sieve stores full
64-character SHA-256 hashes. Any retrospective modification, including editing
an outcome, deleting a row, or rewriting args, breaks the chain and is detected by
`sieve verify`.

New audit databases are created private to the current user, and Sieve removes
group/world permission bits from existing audit DB files when opening them.
Likely secrets in tool args and audit metadata are redacted before logging.

```bash
$ sieve verify audit.db
OK: Chain intact - 42 entries verified.

# After tampering with a row:
$ sieve verify audit.db
TAMPER DETECTED: Hash mismatch at id=7: stored 'a1b2c3...' != computed 'deadbe...'
```

Reproduce the tamper proof locally with one command:

```bash
make tamper-demo
```

If you installed Sieve into a non-default interpreter, run
`PYTHON=/path/to/python make tamper-demo` so the target uses the same environment.

Expected ending:

```text
[3/3] Verifying the audit log...
    TAMPER DETECTED: Hash mismatch at id=1: ...
```

## Error handling

| Exception | When raised |
|-----------|-------------|
| `sieve.PolicyViolation` | Tool call denied by policy or operator |
| `sieve.ApprovalDenied` | `require_approval` rule fired but operator said no |
| `sieve.core.errors.PolicyLoadError` | Policy YAML is missing, malformed, or fails validation |
| `RuntimeError` | `sieve.configure()` was not called before a guarded tool ran |

---

## Architecture and security notes

Read [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) for the interception flow,
what is logged, the hash-chain design, and the threat model.

Before publishing a release, run:

```bash
python -m pytest --cov=sieve --cov-report=term-missing
python -m pip_audit
```

## Custom approval handler

The CLI prompt is the default. For headless agents, CI, or tests, use a
non-interactive handler:

```python
import sieve
from sieve.approval import StaticApprovalHandler

sieve.configure(
    policy_path="policy.yaml",
    approval_handler=StaticApprovalHandler(approve=False),  # auto-deny
)
```

You can also make the terminal prompt auto-deny or auto-approve after a timeout:

```python
from sieve.approval import CLIApprovalHandler

sieve.configure(
    policy_path="policy.yaml",
    approval_handler=CLIApprovalHandler(timeout_seconds=30, timeout_default=False),
)
```

Implement `ApprovalHandler` to connect Sieve to another approval system:

```python
from sieve.approval.base import ApprovalHandler
from sieve.core.decision import ToolCall

class QueueApprovalHandler:
    def request(self, call: ToolCall) -> bool:
        # Send the request to your approval system and return True or False.
        ...

sieve.configure(policy_path="policy.yaml", approval_handler=QueueApprovalHandler())
```

## Cost tracking

Attach `SieveCostCallback` to LangChain LLMs or agents and provide prices for the
models you want enforced. Unknown model names are tracked as token usage but do
not count toward dollar limits.

```python
from sieve.integrations.langchain import SieveCostCallback

cost_callback = SieveCostCallback(
    {
        "example-model": {
            "input_per_1k": 0.01,
            "output_per_1k": 0.03,
        }
    }
)
```

---

## Project structure

```
src/sieve/
├── __init__.py           # public API: configure, guard, PolicyViolation, ApprovalDenied
├── config.py             # SieveConfig, configure(), get_interceptor()
├── decorator.py          # @sieve.guard() for plain sync/async functions
├── cli.py                # sieve verify / sieve tail
├── core/
│   ├── decision.py       # ToolCall, Outcome, Decision dataclasses
│   ├── errors.py         # PolicyViolation, ApprovalDenied, PolicyLoadError
│   └── interceptor.py    # evaluate, approve, audit, execute pipeline
├── policy/
│   ├── models.py         # Rule, Match, ArgMatch, Policy
│   ├── loader.py         # load_policy(path), YAML parse and validation
│   └── engine.py         # PolicyEngine.evaluate(), fnmatch and regex matching
├── audit/
│   ├── models.py         # AuditEntry dataclass
│   └── log.py            # SQLite hash-chained audit log
├── approval/
│   ├── base.py           # ApprovalHandler protocol
│   └── cli.py            # interactive y/N approval handler, default deny
└── integrations/
    ├── langchain.py       # wrap_tool(BaseTool), lazy import
    └── mcp.py             # SieveMiddleware, lazy import fastmcp
```

---

## License

MIT

See [CONTRIBUTING.md](CONTRIBUTING.md) for local setup, tests, and PR expectations.
