Python API reference¶

Every public name you can import, with its real signature. Use this page to look things up. To learn how the pieces fit together, start with concepts and the guides linked from each section.

Everything in the first six sections is importable straight from interlock. Scopes, receipts, the sandbox, and the testing helpers live in subpackages, and each section says which one.

• Setup: install, get_engine, set_engine
• Guarding calls: guard, monitor, guard_mcp_session, guard_langchain_tools, enforce_tool_call, enforce_tool_result
• Rules and engines: Rule, PolicyEngine, deny_tool, deny_when
• Decisions: Verdict, Decision, Blocked
• Events: SensorEvent
• Run identity: span, new_span, current_span, current_principal, current_parent_principal, set_principal
• Scopes: interlock.policy
• Receipts: interlock.sink, interlock.receipt
• Sandbox: interlock.brace
• Testing helpers: interlock.testing

For the Node.js daemon and its Python client, see the sidecar reference.

Setup¶

Guide: concepts, writing rules.

install¶

install(engine=None, rules=None, sink=None) -> PolicyEngine

Sets the process-wide engine that every interceptor consults, and returns it.

Things to know:

• It is process-global. The last call wins.
• Passing rules together with an engine that has no add_rule method (a ScopedEngine, or the sidecar's RemoteEngine) raises TypeError, saying the engine takes no extra rules. Build those engines with their rules first, then call install(engine=...). For a ScopedEngine, bind extra rules at the root scope through a template.
• install() never removes a sink. Call set_sink(None) to turn receipts off.

import interlock
from interlock import deny_tool

engine = interlock.install(rules=[deny_tool("shell_exec")])
assert interlock.get_engine() is engine

get_engine¶

get_engine() -> PolicyEngine

Returns the installed engine. If nothing was ever installed, it creates an empty PolicyEngine, installs it, and returns it. An empty engine allows every call, so a program that forgets install() runs unguarded without any error.

set_engine¶

set_engine(engine) -> None

Installs an engine without touching rules or the sink. It accepts any object with an evaluate(event) -> Decision method and does no validation.

Guarding calls¶

Guides: integrations, MCP.

Every function in this section turns a call into a SensorEvent, asks the engine for a Decision, and enforces it. They all share the same three keyword options:

All of them raise Blocked when a call is denied under blocking enforcement. If a rule itself raises, that exception reaches the caller unchanged and the tool does not run (see PolicyEngine). When a receipt sink is installed, each decision is recorded, including allows. In monitor mode, a decision that would have blocked or rewritten the call is recorded with observed=True, so the log never reads as enforcement that didn't happen.

guard¶

guard(policy_id=None, enforcement="blocking", principal=None) -> Callable

Decorator for your own functions, sync or async. For each call it:

1. builds an event whose action is the function's __name__ and whose args are the call's arguments bound to the signature, defaults filled in;
2. evaluates it and raises Blocked before the body runs on a block;
3. on a modify, applies modified_args wherever each argument was passed: positionally, by keyword, or into **kwargs;
4. runs the function and passes its return value through enforce_tool_result, so result-phase rules apply too.

policy_id labels every decision on this tool that doesn't name a policy of its own, on the call side and the result side, so it shows up in Blocked, logs, and receipts. A rule's own policy_id always wins.

import interlock
from interlock import Blocked, Decision, guard

@guard(policy_id="shell.v1")
def run_shell(cmd: str) -> str:
    return cmd

interlock.install(rules=[lambda e: Decision.block("no shells")])
try:
    run_shell("ls")
except Blocked as e:
    assert e.decision.policy_id == "shell.v1"

import interlock
from interlock import Decision, guard

@guard()
def transfer(amount: int, to: str) -> str:
    return f"sent {amount} to {to}"

def cap(event):
    if event.args.get("amount", 0) > 100:
        return Decision.modify({"amount": 100}, reason="capped", attributed_to="amount")
    return None

interlock.install(rules=[cap])
assert transfer(5000, "bob") == "sent 100 to bob"

monitor¶

monitor(policy_id=None, principal=None) -> Callable

Shorthand for guard(policy_id, enforcement="monitor", principal). Decisions are logged to the "interlock" logger at DEBUG and written to the receipt sink if one is installed, with observed=True on any that would have intervened. Nothing is ever blocked or rewritten.

guard_mcp_session¶

guard_mcp_session(session, *, engine=None, enforcement="blocking", principal=None,
                  origin=None, schemas=None) -> Any

Replaces session.call_tool with a guarded version and returns the same session object. Works whether call_tool is sync or async, and never imports the MCP SDK: any object with a call_tool(name, arguments) method works.

Each call is checked with enforce_tool_call before it is sent, and the server's response is checked with enforce_tool_result before it is returned. An official SDK CallToolResult has its text joined into "result" first (see result_payload), so result rules can read it. Extra positional and keyword arguments to call_tool are passed through untouched.

Labels are facts about a call, so only your code may set them. Before the check, the wrapper drops any TRUST_LABELS key from the arguments the caller passed (those usually come from the model, and a model that wrote "__origin__": "github" could otherwise claim to be a trusted server). It then adds the labels from origin and schemas, and removes every label again before the call goes to the server. TRUST_LABELS is importable from interlock.interceptors.mcp: {"__origin__", "__schema__", "__attestation__", "__tools__"}.

import asyncio
from interlock import PolicyEngine, guard_mcp_session
from interlock.detectors.mcp_trust_registry import mcp_trust_registry

class Server:
    async def call_tool(self, name, arguments=None):
        return {"sent": arguments}

engine = PolicyEngine(rules=[mcp_trust_registry(["github"])])
github = guard_mcp_session(Server(), engine=engine, origin="github")
print(asyncio.run(github.call_tool("create_issue", {"title": "Bug", "__origin__": "evil"})))
# -> {'sent': {'title': 'Bug'}}     the spoofed label was dropped, and so was the real one

Only call_tool is wrapped. Connecting, listing tools, reading resources, and sampling are not intercepted. The MCP trust detectors that act on those events need you to send them yourself (see the MCP guide). mcp_capability_lease never refuses a tools/list listing event, so it can share an engine with mcp_surface_baseline.

guard_langchain_tools¶

guard_langchain_tools(tools, *, engine=None, enforcement="blocking", principal=None) -> List[Any]

Returns guarded copies of LangChain tools as StructuredTools with the same name, description, and argument schema. Each call is checked with enforce_tool_call, the original tool is invoked with the vetted arguments, and its output is checked with enforce_tool_result before it is returned. Result rules such as tool_result_injection_guard therefore apply to LangChain tools too.

Requires the [langchain] extra (langchain-core). Importing interlock does not import LangChain. Calling this function does.

enforce_tool_call¶

enforce_tool_call(action, arguments, *, engine=None, enforcement="blocking", principal=None) -> Dict[str, Any]

The core every adapter uses. Checks one call and returns the arguments to use:

• allow or monitor mode: a copy of arguments (None becomes {}).
• modify: that copy with modified_args merged in.
• block: raises Blocked.

The event gets its own copy of the arguments, so a rule that edits event.args in place can't change what the tool receives. Only a modify decision can.

enforce_tool_result¶

enforce_tool_result(action, result, *, engine=None, enforcement="blocking", principal=None) -> Any

Checks what a tool returned, as a phase="result" event whose args is result_payload(result): a copy of result if it is a dict, the joined text of an MCP CallToolResult, and {"result": result} for anything else.

• allow or monitor mode: returns the original result.
• modify: returns decision.modified_result. If that is None, the original is returned, so a rule cannot replace a result with None.
• block: raises Blocked. The tool has already run and its side effects stand. The caller just never sees the payload.

Rules and engines¶

Guide: writing rules.

Rule¶

Rule = Callable[[SensorEvent], Optional[Decision]]

A rule is any function that takes an event and returns a Decision, or None for "no opinion". Returning Decision.allow() also counts as no opinion: the engine keeps going.

Rules run on call-phase events by default. To make a rule see results instead, set an attribute on the function:

from interlock import Decision

def withhold_errors(event):
    if "Traceback" in str(event.args.get("result", "")):
        return Decision.block("internal error text withheld")
    return None

withhold_errors.phase = "result"   # now it only sees phase="result" events

A rule runs on exactly one phase. To check both, write two rules.

PolicyEngine¶

PolicyEngine(rules=None)

An ordered list of rules.

How opinions combine:

• A block from any rule wins and ends evaluation, even if an earlier rule rewrote the call.
• Rewrites compose. Each one is applied to a working copy of the event, and every later rule judges the rewritten call (or result). A later rewrite of the same argument wins.
• The result is one decision. A single rewrite comes back exactly as the rule returned it. Several are combined into one MODIFY with every change (modified_args merged, or the last modified_result), the reasons joined with "; ", and the first rewrite's policy_id and attributed_to.
• The caller's event is never changed.

from interlock import Decision, PolicyEngine, SensorEvent

engine = PolicyEngine(rules=[
    lambda e: Decision.modify({"limit": 100}, reason="capped", policy_id="p.cap"),
    lambda e: Decision.modify({"dry_run": True}, reason="dry run", policy_id="p.dry"),
])
d = engine.evaluate(SensorEvent(action="query", args={"limit": 5000}))
assert d.modified_args == {"limit": 100, "dry_run": True}
assert (d.reason, d.policy_id) == ("capped; dry run", "p.cap")

evaluate does not catch exceptions. If a rule raises, the exception propagates through the interceptor and the tool does not run. Only deny_when converts a crashing predicate into a clean block. (The sidecar is different: it turns a raising rule into a block.)

deny_tool¶

deny_tool(name, reason="tool not permitted", policy_id=None) -> Rule

Blocks every call whose action equals name exactly. Call phase only.

deny_when¶

deny_when(predicate, reason, policy_id=None) -> Rule

Blocks when predicate(event) is truthy. If the predicate raises, the call is blocked with the reason "rule error, denied fail-closed: <reason> (<error>)" instead of the exception escaping.

Decisions¶

Guide: concepts.

Verdict¶

An IntEnum: ALLOW = 0, BLOCK = 1, MODIFY = 2. The integers are stable and are what the wire protocol sends.

Decision¶

A dataclass describing what to do with one event.

Build decisions with the class methods:

Blocked¶

Blocked(decision, action)

The exception raised when a call or result is denied. It has two attributes, decision (the full Decision) and action (the tool name). Its message is blocked action '<action>': <reason>.

On the call side, the tool never ran. On the result side, it already ran and only its output was withheld.

Events¶

Guide: concepts.

SensorEvent¶

A dataclass describing one action as a rule sees it.

You only construct events yourself when testing a rule directly:

from interlock import SensorEvent, deny_tool

rule = deny_tool("shell_exec")
assert rule(SensorEvent(action="shell_exec", args={"cmd": "ls"})).reason == "tool not permitted"
assert rule(SensorEvent(action="read_file", args={})) is None

result_payload¶

from interlock.event import result_payload

result_payload(result) -> Dict[str, Any]

Shapes a tool's return value into the args of a result-phase event. Every result path uses it.

from dataclasses import dataclass
from interlock.event import result_payload

@dataclass
class Text:
    text: str

@dataclass
class CallToolResult:
    content: list
    structuredContent: dict = None
    isError: bool = False

r = CallToolResult([Text("first"), Text("second")], {"temperature": 21})
assert result_payload(r) == {"result": "first\nsecond", "temperature": 21, "is_error": False}

The view is only what rules see. A result rewrite still replaces the whole value the caller gets, so a rule that rewrites an MCP result hands back whatever it put in modified_result.

Run identity¶

Guide: concepts.

These use contextvars, so values follow await and are separate per thread and per asyncio task.

span¶

span(principal=None, parent_principal=None)   # context manager, yields the span id

Starts a new span id for the block and, if given, sets the principal and parent principal. Everything is restored when the block exits, including on error. Spans nest.

from interlock import current_principal, span

with span(principal="acme/agent42") as span_id:
    assert current_principal() == "acme/agent42"
assert current_principal() is None

new_span¶

new_span(principal=None, parent_principal=None) -> str

Like span, but sets the values for the rest of the current context and never restores them. Returns the new span id.

current_span, current_principal, current_parent_principal¶

current_span() -> Optional[str]
current_principal() -> Optional[str]
current_parent_principal() -> Optional[str]

Read the current values. None when unset.

set_principal¶

set_principal(principal, parent_principal=None) -> None

Sets the principal for the rest of the current context. The parent principal changes only if you pass one, so set_principal("x") keeps an inherited delegation link. Use span when you want the old values restored.

Scopes (interlock.policy)¶

Guide: scopes.

from interlock.policy import Binding, ScopedEngine, ScopeRegistry, ScopeTree, Template
from interlock.policy.scopes import DelegationOverGrant, DelegationDepthExceeded, UnknownTemplate

A scope is a tuple path such as ("root", "acme", "agent42"). interlock.policy also re-exports PolicyEngine, Rule, deny_tool, and deny_when.

Template¶

Template(template_id, factory, union_params=frozenset())

A frozen dataclass: a named rule factory. factory(**params) must return a Rule. Parameters named in union_params accumulate down the tree. All others are overridden by deeper bindings.

Binding¶

Binding(template_id, params={}, enabled=True, delegated_from=None)

A frozen dataclass attaching a template, with parameters, to one scope node.

ScopeTree and PyScopeTree¶

ScopeTree()
PyScopeTree()

The store of bindings. ScopeTree is the native Rust tree when the optional rust/ module is installed, and PyScopeTree otherwise. PyScopeTree is always available.

SCOPE_BACKEND¶

"rust" or "python", depending on which tree ScopeTree is.

Union parameters always come back as a set, whichever tree resolved them. A non-union parameter keeps whatever type you passed.

ScopeRegistry¶

ScopeRegistry(tree, templates, *, maxsize=10000, max_delegation_depth=16)

Resolves scopes and caches one compiled PolicyEngine per scope.

ScopedEngine¶

ScopedEngine(registry, scope_of=None, *, default_scope=(), allow_unscoped=False)

An engine that picks a scope for each event and delegates to that scope's compiled engine. Install it with interlock.install(engine=ScopedEngine(registry)).

An event with no scope is blocked by default. With no principal and no default_scope, evaluate returns a block with the reason no scope for this event (no principal and no default_scope); denied fail-closed and the policy id scoped_engine.unscoped (UNSCOPED_POLICY_ID in interlock.policy.scopes). Set default_scope to a baseline node to give such callers a floor policy instead, or pass allow_unscoped=True if they really should run unchecked.

principal_scope¶

principal_scope(event) -> Tuple[str, ...]

Splits event.principal on /, dropping empty parts: "root/acme/agent42" becomes ("root", "acme", "agent42"). No principal gives ().

delegation_chain¶

delegation_chain(event) -> Tuple[Tuple[str, ...], ...]

The scopes an event acts through, delegator first: (parent_scope, own_scope) when parent_principal is set, otherwise just (own_scope,). Empty scopes are dropped.

Delegation errors¶

Import these from interlock.policy.scopes. They are not re-exported from interlock.policy.

Both are raised from resolve, compile, and for_scope, so a bad grant never reaches the cache. So is UnknownTemplate (a LookupError), for a binding whose template id isn't registered: a binding that silently vanished would be a rule that never exists. ScopedEngine.evaluate catches all three and returns a BLOCK with policy id scoped_engine.resolution, so a call never sees them as raw exceptions.

Receipts (interlock.sink and interlock.receipt)¶

Guide: audit log.

from interlock.sink import FileSink, InMemorySink, verify_file
from interlock.receipt import set_sink

FileSink¶

FileSink(path, *, signer=hmac_sha256, key=None)

Appends one JSON line per receipt to path, chained and signed. key is the signing key in bytes. Without one, a random per-process key is used and only this process can verify the file.

A reopened file continues its chain. When path already holds receipts (after a restart, say), the new FileSink reads the last one and links the next receipt to it, so one file can span restarts. Keep the same key, or the older receipts won't verify under the new one. If the file's last line isn't a readable receipt (a crash mid-write, a foreign file), the constructor raises ValueError instead of appending to damage nobody saw. Move the file aside to start a new log.

Two FileSinks writing the same file at the same time still fork the chain, and verification reports the fork as receipt N: chain break (...). One writer per file at a time.

InMemorySink¶

InMemorySink(*, signer=hmac_sha256, key=None)

Keeps receipts in the receipts list attribute. Useful in tests.

ChainedSink¶

ChainedSink(*, signer=hmac_sha256, key=None)

The base class. Subclass it and implement _append(receipt) and _read_back() to store receipts somewhere else. Any object with a write(receipt) method can be installed as a sink, but only ChainedSink subclasses chain and sign.

Deleting receipts from the end of a log leaves a valid, shorter chain that the linkage alone can't catch. Comparing against a recorded head does:

import interlock, os, tempfile
from interlock import enforce_tool_call
from interlock.receipt import set_sink
from interlock.sink import FileSink, verify_file

key = b"k" * 32
path = os.path.join(tempfile.mkdtemp(), "audit.jsonl")
sink = FileSink(path, key=key)
interlock.install(rules=[], sink=sink)
for _ in range(3):
    enforce_tool_call("read_rows", {"table": "orders"})
head = sink.head
set_sink(None)

with open(path) as fh:
    lines = fh.readlines()
with open(path, "w") as fh:
    fh.writelines(lines[:2])          # someone deletes the last receipt

print(verify_file(path, key=key))             # -> None: the linkage looks fine
print(verify_file(path, key=key, head=head))
# -> chain ends at receipt 1 but the expected head is ...: receipts were cut from the end, or this is not the log that head came from

Receipt¶

A dataclass. One per enforced decision.

to_json() and Receipt.from_json(line) convert to and from the canonical one-line form. fields(include_mac=True) returns that canonical shape as a dict. observed appears in it (and in the signed bytes) only when it is True, so receipts written before the field existed still verify.

Functions and the signer seam¶

Without a verifier, verification recomputes each tag with signer and key, so the verifier needs the signing key. That's the right shape for the default HMAC. For a public-key scheme, sign with the private key and hand auditors a verifier that checks with the public key. Here it is with Ed25519 from the cryptography package (not an interlock dependency):

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
from interlock.sink import FileSink, verify_file

def ed25519_sign(private_bytes, message):            # the writer's Signer
    return Ed25519PrivateKey.from_private_bytes(private_bytes).sign(message)

sink = FileSink("audit.jsonl", signer=ed25519_sign, key=private_bytes)

public_key = Ed25519PublicKey.from_public_bytes(public_bytes)

def ed25519_verify(message, signature):              # the auditor's Verifier
    try:
        public_key.verify(signature, message)
        return True
    except InvalidSignature:
        return False

verify_file("audit.jsonl", verifier=ed25519_verify)   # -> None when intact

Sandbox (interlock.brace)¶

Guide: sandbox.

from interlock.brace import (
    Sandbox, SandboxProfile, compile_profile, available_backends,
    IsolationUnavailable, LimitUnsupported, KNOWN_TOKENS,
)

Sandbox, SandboxProfile, and compile_profile are also importable from interlock.

Sandbox¶

Sandbox(profile=None, *, backend=None, engine=None, enforcement="blocking",
        principal=None, require=())

The constructor raises:

• ValueError("unknown backend: ...") for a backend name that doesn't exist (including "nftables"), and RuntimeError("backend not available on this host: ...") for one this machine can't run;
• ValueError for a require token that isn't in KNOWN_TOKENS, so a typo fails here instead of refusing every run;
• LimitUnsupported when the profile asks for a resource limit this host can't enforce.

backend_name tells you which backend was picked.

import sys
from interlock.brace import IsolationUnavailable, Sandbox

try:
    Sandbox(backend="soft", require=("net",)).run([sys.executable, "-c", "print('hi')"])
except IsolationUnavailable as e:
    print(e)          # nothing ran
# -> refusing to run: the soft backend cannot enforce ['net'] for this profile (it enforces ['env'])

The launch check is an ordinary call event with action="process_spawn" and these args: path (argv[0]), argv, allow_network, allowed_hosts, and write_paths. A rule can block it (run raises Blocked) or return a modify with a new argv list, which is what runs instead. An empty argv raises ValueError. A timeout that expires raises subprocess.TimeoutExpired.

SandboxProfile¶

A frozen dataclass.

On Windows, any resource limit raises LimitUnsupported: there are no POSIX limits to apply. A requested limit is never dropped silently.

SandboxProfile.locked() returns the fully closed default. profile.with_paths(read=(), write=()) returns a copy with extra paths added.

compile_profile¶

compile_profile(resolved, *, workdir=None, base_env=None) -> SandboxProfile

Builds a profile from the {template_id: params} dict that ScopeRegistry.resolve returns. It reads three bindings and ignores the rest:

base_env becomes the child's environment, and workdir its working directory. resolved=None or {} gives a locked profile.

SandboxResult¶

available_backends and select_backend¶

available_backends() -> List[str]
select_backend(prefer=None) -> Backend

available_backends() lists what this machine can run, strongest first. soft is always last and always present. select_backend returns a backend instance, raising the same errors as Sandbox(backend=...). You rarely need it directly.

IsolationUnavailable¶

A RuntimeError raised by Sandbox.run() when a required token isn't among the guarantees the backend would enforce. Nothing was started. Attributes: backend (the backend's name), missing (the required tokens it lacks), and enforced (what it does enforce).

LimitUnsupported¶

A RuntimeError raised when a profile asks for a resource limit the host can't enforce: memory_mb on macOS, or any limit on Windows. It's raised when the Sandbox is built and again by run() (the profile may have changed). The message names the limit and the platform.

The check itself is interlock.brace.limits.check_limits(profile, platform=None). interlock.brace.limits.rlimit_preexec(cpu_seconds, memory_mb) builds the preexec_fn every backend uses to apply the limits in the child.

KNOWN_TOKENS¶

The frozenset of every token a backend can report: {"env", "rlimit", "fs-ro", "fs-rw", "pid", "net"}. require is checked against it.

Testing helpers (interlock.testing)¶

Guide: Node.js client.

The package's top level exports nothing. Its modules exist so that clients in other languages, and your own test suites, can check behaviour against a real daemon. Most users never import them.

import os, tempfile
from interlock import PolicyEngine, SensorEvent, deny_tool
from interlock.sidecar import RemoteEngine
from interlock.testing.harness import serve

path = os.path.join(tempfile.mkdtemp(), "guard.sock")
with serve(PolicyEngine(rules=[deny_tool("shell_exec")]), path):
    with RemoteEngine(path, timeout=5) as remote:
        decision = remote.evaluate(SensorEvent(action="shell_exec", args={}))
        assert decision.verdict.name == "BLOCK"