sieve-layer 0.1.0

Drop-in middleware that intercepts every tool call your AI agents make, blocks unsafe actions, and logs a tamper-proof audit trail.

Sieve

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

---

Quickstart

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:

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

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

2. Copy the example policy

cp examples/policy.yaml policy.yaml

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

# 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

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

@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

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

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

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

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

$ 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:

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:

[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 for the interception flow, what is logged, the hash-chain design, and the threat model.

Before publishing a release, run:

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:

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:

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:

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.

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 for local setup, tests, and PR expectations.