We are taking example of building a monitor Agent
We'll build a real Monitor Agent that you can run locally and later deploy to GCP/AWS. The goal is to make it production-oriented rather than a toy example.
Part 1 – Building a Production-Ready Monitor Agent (Step-by-Step)
Goal: Build an AI-powered, event-driven Monitor Agent that continuously monitors cloud infrastructure, detects incidents, applies deterministic rules where possible, uses an LLM only when necessary, and publishes structured incidents to downstream agents.
Overall Architecture
Cloud Infrastructure
AWS | Azure | GCP | Kubernetes | Git | Terraform
│
▼
Event Collector (FastAPI)
│
▼
Event Normalizer
│
▼
Event Correlation Engine
│
▼
Rule Engine
(Known Problems → No LLM Needed)
┌──────────┴──────────┐
│ │
▼ ▼
Incident Created LLM Reasoning
│
▼
AI Incident Analysis
└──────────┬──────────┘
▼
Incident Publisher
│
▼
Supervisor AgentFolder Structure
agentic-gitops/
│
├── monitor-agent/
│
│ ├── app/
│ │
│ │ ├── main.py # FastAPI entry point
│ │ ├── config.py # Environment variables
│ │ ├── workflow.py # LangGraph orchestration
│ │ ├── models.py # Pydantic models
│ │ ├── prompts.py # LLM prompts
│ │ │
│ │ ├── api/
│ │ │ routes.py # REST endpoints
│ │ │
│ │ ├── agents/
│ │ │ monitor.py # Main Monitor Agent
│ │ │
│ │ ├── pipeline/
│ │ │ collector.py
│ │ │ normalizer.py
│ │ │ correlator.py
│ │ │ rule_engine.py
│ │ │ llm_reasoner.py
│ │ │ publisher.py
│ │ │
│ │ ├── services/
│ │ │ incident_service.py
│ │ │
│ │ └── tools/
│ │ cloud_logging.py
│ │ monitoring.py
│ │ billing.py
│ │ terraform.py
│ │ pubsub.py
│ │
│ ├── tests/
│ │ test_rules.py
│ │ test_workflow.py
│ │
│ ├── Dockerfile
│ ├── requirements.txt
│ └── .env
│
└── terraform/Phase 1 – Project Bootstrap
Step 1 – Infrastructure Setup
terraform/
Create Service Account
monitor-agent-saIAM Permissions
Logging Viewer
Monitoring Viewer
Cloud Asset Viewer
Billing Viewer
Pub/Sub Publisher
Vertex AI UserCreate Pub/Sub Topic
incident-eventsThis topic publishes incidents to the Supervisor Agent.
Enable GCP APIs
Cloud Logging
Cloud Monitoring
Cloud Asset Inventory
Billing Export
Pub/Sub
Vertex AIStep 2 – Create Python Virtual Environment?
Step 3 – Install Dependencies
requirements.txt
fastapi
uvicorn
langgraph
langchain
langchain-google-vertexai
google-cloud-logging
google-cloud-monitoring
google-cloud-pubsub
google-cloud-asset
python-dotenv
pydanticStep-4 Bootstrap main.py
A good main.py is boring. It should read like an application's table of contents: configure the app, initialize shared resources, register middleware and routes, and start serving requests. If you find yourself writing monitoring logic, AI prompts, or cloud API calls in main.py, it's usually a sign that responsibilities need to be moved into dedicated modules.
main.py
from fastapi import FastAPI
app = FastAPI(
title="Monitor Agent",
version="0.1.0"
)
@app.get("/health")
def health():
return {
"status": "healthy"
}Run:
uvicorn app.main:app --reloadVisit:
http://localhost:8000/docsIf Swagger opens, your project is set up correctly.
Step 5 – Set Configuration
.env
PROJECT_ID=my-project
PUBSUB_TOPIC=incident-eventsconfig.py
from dotenv import load_dotenv
import os
load_dotenv()
PROJECT_ID = os.getenv("PROJECT_ID")
PUBSUB_TOPIC = os.getenv("PUBSUB_TOPIC")
MODEL = "gemini-2.5-pro"Step 6 – Add Configuration
After creating config.py:
from fastapi import FastAPI
from app.config import settings
app = FastAPI(
title=settings.APP_NAME,
version=settings.VERSION
)
@app.get("/health")
def health():
return {
"status":"healthy"
}Step 5 – Event Collector
Its only responsibility is to receive events from cloud services.
collector.py
from fastapi import FastAPI
app = FastAPI()
@app.post("/events")
async def receive_event(event: dict):
return process_event(event)Example Event
{
"service":"Cloud SQL",
"event":"Database Scaled",
"user":"john@company.com",
"cpu_before":2,
"cpu_after":16
}Step 6 – Event Normalizer
Different clouds emit different event formats.
Normalize everything into one schema.
normalizer.py
def normalize(event):
return {
"resource": event["service"],
"event_type": event["event"],
"user": event["user"],
"metadata": event
}Example Output
{
"resource":"Cloud SQL",
"event_type":"Database Scaled",
"user":"john@company.com"
}Now every downstream component understands the same structure.
Step 7 – Correlation Engine
One event rarely tells the full story.
Enrich the event by querying other systems.
Example
Cloud SQL scaled
+
Billing Export
+
Terraform State
+
Cloud Logging
+
Maintenance Calendarcorrelator.py
def correlate(event):
event["terraform_drift"] = True
event["cost_change"] = 350
event["maintenance_window"] = False
return eventNow the event contains business context.
Step 8 – Rule Engine
Do not invoke the LLM for every event.
First evaluate deterministic rules.
rule_engine.py
RULES = [
{
"name":"Terraform Drift",
"condition":lambda e: e["terraform_drift"],
"severity":"HIGH"
},
{
"name":"Cost Spike",
"condition":lambda e: e["cost_change"] > 100,
"severity":"MEDIUM"
}
]
---Evaluation---
def evaluate(event):
incidents=[]
for rule in RULES:
if rule["condition"](event):
incidents.append(rule)
return incidentsOutput
[
{
"name":"Terraform Drift",
"severity":"HIGH"
}
]No AI cost.
Step 9 – Decide Whether AI Is Needed
Simple decision logic llm_reasoner.py
def requires_llm(event, incidents):
if len(incidents)==0:
return True
if event.get("multiple_services"):
return True
if event.get("severity")=="UNKNOWN":
return True
return FalseExamples
| Event | LLM Required |
|---|---|
| Terraform Drift | ❌ |
| Public Storage Bucket | ❌ |
| CPU >95% | ❌ |
| Unknown Cost Spike | ✅ |
| Multi-service Failure | ✅ |
| New Security Incident | ✅ |
Step 10 – LLM Reasoning
Only ambiguous incidents reach the LLM.
Prompt
You are an Infrastructure Monitoring AI.
Analyze this infrastructure event.
Cloud SQL
Scaled from 2 CPUs to 16 CPUs
Terraform Drift = Yes
Cost Increase = $350/day
Maintenance Window = No
Determine:
Severity
Root Cause
Business Impact
Recommended Action
Confidence ScoreExpected Response
{
"severity":"HIGH",
"root_cause":"Manual Console Change",
"recommendation":"Restore Terraform State",
"confidence":0.97
}Step 11 – Incident Model
models.py
from pydantic import BaseModel
class Incident(BaseModel):
resource: str
severity: str
cause: str
confidence: float
recommendation: strStep 12 – Publish Incident
Convert everything into one standard incident.
incident = {
"resource":"Cloud SQL",
"severity":"HIGH",
"recommendation":"Restore Terraform"
}publisher.py
from google.cloud import pubsub_v1
publisher = pubsub_v1.PublisherClient()
publisher.publish(
topic_path,
json.dumps(incident).encode()
)The Supervisor Agent receives the same format regardless of whether the incident came from rules or AI.
Step 13 – LangGraph Workflow
START
│
▼
Collect Event
│
▼
Normalize Event
│
▼
Correlate Context
│
▼
Evaluate Rules
│
├────────────► Rule Matched
│ │
│ ▼
│ Create Incident
│
└────────────► No Match
│
▼
LLM Reasoning
│
▼
Create Incident
│
▼
Publish Incident
│
▼
ENDworkflow.py
from langgraph.graph import StateGraph
workflow = StateGraph(dict)
workflow.add_node("analyze")
workflow.set_entry_point("analyze")
graph = workflow.compile()Step 14 – Docker
Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt
CMD ["uvicorn","app.main:app","--host","0.0.0.0","--port","8080"]Step 15 – Local Testing
Run
uvicorn app.main:app --reloadOpen
http://localhost:8000/docsSubmit
{
"resource":"Cloud SQL",
"message":"Database manually scaled"
}Expected Output
{
"severity":"HIGH",
"cause":"Infrastructure Drift",
"recommendation":"Restore Terraform State",
"confidence":0.96
}Final Monitor Agent Pipeline
Cloud Events
│
▼
Event Collector
│
▼
Event Normalizer
│
▼
Correlation Engine
│
▼
Rule Engine
│
┌───┴──────────────┐
│ │
▼ ▼
Known Issue Unknown Issue
│ │
▼ ▼
Incident LLM Analysis
│ │
└──────┬───────────┘
▼
Incident Publisher
│
▼
Supervisor AgentDeliverables
By the end of Part 1, you'll have:
-
✅ FastAPI-based event ingestion
-
✅ Cloud event normalization
-
✅ Context correlation
-
✅ Deterministic rule engine
-
✅ Cost-efficient LLM reasoning for complex cases
-
✅ Structured incident model
-
✅ Pub/Sub incident publishing
-
✅ LangGraph orchestration
-
✅ Dockerized Monitor Agent
-
✅ Production-ready foundation for the Supervisor Agent
Principal Architect Design Decision: Use deterministic rules for known conditions and reserve LLM reasoning for ambiguous or novel incidents. This reduces latency and inference costs while keeping the system predictable, auditable, and scalable.
