[Python] How we can create team bot for Azure monitoring for SRE team

Bot code to monitor azure resources can we use it for teams?

Summary: Azure SRE Assistant Telegram Bot This is a Python-based intelligent SRE chatbot that runs on Telegram and integrates with:

Azure Monitor for real-time metrics and alert diagnostics

Azure Resource Graph for listing and querying resources

Azure OpenAI for answering natural-language SRE queries

Telegram Bot API for user interaction

It uses Service Principal credentials to access Azure securely and provides asynchronous responses for better performance using asyncio.

import os
import re
import logging
import asyncio
from datetime import datetime, timezone, timedelta

from telegram import Update
from telegram.ext import (
ApplicationBuilder,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
)

from azure.identity.aio import ClientSecretCredential as AsyncClientSecretCredential
from azure.identity import ClientSecretCredential as SyncClientSecretCredential

from azure.mgmt.monitor.aio import MonitorManagementClient
from azure.mgmt.monitor import MonitorManagementClient as SyncMonitorClient
from azure.mgmt.resource import ResourceManagementClient

from openai import AsyncAzureOpenAI
from dotenv import load_dotenv

load_dotenv()

# Load environment variables
TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
AZURE_TENANT_ID = os.getenv("AZURE_TENANT_ID")
AZURE_CLIENT_ID = os.getenv("AZURE_CLIENT_ID")
AZURE_CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET")
AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT")
AZURE_OPENAI_KEY = os.getenv("AZURE_OPENAI_KEY")
AZURE_OPENAI_DEPLOYMENT_NAME = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME")

# Setup logging
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)

# Global async and sync credentials
async_credential = AsyncClientSecretCredential(
tenant_id=AZURE_TENANT_ID,
client_id=AZURE_CLIENT_ID,
client_secret=AZURE_CLIENT_SECRET,
)
sync_credential = SyncClientSecretCredential(
tenant_id=AZURE_TENANT_ID,
client_id=AZURE_CLIENT_ID,
client_secret=AZURE_CLIENT_SECRET,
)

# Initialize OpenAI client once
openai_client = AsyncAzureOpenAI(
api_key=AZURE_OPENAI_KEY,
api_version="2024-02-15-preview",
azure_endpoint=AZURE_OPENAI_ENDPOINT,
)

# In-memory user subscriptions store
user_subscriptions = {}


def escape_markdown(text: str) -> str:
"""
Escape Telegram MarkdownV2 special characters.
"""
escape_chars = r'_*[]()~`>#+-=|{}.!'
return re.sub(f'([{re.escape(escape_chars)}])', r'\\\1', text)


def get_monitor_client(subscription_id):
return MonitorManagementClient(async_credential, subscription_id)


def get_sync_monitor_client(subscription_id):
return SyncMonitorClient(sync_credential, subscription_id)


def get_resource_client(subscription_id):
return ResourceManagementClient(sync_credential, subscription_id)


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
user_subscriptions.pop(user_id, None)
await update.message.reply_text(
"Hello! I am your Azure SRE Assistant Bot.\n"
"Please enter your Azure Subscription ID to start."
)


async def subscription_input_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
subscription_id = update.message.text.strip()

if len(subscription_id) < 10: # crude validation, you can improve it
await update.message.reply_text(
"That does not look like a valid subscription ID. Please try again."
)
return

user_subscriptions[user_id] = subscription_id

# IMPORTANT: No MarkdownV2 parse_mode here, to avoid parse errors with dots and special chars.
msg = (
f"Subscription ID set to {subscription_id}.\n"
"You can now use the following commands:\n"
"/check - Get recent alerts\n"
"/ask <question> - Ask SRE questions\n"
"/health - Overall health summary\n"
"/memory <resource_id> - Memory usage\n"
"/cpu <resource_id> - CPU usage\n"
"/listresources - List Azure resources"
)
await update.message.reply_text(msg)


def get_user_subscription(user_id):
return user_subscriptions.get(user_id)


async def check(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
subscription_id = get_user_subscription(user_id)
if not subscription_id:
await update.message.reply_text("Please set your subscription ID first by sending /start.")
return

await update.message.reply_text("Fetching recent Azure Monitor alerts...")

try:
monitor_client = get_monitor_client(subscription_id)
now = datetime.now(timezone.utc)
one_hour_ago = now - timedelta(hours=1)

start_time = one_hour_ago.isoformat()
end_time = now.isoformat()

messages = []

# Activity Log Alerts (async)
filter_query = (
f"eventTimestamp ge '{start_time}' and eventTimestamp le '{end_time}' and category eq 'Alerts'"
)
logs_paged = monitor_client.activity_logs.list(filter=filter_query)
alert_events = [log async for log in logs_paged]

for log in alert_events:
msg = (
f" *Activity Log Alert:* {escape_markdown(log.operation_name.localized_value)}\n"
f" *Resource:* {escape_markdown(log.resource_id)}\n"
f" *Time:* {escape_markdown(str(log.event_timestamp))}\n"
f"️ *Status:* {escape_markdown(log.status.localized_value if log.status else 'N/A')}\n"
f" *Description:* {escape_markdown(log.description or 'N/A')}"
)
messages.append(msg)

# Metric Alerts and Scheduled Query Rules (sync)
sync_monitor_client = get_sync_monitor_client(subscription_id)

def fetch_metric_alerts():
results = []
try:
for alert in sync_monitor_client.metric_alerts.list_by_subscription():
results.append(alert)
except Exception as e:
logger.error(f"Error fetching metric alerts: {e}")
return results

def fetch_scheduled_query_rules():
results = []
try:
for sqr in sync_monitor_client.scheduled_query_rules.list_by_subscription():
results.append(sqr)
except Exception as e:
logger.error(f"Error fetching scheduled query rules: {e}")
return results

metric_alerts = await asyncio.to_thread(fetch_metric_alerts)
scheduled_query_rules = await asyncio.to_thread(fetch_scheduled_query_rules)

for ma in metric_alerts:
state = getattr(ma, "enabled", False)
name = getattr(ma, "name", "N/A")
description = getattr(ma, "description", "")
severity = getattr(ma, "severity", "N/A")
scopes = getattr(ma, "scopes", [])
msg = (
f" *Metric Alert:* {escape_markdown(name)}\n"
f" *Severity:* {escape_markdown(str(severity))}\n"
f" *Scopes:* {escape_markdown(', '.join(scopes)) if scopes else 'N/A'}\n"
f" *Description:* {escape_markdown(description or 'N/A')}\n"
f" *Enabled:* {escape_markdown(str(state))}"
)
messages.append(msg)

for sqr in scheduled_query_rules:
name = getattr(sqr, "name", "N/A")
desc = getattr(sqr, "description", "")
enabled = getattr(sqr, "enabled", False)
query = getattr(sqr, "source", {}).get("query", "N/A") if hasattr(sqr, "source") else "N/A"
msg = (
f"*Scheduled Query Alert:* {escape_markdown(name)}\n"
f"*Description:* {escape_markdown(desc or 'N/A')}\n"
f"*Query:* {escape_markdown(query)}\n"
f"*Enabled:* {escape_markdown(str(enabled))}"
)
messages.append(msg)

if not messages:
await update.message.reply_text("No new alerts found in the last hour.")
return

# Send messages in chunks (5 per message)
for i in range(0, len(messages), 5):
chunk_msg = "\n\n".join(messages[i : i + 5])
await update.message.reply_text(chunk_msg, parse_mode="MarkdownV2")

await monitor_client.close()

except Exception as e:
logger.error(f"Error fetching alerts: {e}", exc_info=True)
await update.message.reply_text(f"Failed to fetch alerts: {e}")


async def ask(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
subscription_id = get_user_subscription(user_id)
if not subscription_id:
await update.message.reply_text("Please set your subscription ID first by sending /start.")
return

if not context.args:
await update.message.reply_text("Please provide a question. Usage: /ask <your question>")
return

question = " ".join(context.args)
await update.message.reply_text("Thinking...")

try:
system_prompt = (
"You are an expert Azure SRE assistant. "
"Answer the user question clearly and concisely based on Azure monitoring, alerts, and metrics."
)

response = await openai_client.chat.completions.create(
model=AZURE_OPENAI_DEPLOYMENT_NAME,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": question},
],
)

answer = response.choices[0].message.content.strip()
await update.message.reply_text(answer)

except Exception as e:
logger.error(f"OpenAI error: {e}", exc_info=True)
await update.message.reply_text(f"Failed to process your request: {e}")


async def memory(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
subscription_id = get_user_subscription(user_id)
if not subscription_id:
await update.message.reply_text("Please set your subscription ID first by sending /start.")
return

if not context.args:
await update.message.reply_text("Please provide a resource ID. Usage: /memory <resource_id>")
return

resource_id = " ".join(context.args)
await update.message.reply_text(f"Checking memory usage for: {resource_id}")

try:
monitor_client = get_monitor_client(subscription_id)

now = datetime.now(timezone.utc)
start_time = (now - timedelta(minutes=15)).isoformat()
end_time = now.isoformat()

metrics_data = await monitor_client.metrics.list(
resource_uri=resource_id,
timespan=f"{start_time}/{end_time}",
interval="PT1M",
metricnames="AvailableMemoryBytes",
aggregation="Average",
)

values = []
for metric in metrics_data.value:
for timeserie in metric.timeseries:
for data in timeserie.data:
if data.average is not None:
values.append(data.average)

if not values:
await update.message.reply_text("No memory metric data found for the last 15 minutes.")
return

avg_mem_bytes = sum(values) / len(values)
avg_mem_mb = avg_mem_bytes / (1024 * 1024)
await update.message.reply_text(f"Average memory available in last 15 minutes: {avg_mem_mb:.2f} MB")

await monitor_client.close()

except Exception as e:
logger.error(f"Error fetching memory metrics: {e}", exc_info=True)
await update.message.reply_text(f"Failed to fetch memory metrics: {e}")


async def cpu(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
subscription_id = get_user_subscription(user_id)
if not subscription_id:
await update.message.reply_text("Please set your subscription ID first by sending /start.")
return

if not context.args:
await update.message.reply_text("Please provide a resource ID. Usage: /cpu <resource_id>")
return

resource_id = " ".join(context.args)
await update.message.reply_text(f"Checking CPU usage for: {resource_id}")

try:
monitor_client = get_monitor_client(subscription_id)

now = datetime.now(timezone.utc)
start_time = (now - timedelta(minutes=15)).isoformat()
end_time = now.isoformat()

metrics_data = await monitor_client.metrics.list(
resource_uri=resource_id,
timespan=f"{start_time}/{end_time}",
interval="PT1M",
metricnames="Percentage CPU",
aggregation="Average",
)

values = []
for metric in metrics_data.value:
for timeserie in metric.timeseries:
for data in timeserie.data:
if data.average is not None:
values.append(data.average)

if not values:
await update.message.reply_text("No CPU metric data found for the last 15 minutes.")
return

avg_cpu = sum(values) / len(values)
await update.message.reply_text(f"Average CPU usage in last 15 minutes: {avg_cpu:.2f}%")

await monitor_client.close()

except Exception as e:
logger.error(f"Error fetching CPU metrics: {e}", exc_info=True)
await update.message.reply_text(f"Failed to fetch CPU metrics: {e}")


async def listresources(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
subscription_id = get_user_subscription(user_id)
if not subscription_id:
await update.message.reply_text("Please set your subscription ID first by sending /start.")
return

await update.message.reply_text("Fetching Azure resources... This may take a moment.")

try:
resource_client = get_resource_client(subscription_id)

resources = []
for res in resource_client.resources.list():
resources.append(f"{res.type}: {res.name} ({res.id})")

if not resources:
await update.message.reply_text("No resources found.")
return

# Send in chunks of 10
chunk_size = 10
for i in range(0, len(resources), chunk_size):
chunk = resources[i : i + chunk_size]
msg = "\n".join(escape_markdown(r) for r in chunk)
await update.message.reply_text(msg, parse_mode="MarkdownV2")

except Exception as e:
logger.error(f"Error fetching resources: {e}", exc_info=True)
await update.message.reply_text(f"Failed to fetch resources: {e}")


def main():
app = ApplicationBuilder().token(TELEGRAM_TOKEN).build()

app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("check", check))
app.add_handler(CommandHandler("ask", ask))
app.add_handler(CommandHandler("memory", memory))
app.add_handler(CommandHandler("cpu", cpu))
app.add_handler(CommandHandler("listresources", listresources))

# Catch-all message handler to accept subscription ID input when user is not set
app.add_handler(MessageHandler(filters.TEXT & (~filters.COMMAND), subscription_input_handler))

logger.info("Bot started...")
app.run_polling()


if __name__ == "__main__":
main()

Continue reading...