API reference

Everything re-exported from interbolt (the package __init__.py). This is the public surface; anything not listed here is internal and may change without notice.

taint, endorse, guard, check, enforce_decision, enforce_decision_sync, configure, default_policy, agent, get_runtime, AgentHandle, RunContext,
track_model_call, Runtime, Policy,
Decision, Event, Finding, Endorsement, Action, Mode, Label, RunIngressEntry, TrustLevel, Capability,
Reporter, ApprovalResolver,
NullReporter, InMemoryReporter, LoggingReporter, JsonlReporter, CompositeReporter, OTelReporter,
describe_decision, describe_event, describe_finding, describe_endorsement,
RECORD_TYPE_EVENT, RECORD_TYPE_FINDING, RECORD_TYPE_ENDORSEMENT,
InterboltError, PolicyViolation, PolicyEvaluationError, ApprovalDenied,
InterboltConfigError, InterboltUsageError,
Tainted, LabeledValue, TaintedBytes,
__version__

taint

def taint(value: Any, *, source: str, derived_from: Iterable[Any] | None = None) -> Any: ...

Marks value with a Label recording source. For str returns a Tainted; for bytes returns a TaintedBytes; for a builtin container (list, tuple, set, frozenset, a Mapping's keys and values) recurses and labels string leaves to the bounded recursion depth; for any other scalar returns a LabeledValue. The label only records the source name; trust is resolved later, at the sink. Needs no configured runtime. See Taint propagation.

If derived_from is given a non-empty iterable, it marks value as derived from those values instead of as a fresh ingress point. source becomes the name of the derivation hop (for example "model"), and the returned label's lineage is the union of every label found among derived_from, so trust resolves at the sink exactly as if those original inputs had reached it directly. If no label is found among derived_from at all, value is returned completely unmarked, and no run-level ingress event is recorded for source in this case. See Taint propagation: model calls and derived values.

endorse

def endorse(value: Any, *, kind: str, note: str | None = None) -> Any: ...

Adds kind to the label's endorsements after an explicit validation step, without touching lineage or how t.trust resolves. Accepts the same shapes as taint(), and a value with no label anywhere in it passes through unchanged. note is free text carried only on the emitted Endorsement record, never on the label. Every call that endorses something emits that record through the configured reporter, or logs at INFO if no runtime is configured yet; endorse() itself needs no runtime. Call it only from deterministic code after a real check, never conditioned on model output. See Auditing: endorsement and Policies: endorsement-aware rules.

track_model_call

@track_model_call                      # source defaults to "model"
def summarize(prompt: str) -> str: ...

@track_model_call(source="gpt-4")      # explicit derivation-hop name
async def summarize(prompt: str) -> str: ...

Wraps a function so its return value is tainted via taint(result, source=source, derived_from=<the function's bound call arguments>). Auto-detects sync vs async the same way guard does. Tracks provenance only; does not evaluate policy, so stack @guard/@handle.guard alongside it if the call into the model should also be gated. Needs no configured runtime, the same as taint() itself. See Taint propagation: model calls and derived values.

guard

@guard                      # tool name defaults to the function name
def send_email(...): ...

@guard(tool="fs.write")     # explicit qualified or bare tool name
def write_file(...): ...

Decorates a function or coroutine function so every call is checked against the current policy before it runs, auto-detecting sync vs async. Internally this is check() followed by enforce_decision(): on block raises PolicyViolation, on require_approval invokes the configured ApprovalResolver, on allow calls through. agent_id and run_id come from the active agent_context, and session_id is never set through this path. See Identity.

check

def check(
    *,
    tool: str,
    args: Mapping[str, Any],
    agent_id: str,
    run_id: str | None = None,
    session_id: str | None = None,
) -> Decision: ...

The framework-agnostic decision core; guard is sugar over check() followed by enforce_decision(). check() only decides and emits; it never raises on a block or require_approval outcome (it raises only PolicyEvaluationError, fail-closed under enforce). Pair it with enforce_decision() (below) to act on the result. Collects labels from args (recursing into containers), evaluates the policy, returns a Decision, and emits the corresponding Event through the configured reporter.

agent_id is explicit here rather than read from the context variable. run_id resolves from the active agent_context when you pass None, so a dispatch loop inside a run shares that run's id, and its run.tainted gating, automatically; a fresh id is minted only when no run is active. Pass run_id= explicitly only to override the ambient value, or to set a specific id for a call made outside any context. Requires configure() to have run; raises InterboltUsageError otherwise. Use this for custom dispatch loops or existing tool registries; see Identity: custom dispatch loops.

agent_id must match ^[A-Za-z0-9_.-]+$, may not be a Tainted/TaintedBytes/LabeledValue, and may not be the literal "default" (reserved for the implicit no-context fallback), or it raises InterboltConfigError. See Identity as a policy input.

Policy testing is just check() invoked with synthetic args and taint, asserted against the returned Decision; there's no separate simulate function. See Testing.

enforce_decision, enforce_decision_sync

async def enforce_decision(decision: Decision) -> None: ...
def enforce_decision_sync(decision: Decision) -> None: ...

The enforcement half of guard, exposed for dispatch loops that call check() directly. Turns a Decision into control flow purely by its action: allow returns, block raises PolicyViolation, require_approval invokes the current runtime's ApprovalResolver and raises ApprovalDenied if it returns falsy. It reads decision.action, which check() has already resolved through the mode, so mode handling lives in one place rather than here: under dry_run the decision arrives as allow and enforces nothing, and under monitor a real block still enforces exactly as it does under enforce.

Only the require_approval path touches the runtime (to reach the resolver); allow and block need no configured runtime. The sync form raises InterboltUsageError if the resolver returns an awaitable, closing that coroutine first so it does not leak; the async form awaits it. Both re-raise PolicyViolation/ApprovalDenied with the offending Decision on .decision.

A custom dispatch loop is then three lines:

# 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="fs.write", args=args, agent_id=agent_id)
await enforce_decision(decision)     # no-op on allow; raises on block / denied approval
result = run_tool(...)

See Identity: custom dispatch loops.

configure

def configure(
    *,
    policy: Policy | None = None,
    reporter: Reporter | None = None,
    approval_resolver: ApprovalResolver = auto_deny,
    mode: Mode | str = Mode.ENFORCE,
    audit: bool = False,
) -> Runtime: ...

Builds a Runtime, installs it as the process-current runtime, and returns it. Has no import-time side effects: a module decorated with @guard can be imported before configure() has run. policy defaults to None, which loads the built-in default policy (default_policy(), below): no sources, no sinks, every guarded call falls through to require_approval. reporter defaults to a fresh NullReporter(). approval_resolver defaults to auto_deny (denies every approval request). INTERBOLT_MODE and INTERBOLT_AUDIT override mode and audit; an invalid mode raises InterboltConfigError. See Policies: modes for the full precedence chain.

Logging on each call, independent of any configured Reporter: one INFO-level summary line (effective mode, policy source, source/sink counts, caller file:line), plus a WARNING if no policy was passed, a WARNING if the policy's fail_mode changed the effective mode away from the mode= argument, and a WARNING if INTERBOLT_MODE changed the effective mode. See Policies.

default_policy

def default_policy() -> Policy: ...

Returns the built-in default policy for programmatic use and testing: the same posture configure(policy=None) uses, exposed directly rather than only implicitly.

agent

def agent(agent_id: str) -> AgentHandle: ...

Returns a durable per-agent handle whose .guard decorates with this agent_id, resolving the runtime lazily at call time. Captures agent_id eagerly, so it is safe at import time and needs no Runtime instance, and it carries identity across a thread-pool boundary where agent_context cannot. runtime.agent(agent_id) is equivalent. agent_id is validated eagerly at construction (see check, above, for the exact rule); this means agent(...) itself can raise InterboltConfigError, not only the first guarded call through the returned handle. See Identity: when to use an AgentHandle.

get_runtime

def get_runtime() -> Runtime: ...

Returns the process-current runtime, for code that did not keep configure()'s return value (a different module, or reaching for Runtime.add_reporter later). This is the same pattern as OpenTelemetry's get_tracer_provider(). Raises InterboltUsageError if configure() has not run yet.

AgentHandle

The type returned by agent(...)/runtime.agent(...). Exposes .guard, usable bare (@handle.guard) or parameterized (@handle.guard(tool=...)), behaving identically to the module-level guard except that agent_id comes from the handle instead of the active agent_context. Also exposes .track_model_call, equivalent to the module-level track_model_call (above), provided for per-handle symmetry: @support.guard and @support.track_model_call are both reachable from one handle. The handle's agent identity plays no role in .track_model_call, since taint derivation is identity-free.

RunContext

class RunContext:  # frozen, slots
    run_id: str
    agent_id: str

The value agent_context/agent_context_sync yield. A frozen, data-only snapshot of the run_id and agent_id bound for the block. It stays readable after the block exits and remains a correct record of what that run was, but the run's registries are cleared at exit and re-binding from it is not supported. run_id is the join key on every Decision, Event, Finding, and Endorsement emitted inside the block. See Identity.

Runtime

The composition root returned by configure(). One per process.

Attributes:

• runtime.policy -> Policy: the compiled policy this runtime enforces.
• runtime.mode -> Mode: the effective mode, after the precedence chain in configure().
• runtime.approval_resolver -> ApprovalResolver: the resolver invoked on a require_approval decision.
• runtime.reporter -> Reporter: the CompositeReporter every record is emitted through (read-only; every Runtime holds one internally, even for a single reporter= passed to configure()). See Reporters.

Methods:

• runtime.agent(agent_id: str) -> AgentHandle: equivalent to the module-level agent(...), above.
• runtime.agent_context(agent_id: str): an async context manager that binds agent_id and mints a run_id for the duration of the block, yielding both as a RunContext (the as run clause is optional). Exiting clears both, along with that run's ingress registry and audit findings. agent_id is validated before anything binds (same rule as check, above); a rejected value raises InterboltConfigError with no partial bind and nothing to clean up. See Identity.
• runtime.agent_context_sync(agent_id: str): the synchronous counterpart; identical binding, validation, cleanup, and RunContext yield, for a call site that cannot use async with.
• runtime.check(*, tool, args, agent_id, run_id=None, session_id=None) -> Decision: the same decision core as the module-level check, against this runtime explicitly. Validates agent_id's charset and rejects a taint carrier unconditionally, but (unlike the module-level check) does not reject the literal "default", since this method is also the path bare guard's no-context fallback resolves through.
• runtime.add_reporter(reporter: Reporter) -> None: attaches an additional reporter to this live runtime without reconfiguring, modeled on OpenTelemetry's add_span_processor. The same non-blocking contract as any other reporter applies; there is no removal, only reconfiguring. See Reporters.
• runtime.audit_findings() -> list[Finding]: the findings recorded so far by the laundering audit, the observer that catches an untrusted value reaching a sink argument with no label (see Auditing), bounded with oldest evicted first once the cap is reached, or [] if audit was not enabled.

Policy

Policy.from_file(path: str) -> Policy
Policy.validate(path: str) -> list[str]

from_file loads, validates, and compiles a policy YAML file in one call; raises PolicyEvaluationError if the file is missing, malformed, fails schema validation, or a rule's CEL expression fails to parse, and InterboltConfigError if a rule's CEL expression uses a disallowed construct, such as .any( in place of exists. validate performs schema and CEL checks only, without executing an agent, and returns a list of human-readable problem descriptions, empty if the policy is valid, capturing every error there instead of raising. policy.document exposes the validated PolicyDocument, policy.sources_table the declared source-to-trust mapping, policy.source the path it was loaded from (None for a programmatically constructed policy), and policy.fingerprint a stable "sha256:..." hash of the normalized document, computed once at construction and stamped onto every Event/Finding/Endorsement this policy produces. See Policies, Policy evaluation internals: the fingerprint, and CI.

Decision

class Decision(BaseModel, frozen=True):
    action: Action                       # ALLOW | BLOCK | REQUIRE_APPROVAL
    matched_rule: str | None             # name of the first matching rule
    matched_condition: str | None        # the matched rule's CEL `when` text
    tool: str                            # qualified name
    contributing_labels: tuple[Label, ...]
    trifecta: frozenset[str]             # this call's legs: from_untrusted + declared capabilities
    untrusted_sources: frozenset[str]    # which contributing source names resolved untrusted
    run_tainted: bool                    # run-level gating
    run_ingress: tuple[RunIngressEntry, ...]  # the structured record behind run_tainted
    run_trifecta: frozenset[str]         # every leg satisfied at run scope; empty outside agent_context
    mode: Mode
    decision_id: str
    agent_id: str
    run_id: str
    session_id: str | None

Returned by check/guard, attached to PolicyViolation/ApprovalDenied on .decision, and embedded in the emitted Event. matched_condition is None for the sink's catch-all rule, when nothing matched, and after an evaluation error, which also clears matched_rule. Otherwise it is human-readable CEL: for a when: rule, the text as written in the policy YAML, and for a require_endorsement: rule, the equivalent expression that shorthand compiles to.

Event, Finding, Endorsement

Every emitted record shares four fields:

schema_version: int
trace_id: str | None      # active OpenTelemetry trace id (W3C hex), or None
span_id: str | None       # active OpenTelemetry span id, or None
timestamp: datetime

trace_id/span_id are populated when an OpenTelemetry span is active at construction, and are None otherwise, including when OpenTelemetry is not installed. Finding and Endorsement additionally carry the identity triple (agent_id, run_id, session_id) directly. Event does not, since its embedded Decision already carries it.

class Event(RecordBase, frozen=True):
    decision: Decision
    sources: frozenset[str]   # every source contributing to the call, trusted or not
    outcome: Outcome          # the real, pre-dry_run-downgrade result

class Finding(IdentifiedRecordBase, frozen=True):
    source: str               # the untrusted source whose content reached the sink
    tool: str
    argument: str             # the argument name the content was found in

class Endorsement(IdentifiedRecordBase, frozen=True):
    kind: str
    note: str | None
    lineage: tuple[str, ...]  # the endorsed label's lineage
    value_id: str             # the fresh label id minted for this endorsement hop

Event deliberately does not duplicate the decision's fields. Reach matched_rule, trifecta, untrusted_sources, run_tainted, run_ingress, mode, and the identity triple through event.decision, the single source of truth for what was decided.

outcome is an Outcome enum with four values: allow, block, require_approval, and evaluation_error. It records what check() actually computed before any mode-based downgrade, which is what makes a dry_run rollout informative. Outcome lives in interbolt.models.core and is not currently re-exported at the top level.

Finding is the laundering-audit record, emitted when untrusted content reaches a sink argument without a label; see Auditing. Endorsement is the audited record of one endorse() call; see Auditing: endorsement. All three travel the same Reporter seam, and EVENT_SCHEMA_VERSION (in interbolt.constants) versions all three together. See Events for the wire shape and the version history.

Action, Mode, TrustLevel

All enum.StrEnum, so the same value round-trips a policy YAML string, an environment variable string, and serialized output with no conversion code.

• Action: ALLOW, BLOCK, REQUIRE_APPROVAL.
• Mode: ENFORCE, MONITOR, DRY_RUN. See Policies.
• TrustLevel: TRUSTED, UNTRUSTED. The result of resolving a source name against the policy at the sink; never stored on a Label.
• Capability: READS_PRIVATE ("reads_private"), REACHES_EXTERNAL ("reaches_external"). The closed set of two tool capabilities, declared with the capabilities: key on a sink entry and read back from Policy.tool_capabilities: the declared tool-to-capabilities mapping, built from the capabilities: key on each sink entry. A tool whose entry omits the key is absent from the mapping and resolves to the empty set rather than raising. See Policies: declaring what a tool does.

Label

class Label(BaseModel, frozen=True):
    source: str
    value_id: str
    lineage: tuple[str, ...]
    ingested_by: tuple[str, ...] = ()
    endorsements: tuple[str, ...] = ()

See Taint propagation and Auditing: endorsement for endorsements.

RunIngressEntry

class RunIngressEntry(BaseModel, frozen=True):
    source: str
    trust: TrustLevel
    ingested_by: tuple[str, ...]

The per-source entry in Decision.run_ingress: the source name, its trust resolved at decision time against the policy's sources table, and the agent ids that called taint() with it. See Events: run_ingress and Policies: the CEL evaluation context.

Tainted, TaintedBytes, LabeledValue

Tainted (a str subclass) and TaintedBytes (a bytes subclass) carry a .label: Label and propagate it through the operation subset described in Taint propagation. LabeledValue wraps a non-string scalar, exposing .value and .label. All three preserve the label under copy.copy/copy.deepcopy and drop it under pickling; see Serialization for the one channel that does carry it across.

pack, unpack, pack_into, unpack_from

def pack(
    value: Any,
    *,
    key: bytes | str | None = None,
    key_id: str | None = None,
    include_run: bool = True,
) -> dict[str, Any]: ...

def unpack(
    envelope: Mapping[str, Any],
    *,
    key: bytes | str | None = None,
) -> Any: ...

def pack_into(
    mapping: Mapping[str, Any],
    *,
    key: bytes | str | None = None,
    key_id: str | None = None,
    include_run: bool = True,
) -> dict[str, Any]: ...

def unpack_from(
    mapping: Mapping[str, Any],
    *,
    key: bytes | str | None = None,
) -> dict[str, Any]: ...

pack strips every Tainted/TaintedBytes/LabeledValue carrier in value down to a plain JSON-representable payload and records what it stripped in a path-keyed sidecar, returning one plain envelope dict. unpack reverses it: validates the envelope, verifies mac when key is given, rebuilds every carrier, and replays the run's ingested source names into the currently active run. pack_into/unpack_from are sugar for the dominant case, a top-level state mapping that must keep its shape, adding or removing one reserved key, constants.WIRE_ENVELOPE_KEY ("__interbolt__"). Both raise InterboltConfigError on every rejection. See Serialization for the wire format, the security model, and integration recipes.

The envelope's run block records each source together with the agent ids that ingested it, and the wire schema version is 2. A version 1 envelope is rejected, since the block changed shape and no compatibility path is kept.

OTelReporter

class OTelReporter:
    def __init__(self) -> None: ...

Maps Event/Finding/Endorsement onto OpenTelemetry spans at the edge; requires the interbolt[otel] extra. Exported lazily via a module __getattr__ in interbolt/__init__.py (PEP 562): import interbolt never requires opentelemetry to be installed, and from interbolt import OTelReporter raises InterboltConfigError with an install hint if the extra is missing. See Reporters: OTelReporter and the OTel guide.

Reporter, ApprovalResolver

Protocols in interbolt.models.protocols. Reporter.export takes an Event | Finding | Endorsement. See Reporters for Reporter and its six implementations (NullReporter, InMemoryReporter, LoggingReporter, JsonlReporter, CompositeReporter, OTelReporter). ApprovalResolver is Callable[[Decision], bool | Awaitable[bool]]: invoked synchronously at a sync call site, awaited at an async call site. A sync call site needs a resolver that returns a plain bool; one that returns an awaitable raises InterboltUsageError. The default, auto_deny, denies every request.

describe_decision, describe_event, describe_finding, describe_endorsement

def describe_decision(decision: Decision) -> str: ...
def describe_event(event: Event) -> str: ...
def describe_finding(finding: Finding) -> str: ...
def describe_endorsement(endorsement: Endorsement) -> str: ...

Each turns its record into a one-line, rich-markup-tagged human summary, meant for a rich.console.Console, not a bare print() (the raw [tag]...[/tag] markup otherwise prints literally). describe_decision is the one to reach for right where a Decision is already in hand (a caught PolicyViolation/ApprovalDenied, or check()'s direct return value): it shows the tool, action, matched rule, matched condition (if any), mode, and untrusted_sources without a trip through the reporter stream. describe_event/describe_finding/describe_endorsement cover the same ground for the emitted, versioned records, and are what interbolt inspect uses internally. See Reporters.

RECORD_TYPE_EVENT, RECORD_TYPE_FINDING, RECORD_TYPE_ENDORSEMENT

The "record_type" string values JsonlReporter tags each line with, and interbolt inspect reads back to recover which model (Event, Finding, or Endorsement) a line deserializes to. In interbolt.constants, re-exported at the top level for a consumer parsing a JsonlReporter log directly.

Errors

InterboltError                                          (base)
├── decision outcomes
│   ├── PolicyViolation        # a real block; carries .decision
│   ├── PolicyEvaluationError  # evaluation failed; fail-closed under enforce
│   └── ApprovalDenied         # resolver returned False
└── misuse (also subclasses the matching builtin)
    ├── InterboltConfigError(InterboltError, ValueError)    # bad config value
    └── InterboltUsageError(InterboltError, RuntimeError)   # API used out of sequence

except InterboltError catches every exception the library raises. Because the misuse classes multiply-inherit the matching builtin, except ValueError and except RuntimeError also catch them by their builtin semantics. taint() needs no configured runtime and works before configure() has run, so it never raises a usage error.

Command line

The interbolt console script has four subcommands.

interbolt init [path]              # write the starter policy (default: policy.example.yaml)
interbolt validate policy.yaml     # schema and CEL checks; see Policies: static validation
interbolt inspect log.jsonl        # render a JsonlReporter log as a console tree
                   [--run-id ID]   # limit the render to one run
interbolt explain policy.yaml --agent ID     # or --group NAME / --tool ns.tool
                              [--show-eliminated]  # also print dead rules, dimmed

validate, init, and inspect exit 0 on success and 1 on failure. explain is a reporting command, not a gate: it always exits 0 regardless of what it finds, since its job is to answer "what can this agent do," not to pass or fail a build. See Explain.

validate separates warnings from errors and exits non-zero only on errors, which is what makes it safe to run in a pipeline. See CI.

__version__

The single source of truth for the package version; pyproject.toml reads it dynamically from src/interbolt/__init__.py.