agentsentinel-sdk 0.1.0

Runtime authority and control for autonomous AI agents

AgentSentinel SDK

The operational circuit breaker for autonomous agents

Runtime authority and control for autonomous AI agents. Decide whether your agents are allowed to act — before they do.

🎯 What is AgentSentinel?

AgentSentinel gives developers and organizations active control over the financial, logical, and compliance risks of autonomous agents in production.

We are not:

• A logging tool (that's a camera)
• An analytics dashboard (that's a report)
• An observability platform (that's hindsight)

We are:

• The brakes on the car
• The circuit breaker in the system
• The authorization layer between intent and action

Core Value: AgentSentinel exists to decide whether an autonomous system is allowed to act — before it does.

✨ The Three Pillars

1. Active Safety 🛡️

Runtime enforcement that overrides agent intent. Every check happens BEFORE execution.

• ⚡ Budget enforcement (hard caps on cost per action, run, session)
• 🚫 Action bans (deny list prevents execution)
• ✅ Action allowlists (restrict to approved actions only)
• ⏱️ Rate limiting (time-windowed execution limits)
• 🛡️ Fail-safe design (never fails open without authorization)

2. Governance 📋

Complete authority over what happened, when, and why.

• 📝 Immutable execution ledger
• 💰 Per-action, per-run, per-session cost tracking
• 🔄 Replay mode for deterministic testing
• 📊 Decision rationale and data lineage
• 🇪🇺 EU AI Act Article 14 compliance metadata

3. Operations ⚙️

The bridge between autonomous agents and human authority.

• 👤 Human-in-the-loop approval workflows
• 📋 Policy engine with YAML configuration
• 🔌 Remote policy synchronization
• 📡 Real-time intervention tracking
• ⏸️ Interruptible execution with resumable authority

📦 Installation

# Basic installation
pip install agentsentinel-sdk

# With remote sync to platform
pip install agentsentinel-sdk[remote]

# With LLM integrations (OpenAI, Anthropic, Grok, Gemini)
pip install agentsentinel-sdk[llm]

# With framework integrations (LangChain, CrewAI, MCP)
pip install agentsentinel-sdk[integrations]

# With everything
pip install agentsentinel-sdk[all]

# Development installation
git clone https://github.com/agent-sentinel/agent-sentinel-sdk.git
cd agent-sentinel-sdk
uv sync

🚀 Quick Start

Basic Usage

from agent_sentinel import guarded_action

# Decorate any function to track it
@guarded_action(name="search_tool", cost_usd=0.01, tags=["search", "api"])
def search_web(query: str) -> dict:
    """Search the web for a query"""
    results = call_search_api(query)
    return {"results": results, "count": len(results)}

# Use it normally - telemetry happens automatically
response = search_web("Python best practices")

Async Support

@guarded_action(name="generate_text", cost_usd=0.05, tags=["llm"])
async def generate_text(prompt: str) -> str:
    """Generate text using an LLM"""
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# Async works automatically
result = await generate_text("Write a haiku about Python")

Complex Objects

The SDK handles complex, non-serializable objects gracefully:

from dataclasses import dataclass
import socket

@dataclass
class Context:
    user_id: int
    session: str
    connection: socket.socket  # Not JSON-serializable!

@guarded_action(name="process_data", cost_usd=0.001)
def process_data(ctx: Context, data: dict):
    # Complex objects are automatically stringified
    return {"processed": True, "user": ctx.user_id}

# No crashes even with non-serializable objects!
ctx = Context(user_id=123, session="abc", connection=socket.socket())
process_data(ctx, {"value": 42})

📊 Log Format

Logs are written as JSON Lines (.jsonl) to .agent-sentinel/ledger.jsonl:

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2025-01-15T10:30:00.123456Z",
  "action": "search_tool",
  "cost_usd": 0.01,
  "duration_ms": 234.5,
  "outcome": "success",
  "tags": ["search", "api"],
  "payload": {
    "inputs": {
      "args": [],
      "kwargs": {"query": "Python best practices"}
    },
    "outputs": {
      "results": ["..."],
      "count": 10
    }
  }
}

Fields Explained

• id: Unique UUID for this action
• timestamp: ISO 8601 timestamp (UTC)
• action: Name of the action (from decorator or function name)
• cost_usd: Cost in USD for this action
• duration_ms: Execution time in milliseconds
• outcome: "success" or "error"
• tags: Custom tags for categorization
• payload.inputs: Function arguments (args and kwargs)
• payload.outputs: Return value (on success) or error message (on failure)

🎨 Advanced Usage

Budget Enforcement

Programmatic Configuration:

from agent_sentinel import PolicyEngine, CostTracker, BudgetExceededError

# Configure budget limits
PolicyEngine.configure(
    session_budget=5.0,    # Max $5 for entire session
    run_budget=1.0,        # Max $1 per run
    action_budgets={
        "expensive_llm": 0.50,  # Max $0.50 for this action
    }
)

# Your actions now respect these limits
@guarded_action(name="expensive_llm", cost_usd=0.30)
def call_gpt4(prompt: str) -> str:
    # This will be blocked if budget exceeded
    return generate_text(prompt)

# Check current costs
print(f"Run total: ${CostTracker.get_run_total():.2f}")

# Reset between agent runs
CostTracker.reset_run()

YAML Configuration:

Create callguard.yaml:

budgets:
  session: 5.0
  run: 1.0
  actions:
    expensive_llm: 0.50

denied_actions:
  - delete_database
  - dangerous_operation

strict_mode: true

Load it:

from agent_sentinel import PolicyEngine

PolicyEngine.load_from_yaml("callguard.yaml")
# Now all decorated actions respect these policies

Action Control Lists

Deny Dangerous Actions:

PolicyEngine.configure(
    denied_actions=["delete_database", "send_money"]
)

@guarded_action(name="delete_database")
def delete_database():
    # This will raise PolicyViolationError immediately
    pass

Allowlist Safe Actions:

PolicyEngine.configure(
    allowed_actions=["read_file", "search_web"]
)

# Only these actions are permitted
# All others will raise PolicyViolationError

Remote Sync to Platform

Enable Background Sync:

from agent_sentinel import enable_remote_sync, flush_and_stop

# Start background sync
sync = enable_remote_sync(
    platform_url="https://api.agentsentinel.dev",
    api_token="your-jwt-token",
    flush_interval=10.0,  # Upload every 10 seconds
)

# Use your agent normally - logs are synced automatically

# At exit, flush remaining logs
flush_and_stop()

Manual Sync Control:

from agent_sentinel import BackgroundSync, SyncConfig

# Create custom sync config
config = SyncConfig(
    platform_url="https://api.agentsentinel.dev",
    api_token="your-jwt-token",
    run_id="custom-run-id",  # Optional
    flush_interval=5.0,
    batch_size=50,
    max_retries=3,
)

# Start sync
sync = BackgroundSync(config)
sync.start()

# Trigger immediate flush
sync.flush_now()

# Stop with final flush
sync.stop()

Fail-Open Design:

The sync is designed to never crash your agent:

• If platform is unreachable, logs stay local
• Retries with exponential backoff
• Continues logging locally even if all uploads fail
• Agent performance unaffected by network issues

Custom Action Names

# Use function name (default)
@guarded_action(cost_usd=0.01)
def my_function():
    pass  # Action name: "my_function"

# Or specify explicitly
@guarded_action(name="custom_name", cost_usd=0.01)
def my_function():
    pass  # Action name: "custom_name"

Tagging and Organization

@guarded_action(
    name="expensive_operation",
    cost_usd=1.50,
    tags=["llm", "gpt4", "production"]
)
def expensive_operation():
    pass

Error Handling

The decorator never catches your exceptions - they propagate normally:

@guarded_action(name="might_fail", cost_usd=0.01)
def risky_operation():
    if something_bad:
        raise ValueError("Oops!")  # This exception is raised normally
    return "success"

try:
    risky_operation()
except ValueError as e:
    print(f"Caught: {e}")
    # Exception is logged with outcome="error" and error message in outputs

Configuration

import os

# Change log directory via environment variable
os.environ["AGENT_SENTINEL_HOME"] = "/var/log/my-agent"

# Now logs go to /var/log/my-agent/ledger.jsonl

Reading Logs Programmatically

import json
from pathlib import Path

ledger_path = Path(".agent-sentinel/ledger.jsonl")

# Read all entries
with open(ledger_path) as f:
    entries = [json.loads(line) for line in f]

# Calculate total cost
total_cost = sum(e["cost_usd"] for e in entries)
print(f"Total cost: ${total_cost:.2f}")

# Filter by outcome
errors = [e for e in entries if e["outcome"] == "error"]
print(f"Found {len(errors)} errors")

🔧 API Reference

@guarded_action(name=None, cost_usd=0.0, tags=None)

Decorator to wrap a function with telemetry and cost tracking.

Parameters:

• name (str, optional): Name for the action. Defaults to function name.
• cost_usd (float, optional): Cost in USD for this action. Default: 0.0
• tags (list[str], optional): Tags for categorization. Default: []

Returns:

• Decorated function that logs telemetry and re-raises exceptions

Example:

@guarded_action(name="my_action", cost_usd=0.05, tags=["api", "external"])
def my_function(arg1, arg2):
    return arg1 + arg2

Exceptions

All exceptions inherit from AgentSentinelError:

from agent_sentinel import (
    AgentSentinelError,       # Base exception
    BudgetExceededError,       # Cost limit exceeded (Phase 2)
    PolicyViolationError,      # Policy rule violated (Phase 2)
    ReplayDivergenceError,     # Replay mismatch (coming soon)
)

# Handle budget violations
try:
    expensive_operation()
except BudgetExceededError as e:
    print(f"Budget exceeded: {e}")
    # Take corrective action

# Handle policy violations
try:
    denied_operation()
except PolicyViolationError as e:
    print(f"Policy violation: {e}")
    # Log security incident

CostTracker

Track costs across your application:

from agent_sentinel import CostTracker

# Get current totals
session_total = CostTracker.get_session_total()
run_total = CostTracker.get_run_total()

# Get per-action statistics
stats = CostTracker.get_action_stats("expensive_llm")
print(f"Calls: {stats['count']}, Cost: ${stats['total_cost']:.2f}")

# Get all action stats
all_stats = CostTracker.get_action_stats()
for action, cost in all_stats['costs'].items():
    count = all_stats['counts'][action]
    print(f"{action}: {count} calls, ${cost:.2f}")

# Reset for new run
CostTracker.reset_run()

# Get complete snapshot
snapshot = CostTracker.get_snapshot()

PolicyEngine

Configure and enforce policies:

from agent_sentinel import PolicyEngine

# Programmatic configuration
PolicyEngine.configure(
    session_budget=10.0,
    run_budget=1.0,
    action_budgets={"expensive_action": 0.50},
    denied_actions=["dangerous_op"],
    allowed_actions=None,  # None = all allowed (except denied)
    strict_mode=True
)

# Load from YAML
PolicyEngine.load_from_yaml("callguard.yaml")

# Check if configured
if PolicyEngine.is_configured():
    config = PolicyEngine.get_config()
    print(f"Run budget: ${config.run_budget}")

# Reset (for testing)
PolicyEngine.reset()

Ledger

Direct access to the ledger (advanced usage):

from agent_sentinel import Ledger

# Manually record an entry
Ledger.record(
    action="manual_entry",
    inputs={"key": "value"},
    outputs={"result": 42},
    cost_usd=0.01,
    duration_ms=100.0,
    outcome="success",
    tags=["manual"]
)

🎯 Use Cases

1. LLM Agent Monitoring

@guarded_action(name="llm_call", cost_usd=0.03, tags=["openai", "gpt4"])
async def call_llm(messages: list[dict]) -> str:
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=messages
    )
    return response.choices[0].message.content

2. Tool Call Tracking

@guarded_action(name="database_query", cost_usd=0.001, tags=["db"])
def query_database(sql: str) -> list:
    return database.execute(sql).fetchall()

@guarded_action(name="api_request", cost_usd=0.01, tags=["external"])
def call_external_api(endpoint: str) -> dict:
    return requests.get(f"{BASE_URL}/{endpoint}").json()

3. Multi-Agent Systems

# Agent 1
@guarded_action(name="planner_think", cost_usd=0.02, tags=["agent1", "planning"])
def plan_tasks(goal: str) -> list[str]:
    return ["task1", "task2", "task3"]

# Agent 2
@guarded_action(name="executor_run", cost_usd=0.05, tags=["agent2", "execution"])
def execute_task(task: str) -> dict:
    return {"status": "done", "task": task}

🐛 Troubleshooting

Logs not appearing?

import logging
logging.basicConfig(level=logging.DEBUG)

# Check if directory is writable
from pathlib import Path
log_dir = Path(".agent-sentinel")
print(f"Log directory: {log_dir.absolute()}")
print(f"Exists: {log_dir.exists()}")
print(f"Writable: {log_dir.exists() and log_dir.is_dir()}")

Large logs?

The SDK appends to the same file. Rotate logs periodically:

# Rotate logs (keep last 5)
mv .agent-sentinel/ledger.jsonl .agent-sentinel/ledger-$(date +%Y%m%d).jsonl
# Compress old logs
gzip .agent-sentinel/ledger-*.jsonl

📄 License

MIT License - see LICENSE file for details

🤝 Contributing

See the main repository README for contribution guidelines.

📧 Support

• Issues: GitHub Issues
• Email: [e-mail]
• Docs: https://docs.agentsentinel.dev (coming soon)