	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > SLA Breach Alert Agent

SLA Breach Alert Agent

Author: Venkata Sudhakar

Service Level Agreements with customers promise delivery within a stated timeframe. When a shipment is at risk of breaching its SLA, the operations team needs an immediate alert to intervene - reroute, escalate to the carrier, or proactively notify the customer.

This tutorial builds a Gemini ADK agent that scans active shipments, computes hours remaining versus carrier average transit time, and raises CRITICAL, WARNING, or ON_TRACK alerts with recommended actions.

The below example shows an SLA breach monitoring agent for ShopMax India in a business context.

import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Active shipments: promised delivery in hours from now, carrier avg delay %
SHIPMENTS = [
    {"id": "SHP-301", "customer": "Rahul Verma", "city": "Mumbai",
     "product": "Apple iPad", "promised_hours": 4, "carrier": "BlueDart",
     "carrier_avg_delay_pct": 12, "status": "In Transit"},
    {"id": "SHP-302", "customer": "Meera Nair", "city": "Kochi",
     "product": "Samsung TV", "promised_hours": 2, "carrier": "Delhivery",
     "carrier_avg_delay_pct": 28, "status": "Out for Delivery"},
    {"id": "SHP-303", "customer": "Ajay Singh", "city": "Jaipur",
     "product": "Bose Speaker", "promised_hours": 18, "carrier": "Ekart",
     "carrier_avg_delay_pct": 8, "status": "In Transit"},
    {"id": "SHP-304", "customer": "Divya Menon", "city": "Hyderabad",
     "product": "Sony Headphones", "promised_hours": 1, "carrier": "XpressBees",
     "carrier_avg_delay_pct": 35, "status": "Hub Scan"},
]

def check_sla_breaches() -> dict:
    alerts = []
    for s in SHIPMENTS:
        delay_risk_hours = s["promised_hours"] * s["carrier_avg_delay_pct"] / 100
        buffer = s["promised_hours"] - delay_risk_hours
        if buffer <= 1:
            level = "CRITICAL"
            action = "Escalate to carrier account manager immediately"
        elif buffer <= 4:
            level = "WARNING"
            action = "Send proactive delay notification to customer"
        else:
            level = "ON_TRACK"
            action = "No action required"
        alerts.append({
            "shipment_id": s["id"],
            "customer": s["customer"],
            "city": s["city"],
            "product": s["product"],
            "promised_hours": s["promised_hours"],
            "carrier": s["carrier"],
            "status": s["status"],
            "sla_level": level,
            "buffer_hours": round(buffer, 1),
            "recommended_action": action,
        })
    alerts.sort(key=lambda a: a["buffer_hours"])
    critical = sum(1 for a in alerts if a["sla_level"] == "CRITICAL")
    warning  = sum(1 for a in alerts if a["sla_level"] == "WARNING")
    return {"total": len(alerts), "critical": critical, "warning": warning, "alerts": alerts}

sla_tool = FunctionTool(func=check_sla_breaches)

agent = LlmAgent(
    name="sla_alert_agent",
    model="gemini-2.0-flash",
    description="Monitors shipments for SLA breach risk and issues prioritised alerts.",
    instruction=(
        "You are an SLA monitoring agent for ShopMax India. "
        "When asked for a breach report, call check_sla_breaches and present results "
        "grouped by severity: CRITICAL first, then WARNING, then ON_TRACK. "
        "Include the recommended action for each shipment."
    ),
    tools=[sla_tool],
)

async def main():
    session_service = InMemorySessionService()
    runner = Runner(agent=agent, app_name="sla_app", session_service=session_service)
    session = await session_service.create_session(app_name="sla_app", user_id="ops1")
    msg = types.Content(role="user", parts=[types.Part(
        text="Run SLA breach check for all active shipments"
    )])
    async for event in runner.run_async(user_id="ops1", session_id=session.id, new_message=msg):
        if event.is_final_response() and event.content:
            print(event.content.parts[0].text)

asyncio.run(main())

It gives the following output,

SLA Breach Report - ShopMax India Operations

Summary: 4 shipments | 2 CRITICAL | 1 WARNING | 1 ON_TRACK

[CRITICAL] SHP-304 - Sony Headphones
  Customer : Divya Menon, Hyderabad
  Carrier  : XpressBees | Status: Hub Scan
  Promise  : 1 hr | Buffer: 0.65 hrs
  Action   : Escalate to carrier account manager immediately

[CRITICAL] SHP-302 - Samsung TV
  Customer : Meera Nair, Kochi
  Carrier  : Delhivery | Status: Out for Delivery
  Promise  : 2 hrs | Buffer: 1.44 hrs
  Action   : Escalate to carrier account manager immediately

[WARNING]  SHP-301 - Apple iPad
  Customer : Rahul Verma, Mumbai
  Carrier  : BlueDart | Status: In Transit
  Promise  : 4 hrs | Buffer: 3.52 hrs
  Action   : Send proactive delay notification to customer

[ON_TRACK] SHP-303 - Bose Speaker
  Customer : Ajay Singh, Jaipur
  Carrier  : Ekart | Status: In Transit
  Promise  : 18 hrs | Buffer: 16.56 hrs
  Action   : No action required

ShopMax India can schedule this agent to run every 30 minutes via a cron job, pushing CRITICAL alerts to a Slack operations channel and automatically triggering an SMS to affected customers. This transforms SLA management from a reactive firefight into a proactive, data-driven workflow.

Send your comments, suggestions or queries regarding this site to [email protected].