agent-safe

Agent-Safe

A governance and policy enforcement layer for AI agents and non-human identities.

Agent-Safe is not an agent. It's the system that controls what agents are allowed to do.

---

The Problem

AI agents are getting access to production infrastructure. They restart deployments, scale services, modify configs, and drain nodes. But there's no standard way to answer:

• Is this agent allowed to do this? Not just "does it have credentials" but "should it?"
• What did it do? A tamper-proof record of every action request and decision.
• How risky is this? A restart-deployment in dev is fine. The same action in prod on a critical service is not.

Agent-Safe answers all three.

How It Works

When an AI agent wants to perform an infrastructure action, Agent-Safe:

1. Checks the action against a versioned registry of known, approved actions
2. Evaluates policy -- is this caller allowed to do this action on this target, right now?
3. Returns a decision -- ALLOW, DENY, or REQUIRE_APPROVAL with a reason
4. Logs everything -- append-only, hash-chained audit trail of every request and decision

Agent: "I want to restart-deployment on prod/api-server"
                    |
                    v
            +---------------+
            |  Agent-Safe   |
            |               |
            |  1. Action?   |-- restart-deployment (medium risk)
            |  2. Target?   |-- prod/api-server (critical sensitivity)
            |  3. Policy?   |-- "prod requires approval"
            |  4. Risk?     |-- medium x critical = CRITICAL
            |  5. Decision  |-- REQUIRE_APPROVAL
            |  6. Audit log |-- logged with hash chain
            +---------------+
                    |
                    v
Agent: "OK, I'll queue this for human review."

Quick Start

Install

pip install agent-safe            # Core (policy engine, DryRun/Subprocess executors)
pip install agent-safe[k8s]       # + K8sExecutor (kubernetes Python client)
pip install agent-safe[aws]       # + AwsExecutor (boto3)
pip install agent-safe[dashboard] # + Web dashboard (FastAPI + uvicorn)
pip install agent-safe[all]       # Everything

Scaffold a project

agent-safe init myproject
cd myproject

This creates agent-safe.yaml (with auto-generated signing key), 5 example actions, policies, and an inventory. All CLI commands and the SDK auto-discover this config file -- no flags needed.

Check an action (CLI)

# Dev target -- should be ALLOW
agent-safe check restart-deployment \
    --target dev/test-app \
    --caller deploy-agent \
    --params '{"namespace": "dev", "deployment": "app"}'

# Prod target -- should be REQUIRE_APPROVAL
agent-safe check restart-deployment \
    --target prod/api-server \
    --caller deploy-agent \
    --params '{"namespace": "prod", "deployment": "api"}'

Check an action (Python SDK)

from agent_safe import AgentSafe

# Zero-config -- auto-discovers agent-safe.yaml
safe = AgentSafe()

decision = safe.check(
    action="restart-deployment",
    target="prod/api-server",
    caller="deploy-agent-01",
    params={"namespace": "prod", "deployment": "api-server"},
)

print(decision.result)         # REQUIRE_APPROVAL
print(decision.reason)         # "Production actions require explicit approval"
print(decision.effective_risk) # critical
print(decision.audit_id)       # evt-a1b2c3d4...

Integrate with your agent

from agent_safe import AgentSafe
from agent_safe.models import DecisionResult

safe = AgentSafe()

def agent_step(action, target, params):
    decision = safe.check(
        action=action,
        target=target,
        caller="my-agent",
        params=params,
    )

    if decision.result == DecisionResult.ALLOW:
        execute_action(action, target, params)
    elif decision.result == DecisionResult.REQUIRE_APPROVAL:
        queue_for_review(decision.reason)
    else:
        log_blocked(decision.reason)

Core Concepts

Key Design Decisions

• Default-deny: If no policy matches, the answer is DENY. An unconfigured system blocks everything.
• Advisory + tickets: Agent-Safe decides and logs. ALLOW decisions include a signed execution ticket that executors can validate before acting.
• Stateless PDP: No database. All context comes from the request + config files. Pure function: inputs -> output.
• Context-aware risk: Risk is not a property of the action alone. restart-deployment is medium risk. restart-deployment on prod/payments (critical sensitivity) is critical risk.

K8s Action Catalogue

Agent-Safe ships with 20 curated Kubernetes action definitions:

Each definition includes parameters with type constraints, risk class, reversibility flag, required K8s RBAC privileges, and tags.

AWS Action Catalogue

Agent-Safe also ships with 13 curated AWS action definitions:

Each definition includes parameters, risk class, credential scoping (IAM actions + resources), reversibility, and rollback params where applicable.

Batch Plan Checking

If your agent plans multi-step operations, check them all at once:

plan = [
    {"action": "scale-deployment", "target": "staging/api", "params": {"namespace": "staging", "deployment": "api", "replicas": 3}},
    {"action": "update-image", "target": "staging/api", "params": {"namespace": "staging", "deployment": "api", "container": "api", "image": "api:v2"}},
    {"action": "restart-deployment", "target": "staging/api", "params": {"namespace": "staging", "deployment": "api"}},
]

decisions = safe.check_plan(plan)
# Returns a list of Decision objects, one per step

Agent Identity (JWT)

For role-based access control, use JWT tokens:

safe = AgentSafe()  # signing_key loaded from agent-safe.yaml

token = safe.identity.create_token(
    agent_id="deploy-agent-01",
    roles=["deployer"],
    groups=["platform-team"],
)

# Pass the token as the caller -- JWT is validated automatically
decision = safe.check(action="restart-deployment", caller=token, ...)

Execution Tickets

When signing_key is set (automatically via agent-safe.yaml), ALLOW decisions include a signed execution ticket -- a JWT that executors can validate before acting:

safe = AgentSafe()

decision = safe.check(action="restart-deployment", target="dev/test-app",
                       caller="agent-01", params={"namespace": "dev", "deployment": "app"})

if decision.result == DecisionResult.ALLOW:
    print(decision.ticket.token)      # Signed JWT
    print(decision.ticket.nonce)      # Single-use nonce
    print(decision.ticket.expires_at) # Short TTL (5 min default)

Validate tickets on the executor side:

from agent_safe import TicketValidator

validator = TicketValidator(signing_key="shared-secret-key")
result = validator.validate(token, expected_action="restart-deployment")
print(result.valid)   # True/False
print(result.reason)  # Human-readable reason

Rate Limiting

Throttle per-agent request rates and auto-pause misbehaving agents:

safe = AgentSafe(
    rate_limit={
        "max_requests": 50,              # Per caller, per window
        "window_seconds": 60,            # Sliding window
        "circuit_breaker_threshold": 10,  # DENY count to trip breaker
        "circuit_breaker_cooldown_seconds": 300,
    },
)

Audit Log

Every check() call is logged to an append-only, hash-chained audit file:

# Verify the audit chain hasn't been tampered with
agent-safe audit verify ./audit.jsonl

# Show the last 10 entries
agent-safe audit show ./audit.jsonl --last 10

# Ship audit log to external storage
agent-safe audit ship ./audit.jsonl --backend filesystem --path ./backup.jsonl
agent-safe audit ship ./audit.jsonl --backend s3 --bucket my-audit-bucket

Ship events to external backends in real-time via the SDK:

safe = AgentSafe(
    registry="./actions",
    policies="./policies",
    audit_log="./audit.jsonl",
    audit_shippers={"webhook_url": "https://siem.example.com/ingest"},
)

Multi-Agent Delegation

When orchestrator agents delegate sub-tasks to workers, delegation chains track provenance:

safe = AgentSafe()  # signing_key loaded from agent-safe.yaml

# Create parent identity
parent_token = safe.identity.create_token(
    agent_id="orchestrator-01",
    roles=["deployer", "reader"],
)

# Delegate to a worker with narrowed scope
result = safe.delegate(
    parent_token=parent_token,
    child_agent_id="worker-01",
    child_roles=["deployer"],  # Must be subset of parent's roles
)

# Worker uses delegation token for checks
decision = safe.check(
    action="restart-deployment",
    target="dev/test-app",
    caller=result.token,  # Carries full delegation chain
)

Delegation-aware policies control who can delegate and to what depth:

rules:
  - name: allow-delegated-from-orchestrator
    match:
      callers:
        delegated_from: [orchestrator-01]
        max_delegation_depth: 2
    decision: allow
    reason: Delegated from trusted orchestrator

Cumulative Risk Scoring

Prevent privilege escalation via action chaining by tracking per-caller risk over time:

safe = AgentSafe(
    cumulative_risk={
        "window_seconds": 3600,             # 1 hour sliding window
        "escalation_threshold": 30,          # ALLOW → REQUIRE_APPROVAL
        "deny_threshold": 75,                # Any → DENY
        "risk_scores": {"low": 1, "medium": 5, "high": 15, "critical": 50},
    },
)

# Individual low-risk actions are fine
d1 = safe.check("get-configmap", target="dev/test-app", caller="agent-01",
                 params={"namespace": "dev", "configmap": "cfg"})
# d1.result = ALLOW, d1.cumulative_risk_score = 1

# But chaining high-risk actions triggers escalation
d2 = safe.check("exec-pod", target="dev/debug-pod", caller="agent-01",
                 params={"namespace": "dev", "pod": "p", "command": ["ls"]})
d3 = safe.check("get-secret", target="dev/test-app", caller="agent-01",
                 params={"namespace": "dev", "secret": "db-creds"})
# d3.escalated_from = ALLOW → REQUIRE_APPROVAL due to cumulative risk

Ticket/Incident Linkage

Link actions to external change management tickets for compliance:

# SDK — pass ticket_id to check()
decision = safe.check(
    action="restart-deployment",
    target="prod/api-server",
    caller="deploy-agent-01",
    params={"namespace": "prod", "deployment": "api-server"},
    ticket_id="JIRA-1234",  # Links to external ticket
)
print(decision.ticket_id)  # "JIRA-1234" — also in audit log

# CLI — use --ticket-id
agent-safe check restart-deployment \
    --target prod/api-server \
    --params '{"namespace": "prod", "deployment": "api"}' \
    --ticket-id JIRA-1234

Policies can require a ticket for certain actions:

rules:
  - name: require-ticket-prod
    priority: 500
    match:
      targets:
        environments: [prod]
      require_ticket: true
    decision: allow
    reason: Production changes allowed with ticket

Policy Testing

Validate your policies against expected outcomes:

agent-safe test ./tests/

CLI Reference

Docker (Sidecar)

docker build -t agent-safe .
docker run -v ./myproject:/project -w /project agent-safe check restart-deployment \
    --target prod/api-server \
    --params '{"namespace": "prod", "deployment": "api-server"}'

Web Dashboard

Agent-Safe includes a read-only governance dashboard for real-time visibility:

pip install agent-safe[dashboard]
agent-safe dashboard
# Opens at http://127.0.0.1:8420

Pages:

• Dashboard -- overview stats, timeline chart, recent decisions
• Audit Log -- filterable, paginated event table
• Actions -- browse all 33 action definitions with risk badges
• Policies -- priority-sorted rules with inventory match analysis
• Activity -- live feed with 5s auto-refresh

For frontend development, use --dev to enable CORS for the Vite dev server:

agent-safe dashboard --dev
cd dashboard/frontend && npm run dev  # Vite at http://localhost:5173

Examples & Demos

Agent-Safe ships with 7 runnable examples in examples/:

Run any demo:

pip install -e ".[dev]"
python examples/demo_rate_limit.py

Integration Testing

The tests/integration/ suite validates executors against real infrastructure:

• K8sExecutor tested against Kind (Kubernetes in Docker)
• AwsExecutor tested against LocalStack (local AWS emulator)
• SubprocessExecutor tested against Kind (kubectl operations)
• Full pipeline: check() → ticket → credentials → execute (real K8s) → state capture → audit

Integration tests skip automatically when infrastructure is not available. See infra/setup.sh for one-command setup.

bash infra/setup.sh                          # Start Kind + LocalStack
pytest tests/integration/ -m integration -v  # Run integration tests
bash infra/teardown.sh                       # Cleanup

Documentation

• Getting Started -- install, configure, first check, integrate with agent
• Writing Actions -- define custom action definitions
• Writing Policies -- write policy rules
• Architecture -- design decisions and data flow

Project Status

Alpha (v0.12.2) -- zero-config setup (agent-safe.yaml auto-discovery), core policy engine, SDK, CLI, audit log, K8s action catalogue (20 actions), AWS action catalogue (13 actions), execution tickets, rate limiting, audit shipping, approval workflows, credential gating, multi-agent delegation, cumulative risk scoring, ticket/incident linkage, before/after state capture, rollback pairing, Runner/Executor framework with DryRunExecutor, SubprocessExecutor, K8sExecutor, and AwsExecutor, web governance dashboard, GitHub Actions CI with trusted PyPI publishing, and integration test suite against real infrastructure. 1,102 unit tests + 27 integration tests.

What's next:

• Team/org features (paid tier)
• Multi-cluster policy management

See docs/ROADMAP.md for the full roadmap.

Contributing

Contributions welcome. The easiest way to start:

git clone https://github.com/sahb4k/agent-safe.git
cd agent-safe
pip install -e ".[dev]"
pytest tests/ -v
ruff check src/ tests/ examples/

Add new K8s actions in actions/, write policies in policies/, and run agent-safe validate to check your work.

License

Apache 2.0