agentsentry 0.1.0

Production Infrastructure for AI Agents - Cost controls, observability, and framework-agnostic deployment

🛡️ AgentSentry

Production Infrastructure for AI Agents

Prevent runaway costs, gain observability, and deploy AI agents with confidence.

Quick Start • Documentation • Contributing • Examples

---

🎯 The Problem

AI agents are transforming how we build software, but deploying them to production introduces critical operational challenges that traditional MLOps tools don't address:

💸 Runaway Costs

• Agents can enter infinite loops, burning through API credits without oversight
• A single bug can cost hundreds or thousands of dollars in a matter of hours
• No standardized way to set budget limits or loop thresholds per session

🔍 Lack of Observability

• Black box execution: you don't know what your agent is doing until it's too late
• Token usage tracking is manual and error-prone
• No distributed tracing for multi-agent or multi-step workflows

🔧 Framework Lock-In

• Each agent framework (CrewAI, AutoGen, LangChain) has its own patterns
• Migrating between frameworks requires rewriting infrastructure code
• No common interface for deployment, monitoring, or cost controls

🚨 Production Instability

• Agents fail silently or unpredictably
• No circuit breakers or safety mechanisms
• Limited audit trails for debugging and compliance

---

✨ The Solution

AgentSentry is an open-source production infrastructure layer that provides:

┌─────────────────────────────────────────────────────────────┐
│                      Your AI Agent                          │
│              (CrewAI, AutoGen, LangChain, etc.)            │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│                     AgentSentry Layer                       │
│  ┌───────────────┐  ┌──────────────┐  ┌─────────────────┐  │
│  │ Token         │  │ Observability│  │ Framework       │  │
│  │ Guardrails    │  │ & Tracing    │  │ Adapters        │  │
│  └───────────────┘  └──────────────┘  └─────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
                  ┌─────────────────────┐
                  │  LLM APIs           │
                  │  (OpenAI, Anthropic)│
                  └─────────────────────┘

🛡️ Core Features

Feature	Description
Budget Guardrails	Circuit breaker pattern that kills execution when USD budget or loop thresholds are exceeded
Token Tracking	Automatic tracking of token usage, costs, and API calls with per-model pricing
Framework Adapters	Unified interface for CrewAI, AutoGen, LangChain, and custom frameworks
OpenTelemetry Integration	Distributed tracing and metrics for agent execution flows
Database Audit Logs	SQLAlchemy-based persistence for session logs, breaches, and metrics
CLI Tools	agentsentry stack deploy, agentsentry monitor sessions, and more

---

🚀 Quick Start

Installation

# Install core package
pip install agentsentry

# Install with framework support
pip install agentsentry[crewai]    # For CrewAI
pip install agentsentry[langchain] # For LangChain
pip install agentsentry[autogen]   # For AutoGen

3 Commands to Production

1️⃣ Initialize Your Project

agentsentry init --framework crewai

This generates:

• agentsentry.yaml - Configuration file
• .env.template - Environment variables template
• example_agent.py - Sample integration code

2️⃣ Configure Your Limits

Edit agentsentry.yaml:

max_budget_usd: 5.0    # Kill agent if cost exceeds $5
max_loops: 20          # Kill agent after 20 iterations
enable_db_logging: true
database_url: "postgresql://localhost/agentsentry"

3️⃣ Run Your Agent with Protection

from agentsentry import AgentRunner, Config
from agentsentry.adapters.crewai import CrewAIAdapter

# Load configuration
config = Config.from_file('agentsentry.yaml')

# Initialize runner
runner = AgentRunner(config=config)
runner.register_adapter('crewai', CrewAIAdapter)

# Run your agent with cost controls
result = runner.run('crewai', {
    'name': 'research-agent',
    'task': 'Research AI trends',
    'model': 'gpt-4'
})

print(f"Cost: ${result['metrics']['total_cost_usd']:.4f}")
print(f"Tokens: {result['metrics']['total_tokens']}")

That's it! Your agent now has:

• ✅ Budget enforcement
• ✅ Token tracking
• ✅ Loop detection
• ✅ Execution logs

---

📊 Architecture

AgentSentry follows a modular, adapter-based architecture designed for extensibility:

agentsentry/
├── core/                    # Core orchestration engine
│   ├── runner.py           # Main AgentRunner class
│   ├── config.py           # Configuration management
│   └── base_adapter.py     # Abstract adapter interface
├── adapters/               # Framework-specific adapters
│   ├── crewai.py          # CrewAI integration
│   ├── autogen.py         # AutoGen integration
│   └── langchain.py       # LangChain integration
├── middleware/            # Interception & control layer
│   └── token_guardrail.py # Budget & loop enforcement
├── observability/         # Telemetry & tracking
│   ├── tracking.py       # Execution tracker
│   └── telemetry.py      # OpenTelemetry provider
├── persistence/          # Database layer
│   ├── database.py       # SQLAlchemy logger
│   └── models.py         # ORM models
└── cli/                  # Command-line interface
    └── commands/
        ├── stack.py      # Stack deployment
        ├── monitor.py    # Monitoring tools
        └── init.py       # Project initialization

🔌 How Adapters Work

Each framework adapter implements the BaseAdapter interface:

class BaseAdapter(ABC):
    @abstractmethod
    def initialize(self) -> None:
        """Initialize framework dependencies"""
    
    @abstractmethod
    def execute(self, agent_config: Dict) -> Dict:
        """Execute agent with AgentSentry controls"""
    
    @abstractmethod
    def shutdown(self) -> None:
        """Clean resource cleanup"""

This allows AgentSentry to support any agent framework through a plugin system.

---

💰 Token Guardrail Deep Dive

The TokenGuardrail middleware implements the Circuit Breaker pattern for LLM API calls:

How It Works

1. Intercepts API Calls: Hooks into framework API calls to track tokens
2. Calculates Costs: Uses per-model pricing to compute USD cost
3. Checks Thresholds: Compares against max_budget_usd and max_loops
4. Kills Process: Raises BudgetExceededError when limits are exceeded
5. Logs Events: Persists metrics to database for audit trails

Example: Guardrail in Action

from agentsentry.middleware import TokenGuardrail, BudgetExceededError

guardrail = TokenGuardrail(config={
    'max_budget_usd': 2.0,
    'max_loops': 10
})

try:
    with guardrail.track_execution(session_id='session-123'):
        # Your agent code here
        for i in range(100):  # Will hit loop limit
            response = llm.call(prompt)
            
            # Track the call
            guardrail.intercept_api_call(
                session_id='session-123',
                model='gpt-4',
                prompt_tokens=100,
                completion_tokens=50,
                is_loop_iteration=True
            )

except BudgetExceededError as e:
    print(f"🚨 Circuit breaker triggered: {e.threshold_type}")
    print(f"Total cost: ${e.total_cost}")
    print(f"Loops executed: {e.loops_executed}")

Supported Pricing Models:

• OpenAI: GPT-4, GPT-4 Turbo, GPT-3.5 Turbo
• Anthropic: Claude 3 Opus, Sonnet, Haiku
• Custom models (configurable in agentsentry.yaml)

---

📈 Observability & Monitoring

AgentSentry integrates with OpenTelemetry for distributed tracing and metrics:

CLI Monitoring

# View active sessions
agentsentry monitor sessions

# Real-time cost analytics
agentsentry monitor costs --period 24h

# Follow a specific session
agentsentry monitor sessions --session-id abc123 --follow

Metrics Collected

Metric	Description
total_tokens	Cumulative token usage (prompt + completion)
total_cost_usd	Total cost in USD
loops_executed	Number of loop iterations
api_calls	Total API calls made
duration_seconds	Execution time

Database Schema

AgentSentry uses SQLAlchemy with two core tables:

session_logs: Records for each agent execution

• session_id, status, total_tokens, total_cost_usd, loops_executed

threshold_breaches: Audit trail for budget/loop violations

• session_id, threshold_type, error_message, breached_at

---

🛠️ How to Contribute

AgentSentry is in private beta but actively seeking core contributors for the open-source substrate. We're particularly interested in:

🎯 Priority Areas

1. New Framework Adapters: Help us support more agent frameworks

   • OpenAI Assistants API
   • Semantic Kernel
   • LlamaIndex Agents
   • Custom agent frameworks

2. Observability Enhancements: Expand telemetry capabilities

   • Prometheus metrics exporter
   • Jaeger tracing integration
   • Custom dashboard templates

3. Cost Optimization: Advanced budget management

   • Multi-tenant cost allocation
   • Budget forecasting
   • Cost anomaly detection

4. Production Hardening: Enterprise features

   • Kubernetes operator
   • High-availability deployments
   • Multi-region support

🤝 Contributing Guide

Step 1: Set Up Development Environment

# Clone the repository
git clone https://github.com/nitindme/agentsentry.git
cd agentsentry

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install in development mode
pip install -e ".[dev]"

# Run tests
pytest

Step 2: Build a New Adapter

The easiest way to contribute is by adding support for a new framework:

# agentsentry/adapters/your_framework.py
from agentsentry.core.base_adapter import BaseAdapter
from agentsentry.middleware.token_guardrail import TokenGuardrail

class YourFrameworkAdapter(BaseAdapter):
    def __init__(self, config):
        super().__init__(framework_name="your_framework", config=config)
        self.guardrail = TokenGuardrail(config=self.config)
    
    def initialize(self):
        # Initialize your framework
        pass
    
    def execute(self, agent_config):
        with self.guardrail.track_execution(agent_config['session_id']):
            # Execute agent
            # Track API calls using self.guardrail.intercept_api_call()
            return {"status": "success"}
    
    def shutdown(self):
        # Cleanup
        pass

Step 3: Add Tests and Documentation

# Add tests
tests/adapters/test_your_framework.py

# Add example
examples/your_framework_example.py

# Update docs
docs/adapters/your_framework.md

Step 4: Submit Pull Request

• Fork the repository
• Create a feature branch (git checkout -b feature/your-adapter)
• Commit your changes (git commit -am 'Add YourFramework adapter')
• Push to the branch (git push origin feature/your-adapter)
• Open a Pull Request with a clear description

📋 Code Standards

• Style: We use black for formatting and ruff for linting
• Types: Type hints are required for public APIs
• Tests: Maintain >80% code coverage
• Docs: Docstrings for all public classes and methods

---

📚 Examples

See the examples/ directory for complete working examples:

• crewai_example.py: Integrating with CrewAI agents
• guardrail_example.py: Using the guardrail middleware directly
• custom_adapter_example.py: Building a custom adapter

---

🗺️ Roadmap

Q1 2026

• ✅ Core infrastructure (runner, guardrails, adapters)
• ✅ CrewAI, AutoGen, LangChain adapters
• ✅ OpenTelemetry integration
• ✅ CLI tools
• 🔄 Private beta launch

Q2 2026

• 🔲 Public open-source release
• 🔲 Kubernetes operator
• 🔲 Web dashboard for monitoring
• 🔲 Multi-tenant cost allocation

Q3 2026

• 🔲 Enterprise features (RBAC, SSO)
• 🔲 Advanced cost forecasting
• 🔲 Mainframe modernization tools (agent-based COBOL translation)

---

🙋 FAQ

Q: Is AgentSentry a framework for building agents?
A: No. AgentSentry is infrastructure that wraps around your existing agent frameworks to add production features like cost controls and observability.

Q: Does it work with my existing agents?
A: Yes! AgentSentry is designed to integrate with minimal code changes. If we don't support your framework yet, you can build a custom adapter.

Q: How much overhead does it add?
A: Minimal. Token tracking adds <10ms per API call. The guardrail middleware is designed for production workloads.

Q: Can I use it without a database?
A: Yes. Set enable_db_logging: false in your config. Metrics will still be tracked in memory.

Q: Is it production-ready?
A: We're currently in private beta. The core infrastructure is stable, but we recommend thorough testing before production deployment.

---

📄 License

AgentSentry is released under the MIT License. See LICENSE for details.

---

🌟 Community & Support

• GitHub Issues: Report bugs or request features
• Discussions: Ask questions and share use cases
• Discord: Join our community server (link coming soon)
• Email: [e-mail]

---

⭐ If AgentSentry solves a problem for you, please star the repo to help others discover it!

Made with ❤️ by the AgentSentry community