PolicyEvaluator — Python Policy Evaluation Engine

Agent action ──► PolicyEvaluator.evaluate(context)
                       │
                  YAML rules (priority-sorted)
                  External backends (OPA / Cedar)
                  Default action (fail-closed deny)
                       │
                 PolicyDecision { allowed, action, reason, audit_entry }

pip install agent-governance-toolkit-core

from agent_os.policies import (
    PolicyEvaluator,
    PolicyDocument,
    PolicyDecision,
    PolicyRule,
    PolicyCondition,
    PolicyDefaults,
    PolicyAction,
    PolicyOperator,
)

PolicyEvaluator(policies=None, root_dir=None)

evaluator.load_policies(directory: str | Path) -> None

evaluator = PolicyEvaluator()
evaluator.load_policies("./policies/global/")
evaluator.load_policies("./policies/team-specific/")
# All rules from all directories are now active

evaluator.evaluate(
    context: dict[str, Any],
    dynamic_context: dict[str, Any] | None = None,
) -> PolicyDecision

evaluator.add_backend(backend: Any) -> None

evaluator.load_rego(
    rego_path: str | None = None,
    rego_content: str | None = None,
    package: str = "agentos",
    mode: str = "local",
) -> OPABackend

evaluator.load_cedar(
    policy_path: str | None = None,
    policy_content: str | None = None,
    entities: list[dict[str, Any]] | None = None,
    mode: str = "auto",
) -> CedarBackend

# Load from YAML file (requires pyyaml)
doc = PolicyDocument.from_yaml("policies/production.yaml")

# Load from JSON file
doc = PolicyDocument.from_json("policies/production.json")

# Serialize back
doc.to_yaml("policies/production.yaml")
doc.to_json("policies/production.json")

from agent_os.exceptions import PolicyViolationError

from agent_os.exceptions import PolicyDeniedError

from agent_os.policies import PolicyEvaluator

evaluator = PolicyEvaluator()
evaluator.load_policies("./policies/")  # loads all .yaml/.yml files

context = {
    "tool_name": "execute_code",
    "token_count": 500,
}
decision = evaluator.evaluate(context)

if decision.allowed:
    print(f"✅ Permitted by rule: {decision.matched_rule or 'default'}")
else:
    print(f"❌ Denied: {decision.reason}")
    print(f"   Rule: {decision.matched_rule}")

from agent_os.policies import (
    PolicyEvaluator, PolicyDocument, PolicyRule,
    PolicyCondition, PolicyAction, PolicyOperator, PolicyDefaults,
)

policy = PolicyDocument(
    name="production-safety",
    description="Blocks dangerous tools in production",
    rules=[
        PolicyRule(
            name="block-dangerous-tools",
            condition=PolicyCondition(
                field="tool_name",
                operator=PolicyOperator.IN,
                value=["execute_code", "run_shell", "delete_file"],
            ),
            action=PolicyAction.DENY,
            priority=100,
            message="Dangerous tool blocked in production",
        ),
        PolicyRule(
            name="audit-all-tool-calls",
            condition=PolicyCondition(
                field="tool_name",
                operator=PolicyOperator.NE,
                value="",
            ),
            action=PolicyAction.AUDIT,
            priority=10,
            message="All tool calls are audit-logged",
        ),
    ],
    defaults=PolicyDefaults(
        action=PolicyAction.ALLOW,
        max_tokens=2048,
        max_tool_calls=5,
        confidence_threshold=0.95,
    ),
)

evaluator = PolicyEvaluator(policies=[policy])

# Allowed tool
decision = evaluator.evaluate({"tool_name": "web_search"})
assert decision.allowed  # True — matches audit rule, action=audit is allowed

# Blocked tool
decision = evaluator.evaluate({"tool_name": "execute_code"})
assert not decision.allowed  # False — matched block-dangerous-tools
print(decision.reason)       # "Dangerous tool blocked in production"

from agent_os.policies import PolicyEvaluator
from agent_os.exceptions import PolicyDeniedError

evaluator = PolicyEvaluator()
evaluator.load_policies("./policies/")

def call_tool(tool_name: str, **kwargs):
    decision = evaluator.evaluate({"tool_name": tool_name, **kwargs})
    
    if not decision.allowed:
        raise PolicyDeniedError(
            f"Tool '{tool_name}' denied: {decision.reason}",
            details=decision.audit_entry,
        )
    
    # Proceed with tool execution
    return run_tool(tool_name, **kwargs)

policy = PolicyDocument(name="my-policy", rules=[...])
policy.to_yaml("policies/my-policy.yaml")

# Load back
loaded = PolicyDocument.from_yaml("policies/my-policy.yaml")

version: "1.0"
name: production
description: Production safety policy

rules:
  - name: block-code-execution
    condition:
      field: tool_name
      operator: eq
      value: execute_code
    action: block
    priority: 100
    message: Code execution is blocked in production

  - name: token-limit
    condition:
      field: token_count
      operator: gt
      value: 2048
    action: deny
    priority: 90
    message: Token count exceeds production limit

  - name: allow-safe-tools
    condition:
      field: tool_name
      operator: in
      value: [web_search, read_file, summarize]
    action: allow
    priority: 70
    message: Tool is on the approved list

defaults:
  action: deny
  max_tokens: 2048
  max_tool_calls: 5
  confidence_threshold: 0.95

from agent_os.policies import PolicyEvaluator

evaluator = PolicyEvaluator()

# Method 1: convenience helper
backend = evaluator.load_rego(
    rego_path="policies/agent_policy.rego",
    package="agentos",
    mode="local",
)

# Method 2: manual registration
from agent_os.policies import OPABackend

opa = OPABackend(
    rego_content="""
    package agentos

    default allow = false

    allow {
        input.tool_name == "web_search"
    }
    """,
    package="agentos",
    mode="local",
)
evaluator.add_backend(opa)

# YAML rules are evaluated first;
# OPA is only consulted when no YAML rule matches.
decision = evaluator.evaluate({"tool_name": "web_search"})