ChokepointInterceptor¶

chokepoint.core.interceptor ¶

ChokepointInterceptor — the integration point between Chokepoint and an agent.

Authorizes every tool call made by one agent. Construct one per agent (the common pattern — see "Multi-agent" in CLAUDE.md), or share a registry across many interceptors for centralized identity management.

Tool arguments and interceptor options travel separately. call() accepts the tool's arguments as **kwargs for brevity, but session_id and domain are named parameters of call() itself, so a tool that happens to declare an argument by one of those names cannot be reached that way. Pass args={...} to keep the two namespaces apart; the interceptor warns when it detects the collision rather than silently dropping the argument.

ChokepointInterceptor ¶

ChokepointInterceptor(
    policies=(),
    mode="enforce",
    *,
    registry=None,
    agent_id=None,
    otel_tracer=None,
    checksum_provider=None,
    consent_provider=None,
    max_sessions=DEFAULT_MAX_SESSIONS,
    call_state=None,
    ledger=None,
    redactor=None,
)

Authorizes every tool call made by one agent.

The interceptor is what turns a list of Policy objects into enforcement: it builds the ExecutionScope carrying caller identity, hands the call to the evaluation engine, and records the outcome to the ledger.

Parameters:

Name	Type	Description	Default
`policies`	`Iterable[Policy]`	Policies evaluated on every call. A policy with no `active_when` applies to every tool routed through this interceptor, which is the most common source of surprise — scope them with `active_when=lambda ctx: ctx.tool_name == "..."`.	`()`
`mode`	`Mode`	`"enforce"` blocks, escalates and undoes for real. `"dry_run"` and `"observe"` evaluate and record every decision but never block, never undo, and never contact an escalation handler.	`'enforce'`
`registry`	`ChokepointRegistry \| None`	Shared identity source. With `agent_id`, populates `ctx.caller_role` / `ctx.trust_level` / `ctx.delegation_chain`, and appends any policies registered for that agent.	`None`
`agent_id`	`str \| None`	Who this interceptor speaks for. Omitting it leaves `ctx.caller_role` permanently `None`, silently defeating any `AgentScopedPolicy` with `allowed_roles` — `chokepoint lint` reports this as an error.	`None`
`otel_tracer`	`Any`	Tracer provider for this interceptor's spans only, overriding the one set globally by `configure_otel()`.	`None`
`checksum_provider`	`Callable[[], str] \| None`	Backs `ctx.state_checksum_matches()`.	`None`
`consent_provider`	`Callable[[str], bool] \| None`	Backs `ctx.patient_consent_on_file()`.	`None`
`max_sessions`	`int \| None`	How many sessions' step counters to keep before evicting the least recently used. `None` for unbounded.	`DEFAULT_MAX_SESSIONS`
`call_state`	`CallState \| None`	Cross-call counters backing rate-limit and budget policies. Private per interceptor by default; pass a shared `CallState` to enforce one quota across several agents.	`None`
`ledger`	`ActionLedger \| None`	Where decisions are recorded. Defaults to the process-wide `ActionLedger.current()`; pass one explicitly to give a tenant its own audit trail inside a shared process.	`None`
`redactor`	`Redactor \| None`	Scrubs arguments and free text at record time. Defaults to the process-wide `current_redactor()`.	`None`

wrapped_tools `property` ¶

wrapped_tools

A snapshot of the tools wrapped through this interceptor, by name.

call ¶

call(
    tool_name,
    func,
    /,
    *,
    args=None,
    session_id=_UNSET,
    domain=_UNSET,
    **kwargs,
)

Run func through this interceptor's policies as tool tool_name.

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	The name policies match on and the ledger records. Positional-only, so a tool argument may share the name.	required
`func`	`Callable[..., Any] \| ReversibleAction`	The tool to call, or a `ReversibleAction` to run with undo support. Positional-only for the same reason.	required
`args`	`Mapping[str, Any] \| None`	The tool's arguments, as an explicit mapping. Use this whenever a tool declares an argument named `session_id` or `domain`, which `**kwargs` cannot express.	`None`
`session_id`	`str`	Groups calls for step counting, rate limits and budgets.	`_UNSET`
`domain`	`str \| None`	Ambient domain available to predicates as `ctx.domain`.	`_UNSET`
`**kwargs`	`Any`	The tool's arguments, for the common case where none of them collide with the parameters above.	`{}`

Returns:

Type	Description
`Any`	Whatever `func` returns.

Raises:

Type	Description
`GuardBlocked`	In `"enforce"` mode, when a policy blocks the call or an escalation is denied or times out.

acall `async` ¶

acall(
    tool_name,
    func,
    /,
    *,
    args=None,
    session_id=_UNSET,
    domain=_UNSET,
    **kwargs,
)

Async sibling of call(), for an async def tool (or a ReversibleAction with an async do_fn). Same arguments, same semantics, same ledger and OTEL behavior.

wrap_tool ¶

wrap_tool(tool_name, func)

Return a callable wrapping func through self.call/self.acall (auto-detected — see _is_async_tool), suitable for registering in place of the original tool on an agent.

Everything the wrapper is called with is forwarded as tool arguments via args=, never as interceptor options: an agent framework invoking a wrapped tool is passing the model's arguments, so a tool parameter named session_id or domain must reach the tool intact.

use ¶

use(agent)

One-line integration point: agent.use(interceptor) ends up calling this. Delegates to chokepoint.adapters.base.wrap for tool auto-discovery.