calute.security.policy

Tool policy enforcement layer for Calute.

Provides configurable allow/deny policies for tool execution at both global and per-agent levels. Policies are evaluated before any tool call is dispatched, blocking unauthorized calls with a clear error.

Design:

• Global policy applies to all agents unless overridden.

• Per-agent policy takes precedence over global for that agent.

• An explicit allow-list means only those tools are permitted.

• An explicit deny-list means all tools except those are permitted.

• If both allow and deny are set, allow takes precedence (intersection).

• Optional tools require explicit opt-in via the allow list.

class calute.security.policy.PolicyAction(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Result of a policy evaluation.

Represents the two possible outcomes when a tool invocation is checked against a ToolPolicy.

ALLOW

The tool invocation is permitted by the policy.

DENY

The tool invocation is blocked by the policy.

Example

>>> action = PolicyAction.ALLOW
>>> action.value
'allow'

ALLOW = 'allow'

DENY = 'deny'

class calute.security.policy.PolicyEngine(global_policy: calute.security.policy.ToolPolicy | None = None, agent_policies: dict[str, calute.security.policy.ToolPolicy] | None = None)

Bases: object

Evaluates tool policies at global and per-agent level.

The engine holds a global_policy that applies to every agent and an optional dict of agent_policies keyed by agent ID. Per-agent policies fully override the global policy for that agent (no merging).

Listeners can be registered to observe every policy check, which is useful for audit logging or metrics collection.

global_policy

The default ToolPolicy applied when no per-agent policy matches.

agent_policies

Mapping of agent ID to ToolPolicy. When an agent ID matches a key in this dict, that policy is used instead of the global policy.

Example

>>> engine = PolicyEngine(
...     global_policy=ToolPolicy(deny={"execute_shell"}),
... )
>>> engine.check("execute_shell", agent_id="coder")
PolicyAction.DENY
>>> engine.set_agent_policy("coder", ToolPolicy(allow={"execute_shell"}))
>>> engine.check("execute_shell", agent_id="coder")
PolicyAction.ALLOW

add_listener(callback: Callable[[str, str | None, PolicyAction], None]) → None

Register a listener that is notified on every policy check.

Listeners are called synchronously after each policy evaluation. If a listener raises an exception, the error is logged as a warning and the remaining listeners are still invoked.

Parameters

callback – A callable that receives (tool_name, agent_id, action) where tool_name is the tool being checked, agent_id is the optional agent identifier, and action is the resulting PolicyAction.

check(tool_name: str, agent_id: str | None = None) → PolicyAction

Check whether a tool invocation is allowed for a given agent.

Resolves the applicable policy (per-agent if available, otherwise global), evaluates it, notifies all registered listeners, and logs denied actions at INFO level.

Parameters

• tool_name – The name of the tool to check.

• agent_id – Optional identifier of the agent requesting the tool. When None, only the global policy is consulted.

Returns

PolicyAction.ALLOW if the tool is permitted, or PolicyAction.DENY if it is blocked.

enforce(tool_name: str, agent_id: str | None = None) → None

Check a tool invocation and raise on denial.

This is a convenience wrapper around check() that raises a ToolPolicyViolation when the policy decision is DENY, making it suitable for use in enforcement points where a blocked tool should halt execution.

Parameters

• tool_name – The name of the tool to check.

• agent_id – Optional identifier of the agent requesting the tool.

Raises

ToolPolicyViolation – If the tool is denied by the applicable policy.

remove_agent_policy(agent_id: str) → None

Remove a per-agent policy so the agent falls back to the global policy.

If no per-agent policy exists for the given agent, this is a no-op.

Parameters

agent_id – The unique identifier of the agent whose policy should be removed.

set_agent_policy(agent_id: str, policy: ToolPolicy) → None

Set or replace the per-agent policy for a specific agent.

When set, this policy fully overrides the global policy for the given agent (no merging occurs).

Parameters

• agent_id – The unique identifier of the agent.

• policy – The ToolPolicy to assign to this agent.

set_global_policy(policy: ToolPolicy) → None

Replace the global policy applied to all agents without a per-agent override.

Parameters

policy – The new ToolPolicy to use as the global default.

class calute.security.policy.ToolPolicy(allow: set[str] = <factory>, deny: set[str] = <factory>, optional_tools: set[str] = <factory>)

Bases: object

A single allow/deny policy for tool invocation.

allow

Explicit set of tool names that are permitted. If non-empty, only these tools can be called.

Type

set[str]

deny

Explicit set of tool names that are blocked. If non-empty, these tools cannot be called.

Type

set[str]

optional_tools

Tools that exist but require explicit opt-in. They are denied unless they appear in allow.

Type

set[str]

allow: set[str]

deny: set[str]

evaluate(tool_name: str) → PolicyAction

Evaluate whether a given tool name is permitted by this policy.

The evaluation follows a strict precedence order:

1. If the allow set is non-empty, the tool must be present in it; otherwise it is denied.

2. If the deny set is non-empty and the tool is in it, the tool is denied.

3. If the tool is in optional_tools but not explicitly in allow, the tool is denied.

4. Otherwise, the tool is allowed.

Parameters

tool_name – The name of the tool to evaluate against this policy.

Returns

PolicyAction.ALLOW if the tool is permitted, PolicyAction.DENY if the tool is blocked.

Example

>>> policy = ToolPolicy(deny={"execute_shell"})
>>> policy.evaluate("execute_shell")
<PolicyAction.DENY: 'deny'>
>>> policy.evaluate("read_file")
<PolicyAction.ALLOW: 'allow'>

optional_tools: set[str]

exception calute.security.policy.ToolPolicyViolation(tool_name: str, agent_id: str | None = None)

Bases: Exception

Raised when a tool call is blocked by policy.

tool_name

The name of the tool that was denied.

agent_id

The agent identifier that attempted the call, or None if no agent context was provided.

Example

>>> raise ToolPolicyViolation("execute_shell", agent_id="coder")
Traceback (most recent call last):
    ...
ToolPolicyViolation: Tool 'execute_shell' is denied by policy for agent 'coder'