core.alerting_system

Alerting and Notification System

Comprehensive alerting system for the MCP Academic RAG Server providing multi-channel notifications, escalation policies, and alert correlation.

Classes

`AlertCorrelation`([...])	Alert correlation configuration
`AlertCorrelationEngine`(config)	Engine for correlating and deduplicating alerts
`AlertingSystem`([config])	Main alerting system coordinator
`ConsoleNotificationProvider`(config)	Console notification provider
`EmailNotificationProvider`(config)	Email notification provider
`EscalationLevel`(value)	Alert escalation levels
`EscalationPolicy`(name, levels[, enabled])	Alert escalation policy configuration
`LogNotificationProvider`(config)	Log-based notification provider
`NotificationChannel`(value)	Available notification channels
`NotificationConfig`(channel[, enabled, config])	Configuration for a notification channel
`NotificationProvider`(config)	Abstract base class for notification providers
`SlackNotificationProvider`(config)	Slack notification provider using webhooks
`WebhookNotificationProvider`(config)	Webhook notification provider

class core.alerting_system.NotificationChannel(value)[source]

Bases: Enum

Available notification channels

EMAIL = 'email'

WEBHOOK = 'webhook'

SLACK = 'slack'

TEAMS = 'teams'

DISCORD = 'discord'

LOG = 'log'

CONSOLE = 'console'

class core.alerting_system.EscalationLevel(value)[source]

Bases: Enum

Alert escalation levels

IMMEDIATE = 'immediate'

AFTER_5_MIN = 'after_5_min'

AFTER_15_MIN = 'after_15_min'

AFTER_1_HOUR = 'after_1_hour'

class core.alerting_system.NotificationConfig(channel, enabled=True, config=None)[source]

Bases: object

Configuration for a notification channel

channel: NotificationChannel

enabled: bool = True

config: Dict[str, Any] | None = None

__init__(channel, enabled=True, config=None)

class core.alerting_system.EscalationPolicy(name, levels, enabled=True)[source]

Bases: object

Alert escalation policy configuration

name: str

levels: List[Dict[str, Any]]

enabled: bool = True

get_escalation_for_duration(duration_minutes)[source]

Get escalation configuration for alert duration

__init__(name, levels, enabled=True)

class core.alerting_system.AlertCorrelation(correlation_window_minutes=5, max_similar_alerts=10, correlation_fields=None)[source]

Bases: object

Alert correlation configuration

correlation_window_minutes: int = 5

max_similar_alerts: int = 10

correlation_fields: List[str] | None = None

__init__(correlation_window_minutes=5, max_similar_alerts=10, correlation_fields=None)

class core.alerting_system.NotificationProvider(config)[source]

Bases: ABC

Abstract base class for notification providers

__init__(config)[source]

abstractmethod async send_notification(alert, context=None)[source]

Send notification for an alert

abstractmethod validate_config()[source]

Validate provider configuration

class core.alerting_system.EmailNotificationProvider(config)[source]

Bases: NotificationProvider

Email notification provider

__init__(config)[source]

validate_config()[source]

Validate email configuration

async send_notification(alert, context=None)[source]

Send email notification

class core.alerting_system.WebhookNotificationProvider(config)[source]

Bases: NotificationProvider

Webhook notification provider

__init__(config)[source]

validate_config()[source]

Validate webhook configuration

async send_notification(alert, context=None)[source]

Send webhook notification

class core.alerting_system.SlackNotificationProvider(config)[source]

Bases: WebhookNotificationProvider

Slack notification provider using webhooks

class core.alerting_system.LogNotificationProvider(config)[source]

Bases: NotificationProvider

Log-based notification provider

__init__(config)[source]

validate_config()[source]

Log provider always valid

async send_notification(alert, context=None)[source]

Log alert notification

class core.alerting_system.ConsoleNotificationProvider(config)[source]

Bases: NotificationProvider

Console notification provider

validate_config()[source]

Console provider always valid

async send_notification(alert, context=None)[source]

Print alert to console

class core.alerting_system.AlertCorrelationEngine(config)[source]

Bases: object

Engine for correlating and deduplicating alerts

__init__(config)[source]

recent_alerts: List[Alert]

correlated_alerts: Dict[str, List[Alert]]

correlate_alert(alert)[source]

Check if alert should be correlated (deduplicated)

class core.alerting_system.AlertingSystem(config=None)[source]

Bases: object

Main alerting system coordinator

__init__(config=None)[source]

providers: Dict[NotificationChannel, NotificationProvider]

notification_configs: List[NotificationConfig]

escalation_policies: Dict[str, EscalationPolicy]

correlation_engine: AlertCorrelationEngine | None

sent_notifications: List[Dict[str, Any]]

escalated_alerts: Dict[str, datetime]

escalation_task: Task | None

async send_alert(alert, context=None)[source]

Send alert through configured notification channels

start_escalation_monitoring()[source]

Start escalation monitoring task

stop_escalation_monitoring()[source]

Stop escalation monitoring

get_notification_stats()[source]

Get notification statistics