APE - Agent Policy Engine

Deterministic, capability-based policy enforcement runtime for AI agents.

APE provides a complete security framework for AI agent systems, enforcing explicit intents, validated plans, and auditable execution. It solves critical security challenges in AI agent deployments including prompt injection, confused deputy attacks, unauthorized privilege escalation, and agent-driven denial of service.

Core Principle

"Prompts guide intent, but never are intent."

User prompts are compiled through APE's Intent Compiler into structured, policy-constrained intents. The prompt itself never becomes executable — only the validated, narrowed intent is enforced.

Features

Security Guarantees

• Deterministic Enforcement: Non-probabilistic, predictable policy decisions
• Cryptographic Binding: Authority tokens bound to specific intent and plan hashes
• Default-Deny: Any action not explicitly allowed is denied
• Complete Audit Trail: Every decision logged for compliance and debugging
• Provenance Tracking: Data origin tracking prevents untrusted data in authority decisions

Core Components

• Action Repository: Canonical registry of known actions with schemas and risk levels
• Intent Compiler: Natural language → structured intent transformation
• Plan Generator: Intent → validated execution plan with LLM support
• Policy Engine: YAML-based policies with parameterized conditions and external evaluator support
• Runtime State Machine: Enforces valid execution flow
• Authority Manager: Single-use, time-limited execution tokens
• Enforcement Gate: Mandatory checkpoint for all tool execution
• Escalation Handler: Routes high-risk actions for approval
• Session Manager: Multi-turn conversation continuity with cumulative tracking
• Rate Limiter: Global, per-action, and per-target rate limiting and quota enforcement
• MCP Scanner: Auto-generates policies from MCP configurations
• APE Orchestrator: Unified one-call API from prompt to execution

Enterprise Integration

• Parameterized Policy Conditions: Constrain parameters at execution time (path prefixes, domain allowlists, size limits)
• External Policy Adapters: Delegate policy decisions to OPA/Rego, AWS Cedar, or XACML engines
• Dynamic Policy Reload: Live policy updates via URL polling or webhook-triggered reload
• Framework SDKs: Native integration adapters for LangChain, AutoGen, and CrewAI

Installation

pip install agent-policy-engine

Or from source:

git clone https://github.com/agent-policy-engine/ape.git
cd ape
pip install -e .

Quick Start

APE offers two integration paths: Orchestrator (simple) and Manual (full control).

Orchestrator Path (Recommended for Most Use Cases)

from ape import APEOrchestrator

# Create orchestrator from policy
orch = APEOrchestrator.from_policy("policies/read_only.yaml")

# Register your tool implementations
orch.register_tool("read_file", lambda path: open(path).read())
orch.register_tool("list_directory", lambda path: os.listdir(path))

# Execute with one-call API
result = orch.execute("Read the config.json file")

if result.success:
    print(result.results[0])  # File contents
else:
    print(f"Error: {result.error}")

Session-Aware Execution

# Create a session for multi-turn conversations
session = orch.create_session(user_id="user_123", ttl_minutes=30)

result1 = session.execute("Read config.json")       # Tracked
result2 = session.execute("Now update it")           # Knows context

# Session tracks cumulative behavior
print(session.actions_executed)   # ["read_file", "write_file"]
print(session.cumulative_risk)    # 0.4
print(session.time_remaining)     # 1740 seconds
session.get_usage_summary()

Manual Path (Maximum Control)

from ape import (
    PolicyEngine, IntentManager, PlanManager,
    RuntimeOrchestrator, AuthorityManager, EnforcementGate,
    ActionRepository, IntentCompiler, PlanGenerator,
    Action, Provenance, RuntimeState,
    create_standard_repository,
)

# 1. Setup components
repository = create_standard_repository()
policy = PolicyEngine("policies/read_only.yaml")
compiler = IntentCompiler(repository)
generator = PlanGenerator(repository)

# 2. Compile intent from prompt
intent = compiler.compile(
    prompt="Read the config.json file",
    policy_allowed=policy.get_all_allowed_actions(),
)

# 3. Generate plan
plan = generator.generate(intent)

# 4. Setup APE runtime
runtime = RuntimeOrchestrator()
intent_manager = IntentManager()
plan_manager = PlanManager(intent_manager)
authority = AuthorityManager(runtime)
enforcement = EnforcementGate(authority)

# 5. Execute through APE
intent_version = intent_manager.set(intent.to_ape_intent(), Provenance.USER_TRUSTED)
runtime.transition(RuntimeState.INTENT_SET)

plan_hash = plan_manager.submit(plan.to_ape_plan(), Provenance.USER_TRUSTED)
plan_manager.approve()
runtime.transition(RuntimeState.PLAN_APPROVED)

runtime.transition(RuntimeState.EXECUTING)

for idx, step in enumerate(plan.steps):
    action = Action(
        action_id=step.action_id,
        tool_id=step.tool_id,
        parameters=step.parameters,
        intent_version=intent_version,
        plan_hash=plan_hash,
        plan_step_index=idx,
    )
    
    policy.evaluate_or_raise(action.action_id)
    token = authority.issue(intent_version, plan_hash, action)
    result = enforcement.execute(token, my_tool, action, **step.parameters)

runtime.transition(RuntimeState.TERMINATED)

Architecture

┌──────────────────────────────────────────────────────────────────────────┐
│                          APE Orchestrator                                │
│               (Unified API - orchestrates all components)                │
├──────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐      │
│  │ Action          │    │ Intent          │    │ Plan            │      │
│  │ Repository      │───►│ Compiler        │───►│ Generator       │      │
│  │                 │    │                 │    │                 │      │
│  │ • Known actions │    │ • Prompt→Intent │    │ • Intent→Plan   │      │
│  │ • Schemas       │    │ • Policy narrow │    │ • LLM parsing   │      │
│  │ • Risk levels   │    │ • Risk filter   │    │ • Validation    │      │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘      │
│                                                          │               │
├──────────────────────────────────────────────────────────┼───────────────┤
│                          APE Core                        │               │
│                                                          ▼               │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐      │
│  │ Intent          │    │ Plan            │    │ Policy          │      │
│  │ Manager         │───►│ Manager         │───►│ Engine          │      │
│  └─────────────────┘    └─────────────────┘    │ • Conditions    │      │
│           │                      │              │ • Adapters      │      │
│           ▼                      ▼              └─────────────────┘      │
│  ┌─────────────────┐    ┌─────────────────┐             │               │
│  │ Runtime         │    │ Authority       │             ▼               │
│  │ Orchestrator    │───►│ Manager         │───►┌─────────────────┐      │
│  │                 │    │                 │    │ Enforcement     │      │
│  │ State machine   │    │ Token lifecycle │    │ Gate            │      │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘      │
│                                                          │               │
│  ┌─────────────────┐    ┌─────────────────┐             ▼               │
│  │ Session         │    │ Rate            │    ┌─────────────────┐      │
│  │ Manager         │    │ Limiter         │    │ Audit Logger    │      │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘      │
└──────────────────────────────────────────────────────────────────────────┘

Policy Configuration

Policies are YAML files that define allowed actions with optional parameter conditions.

# policies/read_only.yaml
name: read_only
version: "1.0"
description: "Read-only file access"

default_decision: deny

rules:
  - action: read_file
    decision: allow
    
  - action: list_directory
    decision: allow
    
  - action: write_file
    decision: deny
    
  - action: delete_file
    decision: deny

Parameterized Conditions

# policies/scoped_access.yaml
rules:
  - action: write_file
    decision: allow
    conditions:
      path:
        prefix: ["/tmp/", "/home/user/workspace/"]
      size_bytes:
        max: 10485760

  - action: http_get
    decision: allow
    conditions:
      domain:
        allowlist: ["api.github.com", "*.internal.corp"]

Example Policies

Policy	Description
minimal_safe.yaml	Minimal read-only, maximum safety
read_only.yaml	File reading only
development.yaml	Broader permissions for development
filesystem_scoped.yaml	Path-constrained file access
human_in_loop.yaml	All actions require escalation

Risk Levels

Actions are classified by risk:

Level	Description	Default Behavior
MINIMAL	Read-only, no side effects	Allowed
LOW	Reversible side effects	Allowed
MODERATE	Irreversible but recoverable	Allowed with caution
HIGH	Potentially destructive	Requires escalation
CRITICAL	Requires explicit approval	Always escalates

CLI Tools

# Validate a policy file
ape validate policies/my_policy.yaml

# Test a prompt through the full pipeline
ape test-prompt policies/read_only.yaml "Read the config file"

# Analyze a prompt without policy check
ape analyze "Read config.json and delete temp files"

# Simulate policy evaluation for an action
ape simulate policies/read_only.yaml read_file

# Compare two policies
ape diff policies/read_only.yaml policies/development.yaml

# List available actions
ape actions --by-category

# Scan MCP configuration and generate policy
ape mcp-scan ~/.config/claude/claude_desktop_config.json -o policies/mcp_generated.yaml

Error Handling

APE provides typed, deterministic errors:

from ape import (
    PolicyDenyError,
    EscalationRequiredError,
    IntentAmbiguityError,
    PlanValidationError,
    RateLimitExceededError,
    SessionExpiredError,
    PolicyConditionError,
)

try:
    result = orch.execute("Delete all files")
except PolicyDenyError as e:
    print(f"Policy denied: {e.action_id}")
except EscalationRequiredError as e:
    print(f"Needs approval: {e.action_id}")
except RateLimitExceededError:
    print("Rate limit exceeded, try again later")

Security Considerations

Threat Model

APE mitigates:

Threat	Mitigation
Prompt Injection	Prompts compiled, never executed directly
Confused Deputy	Actions bound to explicit intent
Privilege Escalation	Policy narrowing, risk-based filtering
Token Replay	Single-use, time-limited tokens
Plan Tampering	Cryptographic plan hashing
Agent DDoS	Rate limiting with per-action and per-target quotas
Parameter Abuse	Parameterized conditions on action parameters

Best Practices

1. Start with minimal policies — Add permissions as needed
2. Use parameterized conditions — Constrain file paths, domains, and resource limits
3. Use sessions for multi-turn agents — Track cumulative risk across a conversation
4. Configure rate limits — Prevent runaway API calls and resource exhaustion
5. Use escalation for high-risk actions — Human approval for destructive operations
6. Enable audit logging — Complete trail for compliance
7. Validate all policies — Use ape validate before deployment
8. Test with dry-run — Use orch.analyze_prompt() before execution

Development

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy ape

# Linting
ruff check ape

License

Apache License 2.0 — See LICENSE for details.

Contributing

Contributions welcome! Please read our contributing guidelines and submit pull requests.

Changelog

See CHANGELOG.md for version history.