Spaces:
Sleeping
Sleeping
File size: 2,973 Bytes
c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 8aa8f36 c322df2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
import json
from typing import List, Optional, Dict, Any
from src.alert_mcp.db import get_db
from src.alert_mcp import mcp_tools
# We use synchronous calls directly to the database logic
# This avoids the need for a separate backend server process in the Space.
def log_alert(
provider_id: int,
severity: str,
window_days: int,
message: str,
credential_id: Optional[int] = None,
channel: Optional[str] = "ui"
) -> Dict[str, Any]:
"""
Log a new alert in the system.
Args:
provider_id: The ID of the provider.
severity: One of "info", "warning", "critical".
window_days: The window in days.
message: The alert message.
credential_id: Optional credential ID.
channel: Notification channel (default: "ui").
"""
db = next(get_db())
try:
alert = mcp_tools.log_alert(
db=db,
provider_id=provider_id,
credential_id=credential_id,
severity=severity,
window_days=window_days,
message=message,
channel=channel
)
return alert.model_dump(mode='json')
except ValueError as e:
return {"error": str(e)}
except Exception as e:
return {"error": f"An error occurred: {str(e)}"}
finally:
db.close()
def get_open_alerts(
provider_id: Optional[int] = None,
severity: Optional[str] = None
) -> List[Dict[str, Any]]:
"""
Get a list of open (unresolved) alerts.
Args:
provider_id: Optional filter by provider ID.
severity: Optional filter by severity.
"""
db = next(get_db())
try:
alerts = mcp_tools.get_open_alerts(db=db, provider_id=provider_id, severity=severity)
return [a.model_dump(mode='json') for a in alerts]
except Exception as e:
return [{"error": str(e)}]
finally:
db.close()
def mark_alert_resolved(
alert_id: int,
resolution_note: Optional[str] = None
) -> Dict[str, Any]:
"""
Mark an alert as resolved.
Args:
alert_id: The ID of the alert to resolve.
resolution_note: A note explaining the resolution.
"""
db = next(get_db())
try:
alert = mcp_tools.mark_alert_resolved(db=db, alert_id=alert_id, resolution_note=resolution_note)
return alert.model_dump(mode='json')
except ValueError as e:
return {"error": str(e)}
except Exception as e:
return {"error": str(e)}
finally:
db.close()
def summarize_alerts(window_days: Optional[int] = None) -> Dict[str, Any]:
"""
Get a summary of alerts (counts by severity).
Args:
window_days: Optional window in days to summarize over.
"""
db = next(get_db())
try:
summary = mcp_tools.summarize_alerts(db=db, window_days=window_days)
return summary.model_dump(mode='json')
except Exception as e:
return {"error": str(e)}
finally:
db.close()
|