agent-risk-engine

A layered protocol and reference implementation for codifying risk in autonomous agent actions.

See PROTOCOL.md for the language-agnostic protocol specification.

Installation

uv

uv add agent-risk-engine

pip

pip install agent-risk-engine

Quick Start

from agent_risk_engine import RuleGate, RiskEvaluator, Action, GateResult

gate = RuleGate(threshold="cautious")
evaluator = RiskEvaluator(rule_gate=gate)

result = await evaluator.evaluate(Action(kind="tool_call", name="read_file", risk=1))
assert result.decision == GateResult.ALLOWED

result = await evaluator.evaluate(
    Action(kind="tool_call", name="execute_shell", parameters={"command": "rm -rf /"}, risk=5)
)
assert result.decision == GateResult.NEEDS_APPROVAL

Architecture

Actions pass through a 3-layer pipeline:

flowchart LR
    A["Action arrives"] --> B

    B["**RuleGate** · L1\nFast static rules\nNo LLM · Microseconds"]
    B -->|denied| Z["DENIED"]
    B -->|passes| C

    C["**ActionAnalyzer** · L2\nArgument-aware scoring\nPassthrough stub by default"]
    C -->|scored| D

    D["**ActionGate** · L3\nRisk vs utility tradeoff\nOnly escalates, never relaxes"]
    D --> E["ALLOWED / NEEDS_APPROVAL / DENIED"]

L1 (RuleGate) and the RiskUtilityGate implementation of L3 are fully implemented. L2 ships as a passthrough stub — plug in your own ActionAnalyzer.

Risk Levels

RuleGate

Fast, deterministic, no LLM required. Supports per-kind threshold routing:

gate = RuleGate(
    threshold="cautious",
    kind_thresholds={
        "tool_call": "standard",
        "file_write": 2,
        "code_execution": 1,
    },
    denied={"delete_database"},
    allowed={"read_logs"},
)

Evaluation order: denied → allowed → approve → threshold comparison.

Threshold Aliases

PatternAnalyzer

Scores actions by matching regex patterns against serialized parameters. Supports kind-scoped patterns:

from agent_risk_engine import PatternAnalyzer, RiskPattern

analyzer = PatternAnalyzer(extra_patterns=[
    RiskPattern(r"\bDROP\b", 5, "SQL drop", kinds=frozenset({"database_query"})),
])

Pass it to RiskEvaluator(rule_gate=gate, action_analyzer=analyzer).

RiskUtilityGate

Weighs risk against caller-provided utility. Utility is an input, not computed internally — the library evaluates risk; your framework understands agent goals.

evaluator = RiskEvaluator(
    rule_gate=RuleGate(threshold="standard"),
    action_gate=RiskUtilityGate(),
)

result = await evaluator.evaluate(
    Action(kind="tool_call", name="write_file", risk=3),
    utility=UtilityScore(level=4, reasoning="User explicitly requested"),
)

The gate only escalates, never relaxes — it cannot make a decision less restrictive than L1.

Extending with Custom Analyzers

ActionAnalyzer is a Protocol. Implement analyze(action: Action) -> RiskScore:

from agent_risk_engine import RiskEvaluator, RuleGate, RiskScore, Action

class LLMAnalyzer:
    """Use an LLM to evaluate the actual risk of action arguments."""

    async def analyze(self, action: Action) -> RiskScore:
        # Inspect action.parameters, reason about consequences
        assessed_level = await my_llm_judge(action.name, action.parameters)
        return RiskScore(level=assessed_level, reasoning="LLM analysis")

evaluator = RiskEvaluator(
    rule_gate=RuleGate(threshold="cautious"),
    action_analyzer=LLMAnalyzer(),
)

CallTracker

Standalone loop and repetition detection. Not a pipeline layer — use it to build context before evaluating:

from agent_risk_engine import CallTracker

tracker = CallTracker()
tracker.record(action.name)
context = tracker.check()
# Merge into action metadata before evaluating
action = Action(kind=action.kind, name=action.name, risk=action.risk, metadata=context)

check() returns {"healthy": bool, "warnings": list[str]}.

Framework Integration

Write a thin adapter that maps your framework's action primitives to Action:

async def before_action_hook(action_name, args, action_risk):
    action = Action(kind="tool_call", name=action_name, parameters=args, risk=action_risk)
    result = await evaluator.evaluate(action)

    if result.decision == GateResult.DENIED:
        raise PermissionError(result.reasoning)
    if result.decision == GateResult.NEEDS_APPROVAL:
        approved = await prompt_user(f"Allow '{action_name}'? Risk: {result.risk_score.level}/5")
        if not approved:
            raise PermissionError("User denied")

License

MIT