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}

🔌 Framework Integrations

AgentSentinel provides robust integrations with popular AI frameworks, featuring active policy enforcement and intervention tracking.

LangChain Integration

The LangChain integration provides 4 critical features:

1. Pre-Authorization Checks: Blocks tools/LLMs if budget exceeded or policy violated (like a credit card terminal checking funds)
2. Intervention Recording: Tracks blocked/modified actions for dashboard visibility
3. Async Support: Fully compatible with async LangChain agents in production
4. Context Propagation: Maintains agent identity and budget across nested chains

Basic Usage:

from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, Tool, AgentType
from agent_sentinel.integrations.langchain import SentinelCallbackHandler

# Create callback handler with enforcement
sentinel = SentinelCallbackHandler(
    run_name="my_agent_run",
    track_costs=True,
    track_tools=True,
    enforce_policies=True  # Enable active blocking
)

# Use with LangChain LLM
llm = ChatOpenAI(
    temperature=0,
    model="gpt-4o-mini",
    callbacks=[sentinel]
)

# Or with agents
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    callbacks=[sentinel],
    verbose=True
)

# Run agent - enforcement happens automatically
try:
    result = agent.run("What's the weather in SF?")
except BudgetExceededError as e:
    print(f"Blocked: {e}")
    # View interventions
    from agent_sentinel import InterventionTracker
    interventions = InterventionTracker.get_interventions()

With Policy Enforcement:

from agent_sentinel import PolicyEngine

# Configure budget and deny lists
PolicyEngine.configure(
    run_budget=0.50,  # Max $0.50 per run
    denied_actions=["delete_database", "send_email"],
    strict_mode=True
)

# Create handler - it will enforce policies before each action
sentinel = SentinelCallbackHandler(
    run_name="policy_enforced_agent",
    enforce_policies=True
)

llm = ChatOpenAI(callbacks=[sentinel])

# Second LLM call will be BLOCKED if over budget
# Dangerous tools will be BLOCKED before execution
# All blocks are recorded as interventions for dashboard

Async Agents:

# Works with async LangChain agents automatically
async def run_async_agent():
    sentinel = SentinelCallbackHandler(
        run_name="async_agent",
        enforce_policies=True
    )
    
    llm = ChatOpenAI(callbacks=[sentinel])
    # All async callbacks are supported
    result = await llm.apredict("Hello!")
    return result

Rate Limiting:

PolicyEngine.configure(
    rate_limits={
        "expensive_search": {
            "max_count": 10,
            "window_seconds": 60
        }
    }
)

# Tool will be blocked after 10 calls per minute
sentinel = SentinelCallbackHandler(enforce_policies=True)
agent = initialize_agent(tools, llm, callbacks=[sentinel])

View Interventions:

from agent_sentinel import InterventionTracker

# After running your agent
interventions = InterventionTracker.get_interventions()

for intervention in interventions:
    print(f"Type: {intervention.intervention_type.value}")
    print(f"Outcome: {intervention.outcome.value}")
    print(f"Action: {intervention.action_name}")
    print(f"Reason: {intervention.reason}")
    print(f"Risk Level: {intervention.risk_level}")

Examples:

• Basic tracking: examples/langchain_integration.py
• Policy enforcement: examples/langchain_policy_enforcement.py

CrewAI Integration

from agent_sentinel.integrations.crewai import SentinelCrewCallbacks
from crewai import Agent, Task, Crew

# Create callbacks
callbacks = SentinelCrewCallbacks(
    run_name="crewai_agent",
    track_costs=True
)

# Create agent
agent = Agent(
    role="Researcher",
    goal="Research topics",
    backstory="Expert researcher",
    callbacks=callbacks
)

# Create and run crew
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()

See examples/crewai_integration.py for more details.

OpenAI Integration

from agent_sentinel.integrations.openai import track_openai

# Track OpenAI calls automatically
track_openai()

# Now all OpenAI calls are tracked
import openai
response = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}]
)

See examples/llm_instrumentation.py for more integrations.

🐛 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: hello@agentsentinel.dev
• Docs: https://docs.agentsentinel.dev (coming soon)