AI Agent Deployment Checklist: Security, Scaling, and Monitoring

AI Agent Deployment Checklist: Security, Scaling, and Monitoring

Deploying AI agents into production is fundamentally different from shipping traditional software. These agents operate autonomously, consume unpredictable amounts of compute, interact with external tools, and often handle sensitive data across multiple trust boundaries. A single oversight can lead to prompt injection vulnerabilities, runaway resource consumption, or silent failures that go undetected for days. This tutorial provides a comprehensive, actionable checklist covering the three pillars of production-grade AI agent deployment: security, scaling, and monitoring.

What Is an AI Agent Deployment Checklist?

An AI agent deployment checklist is a structured set of requirements, configurations, and validations that engineering teams apply before and during the rollout of autonomous agent systems. It goes far beyond generic application deployment—it addresses agent-specific risks such as tool-calling permission boundaries, recursive agent loops, model output validation, and the observability of multi-step reasoning chains. The checklist functions as both a pre-flight verification and a continuous runtime safeguard.

Why It Matters

AI agents bridge the gap between deterministic software and non-deterministic language models. This hybrid nature introduces failure modes that traditional deployment practices do not cover. Without a specialized checklist, teams risk exposing internal APIs to prompt-injected instructions, exceeding infrastructure budgets due to unbounded agent loops, or failing to detect when agents silently produce incorrect outputs. A disciplined deployment checklist reduces incident frequency, accelerates root-cause analysis, and ensures compliance with data governance requirements.

Security Checklist

Security for AI agents spans multiple layers: the inbound prompt interface, the agent's internal reasoning boundary, and every external tool or API the agent is permitted to invoke. Each layer must enforce strict isolation and validation.

1. Authentication & Authorization for Agent Endpoints

Every endpoint that accepts user input destined for an AI agent must enforce authentication. Beyond user identity, implement identity propagation so that tools called by the agent operate under the same user's permissions—never under a privileged service account that the agent shares across all users.

• Use OAuth2 or API key validation on the agent's entry point
• Propagate user context through the agent's tool-calling chain
• Apply the principle of least privilege to every tool binding
• Rotate agent-specific API keys on a regular schedule

# Example: Propagating user identity to agent tools (Python)
class AgentToolExecutor:
    def __init__(self, user_context: UserContext):
        self.user_id = user_context.user_id
        self.permissions = user_context.permissions
    
    def execute_tool(self, tool_name: str, params: dict) -> dict:
        if tool_name not in self.permissions.allowed_tools:
            raise PermissionError(f"User {self.user_id} lacks access to {tool_name}")
        # All downstream calls use user-scoped credentials
        return self._invoke_with_user_scope(tool_name, params)

2. Input Sanitization & Prompt Injection Prevention

Prompt injection is the most critical vector for AI agents. Attackers embed instructions that override the agent's system prompt, causing it to execute unintended tool calls or leak data. Defense requires layered mitigations, as no single technique is foolproof.

• Apply input sanitization: strip control characters, enforce length limits, reject suspicious patterns
• Use a hardened system prompt that explicitly instructs the model to distinguish user data from commands
• Implement input fencing: wrap user input in delimiter markers that the model is trained to respect
• Validate all tool call parameters before execution—never pass raw model output directly to tools
• Run a secondary classifier that scores user input for injection attempts before processing

# Example: Input fencing and tool parameter validation
def fence_user_input(raw_input: str) -> str:
    # Wrap user data in explicit boundary markers
    fenced = f"<user_input>\n{raw_input.strip()}\n</user_input>"
    return fenced

def validate_tool_parameters(tool_name: str, params: dict) -> dict:
    # Whitelist-based validation for each tool
    SCHEMAS = {
        "search_docs": {"query": str, "max_results": int},
        "send_email": {"to": str, "subject": str, "body": str},
    }
    schema = SCHEMAS.get(tool_name)
    if not schema:
        raise ValueError(f"Unknown tool: {tool_name}")
    validated = {}
    for key, expected_type in schema.items():
        if key not in params:
            raise ValueError(f"Missing parameter: {key}")
        if not isinstance(params[key], expected_type):
            raise TypeError(f"Parameter {key} must be {expected_type}")
        validated[key] = params[key]
    # Additional semantic checks
    if tool_name == "send_email":
        if len(params["to"]) > 254 or "@" not in params["to"]:
            raise ValueError("Invalid email address")
    return validated

3. Data Encryption & Secrets Management

AI agents frequently handle sensitive data across multiple services. Encryption must cover data at rest, data in transit, and—critically—data in memory during agent reasoning cycles. Secrets such as model provider API keys, database credentials, and third-party service tokens must never appear in logs, prompts, or agent memory.

• Use TLS 1.3 for all inter-service communication
• Store secrets in a dedicated secrets manager (HashiCorp Vault, AWS Secrets Manager, etc.)
• Inject secrets at runtime via environment variables or mounted volumes—never hardcode
• Redact secrets from agent logs using pattern-matching scrubbers
• Encrypt agent state and memory if persisted to disk

# Example: Secrets injection and log scrubbing
import os
import re
import logging
from typing import Any

class SecretScrubber(logging.Filter):
    SENSITIVE_PATTERNS = [
        r'(?i)(api[_-]?key\s*[:=]\s*)[^\s]+',
        r'(?i)(authorization:\s*Bearer\s*)[^\s]+',
        r'(?i)(sk-[a-zA-Z0-9]{32,})',
    ]
    
    def filter(self, record: logging.LogRecord) -> bool:
        if hasattr(record, 'msg'):
            for pattern in self.SENSITIVE_PATTERNS:
                record.msg = re.sub(pattern, r'\1[REDACTED]', str(record.msg))
        return True

# Configure logging with scrubbing
logger = logging.getLogger("agent")
logger.addFilter(SecretScrubber())

# Load secrets from environment (set via secrets manager at deploy time)
MODEL_API_KEY = os.environ.get("MODEL_API_KEY")
DB_PASSWORD = os.environ.get("DB_PASSWORD")

4. Rate Limiting & Abuse Prevention

AI agents amplify compute costs—each user request may trigger multiple model calls and tool invocations. Without rate limiting, a single abusive user or a runaway agent loop can exhaust quotas and incur significant financial damage.

• Apply token-bucket rate limiting on the agent entry point (per-user and per-IP)
• Cap the maximum number of reasoning steps per session
• Set hard timeouts on individual tool calls and overall agent execution
• Implement circuit breakers that halt agent execution when tool failures exceed a threshold
• Monitor and alert on anomalous usage spikes

# Example: Comprehensive rate limiting middleware (FastAPI)
from fastapi import FastAPI, Request, HTTPException
from datetime import datetime
import hashlib
import time

class AgentRateLimiter:
    def __init__(self, redis_client, max_requests_per_minute: int = 30):
        self.redis = redis_client
        self.max_requests = max_requests_per_minute
        self.window_seconds = 60
    
    async def check(self, request: Request) -> bool:
        user_id = request.headers.get("X-User-ID", request.client.host)
        key = f"rate_limit:agent:{hashlib.sha256(user_id.encode()).hexdigest()[:16]}"
        current = int(time.time() / self.window_seconds)
        window_key = f"{key}:{current}"
        
        count = self.redis.incr(window_key)
        if count == 1:
            self.redis.expire(window_key, self.window_seconds)
        
        if count > self.max_requests:
            raise HTTPException(
                status_code=429,
                detail="Rate limit exceeded. Please reduce request frequency."
            )
        return True

# Agent execution bounds
MAX_AGENT_STEPS = 20
TOOL_TIMEOUT_SECONDS = 30
AGENT_EXECUTION_TIMEOUT = 300  # 5 minutes total

5. Complete Security Middleware Example

The following example demonstrates a production-ready middleware stack for an AI agent endpoint, combining authentication, input fencing, rate limiting, and tool parameter validation into a single request pipeline.

# Complete security pipeline for AI agent requests (Python/FastAPI)
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any
import time, re, hashlib, logging

app = FastAPI()
security = HTTPBearer()
logger = logging.getLogger("agent_security")

# ---------- Data Models ----------
class AgentRequest(BaseModel):
    user_message: str = Field(..., max_length=4000)
    session_id: Optional[str] = None
    context: Optional[Dict[str, Any]] = None

class AgentResponse(BaseModel):
    agent_reply: str
    tool_calls_made: int
    execution_time_ms: float

# ---------- Injection Detection ----------
INJECTION_PATTERNS = [
    r'(?i)ignore\s+(all\s+)?(previous|above|prior)\s+(instructions|directives|commands)',
    r'(?i)you\s+are\s+(now|no\s+longer)\s+(a|an)\s+(assistant|helper)',
    r'(?i)system\s*(prompt|message|instruction)\s*[:=]',
    r'[<>].*system.*[<>]',
]

def detect_injection(text: str) -> float:
    score = 0.0
    for pattern in INJECTION_PATTERNS:
        matches = re.findall(pattern, text, re.IGNORECASE)
        score += len(matches) * 0.3
    return min(score, 1.0)

# ---------- Input Sanitization ----------
def sanitize_input(text: str) -> str:
    text = text.replace('\x00', '')  # Remove null bytes
    text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]', '', text)
    return text.strip()

def fence_user_content(text: str) -> str:
    return f"\n<authenticated_user_message>\n{text}\n</authenticated_user_message>\n"

# ---------- Tool Parameter Validator ----------
class ToolValidator:
    ALLOWED_TOOLS = {
        "read_file": {"path": str, "max_lines": int},
        "search_kb": {"query": str, "top_k": int},
        "create_ticket": {"title": str, "priority": str},
    }
    
    @classmethod
    def validate(cls, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        if tool_name not in cls.ALLOWED_TOOLS:
            raise HTTPException(status_code=403, detail=f"Tool '{tool_name}' not permitted")
        schema = cls.ALLOWED_TOOLS[tool_name]
        validated = {}
        for param, expected_type in schema.items():
            if param not in arguments:
                raise HTTPException(status_code=400, detail=f"Missing '{param}' for tool '{tool_name}'")
            value = arguments[param]
            if not isinstance(value, expected_type):
                raise HTTPException(status_code=400, detail=f"'{param}' must be {expected_type.__name__}")
            validated[param] = value
        return validated

# ---------- Main Endpoint ----------
@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(
    request: Request,
    body: AgentRequest,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    start_time = time.time()
    
    # Step 1: Authenticate user
    user_id = verify_token(credentials.credentials)  # Your token verification logic
    
    # Step 2: Rate limit check
    await rate_limiter.check(request)
    
    # Step 3: Injection detection
    injection_score = detect_injection(body.user_message)
    if injection_score > 0.7:
        logger.warning(f"Injection detected for user {user_id}, score={injection_score}")
        raise HTTPException(status_code=400, detail="Input contains suspicious patterns")
    
    # Step 4: Sanitize and fence input
    clean_input = sanitize_input(body.user_message)
    fenced_input = fence_user_content(clean_input)
    
    # Step 5: Execute agent with validated tool calling
    agent = build_agent(user_id=user_id, tool_validator=ToolValidator.validate)
    result = await agent.execute(
        user_input=fenced_input,
        max_steps=MAX_AGENT_STEPS,
        timeout=AGENT_EXECUTION_TIMEOUT,
        session_id=body.session_id
    )
    
    execution_time = (time.time() - start_time) * 1000
    
    return AgentResponse(
        agent_reply=result.final_output,
        tool_calls_made=result.tool_call_count,
        execution_time_ms=execution_time
    )

Scaling Checklist

AI agents exhibit bursty, compute-intensive workloads. A single user query can trigger dozens of model calls and tool invocations. Scaling infrastructure to handle this requires careful architectural decisions around state management, concurrency, and resource isolation.

1. Stateless vs. Stateful Agent Architecture

The most consequential scaling decision is whether agent sessions are stateless or stateful. Stateless architectures are easier to scale horizontally but require externalizing all session state. Stateful architectures reduce latency but complicate failover and load distribution.

• Prefer stateless agents with externalized session storage (Redis, database)
• If using stateful agents (e.g., WebSocket-based), implement sticky sessions with graceful drain on shutdown
• Serialize agent state to allow resumption on a different node if the original node fails
• Keep session payloads compact—avoid storing full conversation histories in memory

# Example: Externalized session state for stateless agent scaling
import json
import pickle
from datetime import datetime, timedelta

class SessionStore:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = timedelta(hours=24)
    
    def save(self, session_id: str, agent_state: dict) -> None:
        key = f"agent:session:{session_id}"
        serialized = json.dumps(agent_state, default=str)
        self.redis.setex(key, int(self.ttl.total_seconds()), serialized)
    
    def load(self, session_id: str) -> dict | None:
        key = f"agent:session:{session_id}"
        data = self.redis.get(key)
        if data:
            return json.loads(data)
        return None
    
    def delete(self, session_id: str) -> None:
        self.redis.delete(f"agent:session:{session_id}")

# Agent factory that reconstructs state from store
async def build_stateless_agent(session_id: str, session_store: SessionStore):
    cached_state = session_store.load(session_id)
    if cached_state:
        agent = Agent.rehydrate(cached_state)
    else:
        agent = Agent.create_fresh()
    return agent

2. Load Balancing & Horizontal Scaling

Agent workloads are CPU-bound (model inference orchestration) and I/O-bound (tool API calls). Use load balancers with health checks that account for agent-specific readiness signals.

• Deploy agent workers behind a Layer 7 load balancer with path-based routing
• Implement health checks that verify model provider connectivity and tool endpoint reachability
• Use Kubernetes horizontal pod autoscaling based on custom metrics (active agent sessions, queue depth)
• Configure pod disruption budgets to ensure graceful draining of in-flight agent executions

# Example: Kubernetes deployment with custom HPA metrics for agent workers
# k8s-agent-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-worker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-worker
  template:
    metadata:
      labels:
        app: agent-worker
    spec:
      containers:
      - name: agent-worker
        image: myregistry/agent-worker:latest
        env:
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: redis-url
        - name: MODEL_API_KEY
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: model-api-key
        resources:
          requests:
            cpu: "2"
            memory: "4Gi"
          limits:
            cpu: "4"
            memory: "8Gi"
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 30
        livenessProbe:
          httpGet:
            path: /health/live
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 60
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-worker
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: Pods
    pods:
      metric:
        name: active_agent_sessions
      target:
        type: AverageValue
        averageValue: "10"
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

3. Caching Strategies for Model Responses & Tool Results

AI agent workloads benefit enormously from caching. Identical tool calls, repeated reasoning patterns, and similar user queries can be served from cache, dramatically reducing latency and cost.

• Cache LLM responses keyed by deterministic prompt hash (semantic caching)
• Cache tool call results with appropriate TTLs based on data freshness requirements
• Use Redis or Memcached for hot-path caching with sub-millisecond latency
• Implement cache invalidation hooks tied to data update events

# Example: Multi-level caching for AI agent operations
import hashlib
import asyncio
from functools import wraps
from typing import Callable, Any

class AgentCache:
    def __init__(self, redis_client, local_ttl: int = 60, remote_ttl: int = 3600):
        self.redis = redis_client
        self.local_cache: dict = {}
        self.local_ttl = local_ttl
        self.remote_ttl = remote_ttl
    
    def cache_key(self, *args, **kwargs) -> str:
        raw = f"{str(args)}:{str(sorted(kwargs.items()))}"
        return f"agent:cache:{hashlib.sha256(raw.encode()).hexdigest()[:32]}"
    
    async def get_or_set(self, key: str, fetch_fn: Callable, ttl: int = None) -> Any:
        ttl = ttl or self.remote_ttl
        
        # Level 1: Local in-memory cache
        if key in self.local_cache:
            entry, timestamp = self.local_cache[key]
            if (asyncio.get_event_loop().time() - timestamp) < self.local_ttl:
                return entry
        
        # Level 2: Redis remote cache
        cached = self.redis.get(key)
        if cached:
            result = json.loads(cached)
            self.local_cache[key] = (result, asyncio.get_event_loop().time())
            return result
        
        # Level 3: Compute and store
        result = await fetch_fn()
        serialized = json.dumps(result, default=str)
        self.redis.setex(key, ttl, serialized)
        self.local_cache[key] = (result, asyncio.get_event_loop().time())
        return result

# Usage with tool calls
async def cached_tool_execution(tool_name: str, params: dict) -> dict:
    cache = get_agent_cache()  # Singleton
    key = cache.cache_key(tool_name, params)
    
    async def execute():
        return await actual_tool_call(tool_name, params)
    
    return await cache.get_or_set(key, execute, ttl=300)

4. Database & Storage Scaling for Agent Data

Agent systems generate substantial metadata: conversation histories, reasoning traces, tool call logs, and evaluation metrics. This data grows rapidly and must be stored efficiently.

• Partition agent logs by time and user for efficient querying and retention management
• Use column-oriented databases (ClickHouse, BigQuery) for analytical queries on agent metrics
• Implement TTL-based automatic cleanup for ephemeral agent session data
• Compress stored conversation histories—use message deduplication and delta encoding

# Example: Partitioned agent event storage with automatic cleanup
CREATE TABLE agent_events (
    event_id UUID DEFAULT gen_random_uuid(),
    user_id VARCHAR(255) NOT NULL,
    session_id UUID NOT NULL,
    event_type VARCHAR(50) NOT NULL,  -- 'user_message', 'tool_call', 'model_response', 'error'
    payload JSONB NOT NULL,
    timestamp TIMESTAMPTZ DEFAULT now(),
    execution_time_ms INTEGER,
    token_count INTEGER
) PARTITION BY RANGE (timestamp);

-- Create monthly partitions automatically
CREATE TABLE agent_events_2025_01 PARTITION OF agent_events
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');
CREATE TABLE agent_events_2025_02 PARTITION OF agent_events
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

-- Index for session-based queries
CREATE INDEX idx_agent_events_session ON agent_events (session_id, timestamp);

-- Automatic retention: drop partitions older than 90 days
-- Run via scheduled cron or pg_cron extension
SELECT format('DROP TABLE IF EXISTS %I', tablename)
FROM pg_tables
WHERE schemaname = 'public'
  AND tablename LIKE 'agent_events_%'
  AND tablename < 'agent_events_' || to_char(now() - interval '90 days', 'YYYY_MM');

Monitoring Checklist

Monitoring AI agents requires observability into three distinct layers: the model provider's performance, the agent's internal reasoning process, and the downstream tools the agent invokes. Traditional application monitoring covers only the last of these.

1. Structured Logging Infrastructure

Agent logs must capture the full reasoning trace—not just final outputs. Structured logging enables searching, alerting, and cost attribution across multi-step agent executions.

• Log every agent step with a correlation ID that ties together all model calls, tool invocations, and intermediate thoughts
• Structure logs as JSON for ingest by log aggregation systems (Elasticsearch, Loki, Datadog)
• Include latency, token count, and cost metadata in each log entry
• Log at decision points: which tool was selected, why, and with what parameters

# Example: Structured agent logging with correlation IDs
import uuid
import json
import time
from contextvars import ContextVar

correlation_id_var: ContextVar[str] = ContextVar('correlation_id', default='')
session_id_var: ContextVar[str] = ContextVar('session_id', default='')

class AgentLogger:
    def __init__(self, logger: logging.Logger):
        self.logger = logger
    
    def log_step(self, step_number: int, event_type: str, data: dict):
        entry = {
            "correlation_id": correlation_id_var.get(),
            "session_id": session_id_var.get(),
            "step": step_number,
            "event_type": event_type,
            "timestamp": time.time(),
            "data": data,
        }
        self.logger.info(json.dumps(entry))
    
    def log_tool_call(self, step: int, tool_name: str, params: dict, result: dict, latency_ms: float):
        self.log_step(step, "tool_call", {
            "tool": tool_name,
            "parameters": params,
            "result_summary": str(result)[:200],
            "latency_ms": latency_ms,
        })
    
    def log_model_call(self, step: int, prompt_tokens: int, completion_tokens: int, latency_ms: float, cost: float):
        self.log_step(step, "model_inference", {
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
            "latency_ms": latency_ms,
            "estimated_cost_usd": round(cost, 6),
        })
    
    def log_error(self, step: int, error_type: str, message: str, context: dict):
        self.log_step(step, "error", {
            "error_type": error_type,
            "message": message,
            "context": context,
        })

# Usage in agent loop
async def agent_reasoning_loop(agent, user_input: str, session_id: str):
    cid = str(uuid.uuid4())
    correlation_id_var.set(cid)
    session_id_var.set(session_id)
    agent_log = AgentLogger(logging.getLogger("agent"))
    
    for step in range(MAX_AGENT_STEPS):
        start = time.time()
        response = await agent.llm.generate(...)
        latency = (time.time() - start) * 1000
        agent_log.log_model_call(
            step, 
            prompt_tokens=response.usage.prompt_tokens,
            completion_tokens=response.usage.completion_tokens,
            latency_ms=latency,
            cost=calculate_cost(response.usage)
        )
        # ... tool selection and execution with similar logging ...

2. Metrics & Observability Stack

Metrics provide real-time visibility into agent health, performance, and cost. A robust observability stack combines application metrics, infrastructure metrics, and business metrics specific to agent operations.

• Expose Prometheus metrics from agent workers: request count, step count, tool call latency, error rate
• Track LLM provider metrics: tokens per request, latency percentiles, rate limit remaining
• Monitor tool dependencies: success rate, latency, timeout frequency per external API
• Build dashboards that correlate agent behavior with infrastructure load

# Example: Prometheus metrics instrumentation for AI agent workers
from prometheus_client import Counter, Histogram, Gauge, Summary, CollectorRegistry
import time

registry = CollectorRegistry()

# Counters
agent_requests_total = Counter(
    'agent_requests_total', 
    'Total agent requests processed',
    ['status', 'agent_type'], 
    registry=registry
)
agent_steps_total = Counter(
    'agent_steps_total',
    'Total reasoning steps executed',
    ['agent_type'],
    registry=registry
)
tool_calls_total = Counter(
    'agent_tool_calls_total',
    'Total tool invocations',
    ['tool_name', 'status'],
    registry=registry
)

# Histograms for latency distribution
agent_request_duration = Histogram(
    'agent_request_duration_seconds',
    'End-to-end agent request latency',
    ['agent_type'],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0, 120.0],
    registry=registry
)
tool_call_duration = Histogram(
    'agent_tool_call_duration_seconds',
    'Tool execution latency',
    ['tool_name'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0],
    registry=registry
)
llm_call_duration = Histogram(
    'agent_llm_call_duration_seconds',
    'LLM provider latency',
    ['model_name'],
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 40.0],
    registry=registry
)

# Gauges for current state
active_sessions = Gauge(
    'agent_active_sessions',
    'Number of active agent sessions',
    registry=registry
)
llm_rate_limit_remaining = Gauge(
    'agent_llm_rate_limit_remaining',
    'Remaining LLM provider rate limit tokens',
    ['model_name'],
    registry=registry
)

# Token usage summary
token_usage_summary = Summary(
    'agent_token_usage_per_request',
    'Token consumption per agent request',
    ['model_name'],
    registry=registry
)

# Usage decorator
def instrument_agent_request(agent_type: str):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            start = time.time()
            active_sessions.inc()
            try:
                result = await func(*args, **kwargs)
                agent_requests_total.labels(status='success', agent_type=agent_type).inc()
                return result
            except Exception as e:
                agent_requests_total.labels(status='error', agent_type=agent_type).inc()
                raise
            finally:
                duration = time.time() - start
                agent_request_duration.labels(agent_type=agent_type).observe(duration)
                active_sessions.dec()
        return wrapper
    return decorator

3. Alerting & Incident Response

Effective alerting for AI agents requires thresholds tuned to the agent's specific cost and reliability profile. Alerts should trigger on anomalies in token consumption, tool failure rates, and execution loop counts—not just on generic error rates.

• Alert when token consumption per request exceeds the 95th percentile by 3x (indicating runaway loops)
• Alert on tool call failure rate exceeding 10% over a 5-minute window
• Alert when model provider rate limits are approaching exhaustion
• Create runbooks for common failure modes: injection detected, tool API outage, model unavailability
• Implement automated circuit breaking: if tool failures exceed threshold, temporarily disable that tool binding

# Example: Alerting rules for AI agent metrics (Prometheus alert rules)
# prometheus-alerts.yaml
groups:
  - name: agent_alerts
    rules:
      - alert: AgentRunawayLoop
        expr: |
          rate(agent_steps_total[5m]) > 100
        for: 2m
        labels:
          severity: critical
          component: agent
        annotations:
          summary: "Agent step rate abnormally high"
          description: "Agent {{ $labels.agent_type }} is executing {{ $value }} steps/sec, possible runaway loop"
          runbook: "https://runbooks.internal/agent-runaway-loop"
      
      - alert: ToolCallFailureRate
        expr: |
          (sum(rate(agent_tool_calls_total{status='failure'}[5m])) by (tool_name) /
           sum(rate(agent_tool_calls_total[5m])) by (tool_name)) > 0.1
        for: 5m
        labels:
          severity: warning
          component: agent
        annotations:
          summary: "Tool {{ $labels.tool_name }} failure rate > 10%"
          description: "Failure rate is {{ $value | humanizePercentage }} over 5min"
          runbook: "https://runbooks.internal/tool-failure"
      
      - alert: LLMRateLimitApproaching
        expr: |
          agent_llm_rate_limit_remaining < 100
        for: 1m
        labels:
          severity: warning
          component: llm-provider
        annotations:
          summary: "LLM rate limit nearly exhausted"
          description: "Model {{ $labels.model_name }} has only {{ $value }} tokens remaining"
      
      - alert: AgentRequestLatencySpike
        expr: |
          histogram_quantile(0.95, rate(agent_request_duration_seconds_bucket[5m])) > 60
        for: 3m
        labels:
          severity: warning
          component: agent
        annotations:
          summary: "P95 agent latency > 60s"
          description: "95th percentile latency is {{ $value }}s, may indicate degraded model or tool service"

4. Complete Monitoring Dashboard Query Examples

Below are ready-to-use queries for building observability dashboards that track agent health across all dimensions.

<