SRE reliability instrumentation for agentic AI in production.
🛡 IdempotencyGuard — block a retried write that already succeeded. DuplicateActionRate is the SLI dashboards miss.
📝 Agent Sprawl is Your Next Production Incident — LinkedIn (April 30, 2026)
📄 Governing Agent Sprawl on AWS — AWS Community Builders
✍️ Agent Sprawl: SRE Response to Datadog's State of AI Engineering 2026 — DEV Community
Your AI agent is returning HTTP 200. Uptime is 99.9%. Every health check is green.
And it's making wrong decisions 30% of the time.
Your current observability stack won't tell you.
agentsre implements the four SLIs that catch what CloudWatch, Datadog, and Grafana miss — plus an A2A semantic boundary validator, agent chain circuit breaker, and Agent Sprawl governance module for multi-model, multi-framework production deployments.
An agent can return HTTP 200 and still fail. TaskRecord feeds four semantic SLIs.
IdempotencyGuard, the A2A validator and the circuit breaker sit in front of anything
that writes or fans out to another agent. CloudWatch (optional extra) is a sink, not the source of truth.
An AI agent in a production environment ran for 6 hours making bad decisions before it caused an outage.
Every metric was green. HTTP 200 responses. 99.99% uptime. P99 latency 142ms. CloudWatch healthy. Datadog healthy. PagerDuty silent.
But the agent was failing at the semantic layer:
- Tool Invocation Efficiency: 2.1 calls/task → 6.8 (3× normal)
- Approval queue: 12 pending → 847 pending
- Decision confidence: 0.92 → 0.41
- Human rejection rate: 1.2% → 19%
None of that triggered an alert. Traditional observability measures infrastructure. AI agents fail at the semantic layer — wrong decision, right HTTP 200.
That's why this library exists.
Agent returns HTTP 200 ✓ — standard SLIs: healthy
Agent confidence drops 92% → 41% — standard SLIs: healthy
Agent making 6 tool calls instead of 2 — standard SLIs: healthy
Agent escalating 18% of tasks vs 3% — standard SLIs: healthy
| SLI | What it catches | Layer |
|---|---|---|
| Decision Quality Rate (DQR) | Semantic drift from behavioral baseline | Leading indicator |
| Tool Invocation Efficiency (TIE) | Agent compensating for degraded tools/context | Early warning |
| Human Escalation Rate (HER) | Direct operational cost of unreliability | Lagging proxy |
| Approval Queue Depth Drift (AQDD) | Human-blocked tasks; queue silently growing¹ | Missing entirely |
¹ AQDD is the failure mode standard SLO burn-rate alerts miss entirely: work submitted, never approved, queue depth growing while dashboards show nothing. Introduced in this LinkedIn discussion.
pip install agentsre # core — zero dependencies
pip install agentsre[aws] # + boto3 for CloudWatch publishingfrom agentsre import AgentSLICollector, TaskRecord
collector = AgentSLICollector()
collector.record(TaskRecord(
task_id="t-001",
task_class="payment-routing",
tool_calls=3,
required_escalation=False,
pending_approval=False,
decision_confidence=0.91, # 0.0–1.0 from your model output
completed=True,
))
for result in collector.collect("payment-routing"):
print(result)
# [DecisionQualityRate] payment-routing: 91.0% 🟢
# [ToolInvocationEfficiency] payment-routing: 3.0 calls 🟢
# [HumanEscalationRate] payment-routing: 0.0% 🟢
# [ApprovalQueueDepthDrift] payment-routing: 0 pending 🟢
if collector.breached("payment-routing"):
alert_oncall()from agentsre.cloudwatch import CloudWatchPublisher
publisher = CloudWatchPublisher(agent_id="my-agent")
publisher.publish(collector.collect("payment-routing"))
# → Namespace: AgentReliability
# → Metrics: DecisionQualityRate, ToolInvocationEfficiency,
# HumanEscalationRate, ApprovalQueueDepthDrift
# → Dimensions: AgentId=my-agent, TaskClass=payment-routingCatches the multi-agent failure mode where HTTP 200s mask semantically wrong output propagating through your agent chain.
from agentsre import A2ASemanticValidator
validator = A2ASemanticValidator(behavioral_threshold=0.75)
validator.register_schema("risk-assessment", {
"required_fields": ["risk_score", "confidence", "factors"],
"field_types": {"risk_score": (int, float), "confidence": float},
})
result = validator.validate(
task_result={"output": {"risk_score": 7.2, "confidence": 0.88, "factors": [...]}},
sub_agent_id="risk-agent-v2",
task_class="risk-assessment",
)
if not result.valid:
route_to_escalation(result) # do NOT pass to orchestratorOperates at the semantic validation success rate — not the HTTP success rate.
from agentsre import AgentChainCircuitBreaker
breaker = AgentChainCircuitBreaker(
open_threshold=85.0, # open when success rate drops below 85%
close_threshold=95.0, # close after recovery probe exceeds 95%
on_state_change=lambda s: page_oncall(s),
)
if not breaker.allow_request("risk-agent-v2", "risk-assessment"):
return degraded_mode_handler(task)
breaker.record_result("risk-agent-v2", "risk-assessment", success=validation.valid)Governs the condition where AI agent complexity grows faster than your ability to measure and govern its reliability.
from agentsre.sprawl import AgentFleetInventory, FleetComponent, ComponentType
from agentsre.sprawl import FrameworkVersionGovernance, UpgradeDecision
# Fleet inventory with SLO ownership + deprecation tracking
inventory = AgentFleetInventory()
inventory.register(FleetComponent(
component_id="anthropic.claude-sonnet-4-6",
component_type=ComponentType.MODEL,
agent_id="payment-processor",
task_classes=["payment-routing"],
slo_owner="[email protected]", # named human — not a team
baseline_established_at="2026-04-01",
deprecation_date="2027-06-01",
last_slo_review="2026-04-01",
current_tie_baseline=2.4,
current_dqr_baseline=91.2,
))
print(inventory.summary())
alerts = inventory.pending_deprecation_alerts() # fires at 60/30/7 days
report = inventory.quarterly_review_report() # P0-P3 action items
# Framework upgrade canary — blocks promotion if TIE drifts >15%
gov = FrameworkVersionGovernance(tie_drift_threshold=1.15, dqr_drift_threshold=0.85)
gov.snapshot_baseline("payment-processor", "payment-routing", "langchain-0.2.x",
tie_values=[2.1, 2.3, 2.0], dqr_values=[91.2, 89.5, 92.0])
result = gov.evaluate_upgrade("payment-processor", "payment-routing",
"langchain-0.2.x", "langchain-0.3.x")
if result.decision == UpgradeDecision.BLOCK:
rollback() # framework added hidden overhead — don't promote| SLI | Conservative | Aggressive |
|---|---|---|
| DQR | > 85% | > 92% |
| TIE | < 1.5× baseline | < 1.2× baseline |
| HER | < 5% | < 2% |
| AQDD | < 2× baseline | < 1.5× baseline |
Rule: Run a 30-day observation window before committing to any SLO target. You cannot commit to reliability you have not yet measured.
Level 1 — DQR drops 15% or TIE hits 1.5× baseline
Enhanced logging. No autonomy reduction.
Level 2 — DQR drops 25% or TIE hits 2× baseline
Human approval required for write operations.
Level 3 — HER exceeds 2× target
Read-only mode. All writes require explicit authorization.
Level 4 — Level 3 sustained for 30+ minutes
Suspend autonomous operation. Page on-call.
Full implementation guides on AWS Community Builders:
→ Governing Agent Sprawl on AWS: Fleet Inventory, Framework Canary, Deprecation Alerting
Key AWS services:
- CloudWatch Custom Metrics — DQR, TIE, HER, AQDD per task class + per model
- DynamoDB — behavioral baseline store + agent fleet inventory
- EventBridge — breach-triggered investigation workflows + deprecation alerts
- SSM Automation — progressive autonomy constraint ladder
- X-Ray — distributed tracing across A2A agent boundaries
- CodePipeline — framework upgrade canary gate
LinkedIn:
- Agent Sprawl is Your Next Production Incident (April 30, 2026)
- A2A + MCP in Production: The SRE Reliability Framework Nobody Has Written Yet
- SLOs for Agentic AI: The Reliability Framework Production Teams Are Missing
DEV Community:
- Agent Sprawl: SRE Response to Datadog's State of AI Engineering 2026
- Why SRE Principles Are the Missing Layer in MCP Security
AWS Community Builders:
PRs welcome — especially:
- Alternative cloud implementations — GCP (Cloud Monitoring), Azure (Application Insights)
- Framework integrations — LangChain, CrewAI, AutoGen, Amazon Bedrock Agents, LlamaIndex
- Prometheus exporter —
/metricsendpoint for self-hosted deployments - Additional task-class baseline examples
See CONTRIBUTING.md to get started.
An agent skill that teaches coding agents how to use this library is in skills/agentsre/SKILL.md.
MIT © Ajay Devineni
If this helped you instrument your agents in production, a ⭐ means a lot — it helps other SRE practitioners find this library.
Production cost tracking for AI agents.
Features:
- Real-time cost tracking (OpenAI, Anthropic models)
- Cost per agent, cost per task
- Optimization recommendations
- Budget alerts
Quick start:
from agentsre.cost_optimizer import CostTracker, CostOptimizer
tracker = CostTracker()
# Track API calls
tracker.track_api_call(
agent_id="payment-router",
model="claude-opus",
input_tokens=1000,
output_tokens=500
)
# Get metrics
metrics = tracker.get_metrics("payment-router")
print(f"Cost per call: ${metrics.cost_per_call():.4f}")
# Get suggestions
optimizer = CostOptimizer(tracker)
suggestions = optimizer.analyze()| Google Component | agentsre Module | Description |
|---|---|---|
| Actus (execution guardrails) | pre_action_gate.py |
Error budget + AQDD + HER gate before action |
| IRM Analyzer (eval pipelines) | eval_pipeline.py |
DQR trend + regression detection |
| AI Operator (agent governance) | aro.py + sprawl_registry.py |
ARO registration + fleet PRR |
| Human operational memory | reasoning_trace.py |
RTD traces as ground truth corpus |
Reference: https://sre.google/resources/practices-and-processes/ai-engineering-reliable-operations/