Karma MCP Server

Instructions

Implementation Reference

• src/karma_mcp/server.py:532-623 (handler)

  The implementation of the `list_active_alerts` tool, which fetches data from the Karma API, filters for alerts with a state of 'active', and returns a formatted list of these alerts.

  async def list_active_alerts() -> str:
      """List only active (non-suppressed) alerts"""
      try:
          async with httpx.AsyncClient() as client:
              response = await client.post(
                  f"{KARMA_URL}/alerts.json",
                  headers={"Content-Type": "application/json"},
                  json={},
              )
  
              if response.status_code == 200:
                  data = response.json()
  
                  active_alerts = []
                  grids = data.get("grids", [])
  
                  for grid in grids:
                      for group in grid.get("alertGroups", []):
                          # Get group labels (contains alertname)
                          group_labels_dict = {}
                          for label in group.get("labels", []):
                              group_labels_dict[label.get("name", "")] = label.get(
                                  "value", ""
                              )
  
                          alertname = group_labels_dict.get("alertname", "unknown")
  
                          for alert in group.get("alerts", []):
                              alert_state = alert.get("state", "unknown")
  
                              # Only include active alerts (not suppressed)
                              if alert_state.lower() == "active":
                                  # Convert alert labels to dict
                                  alert_labels_dict = {}
                                  for label in alert.get("labels", []):
                                      alert_labels_dict[label.get("name", "")] = (
                                          label.get("value", "")
                                      )
  
                                  active_alerts.append(
                                      {
                                          "name": alertname,
                                          "state": alert_state,
                                          "severity": resolve_severity(
                                              group_labels_dict, alert_labels_dict
                                          ),
                                          "namespace": alert_labels_dict.get(
                                              "namespace", "N/A"
                                          ),
                                          "instance": alert_labels_dict.get(
                                              "instance", "N/A"
                                          ),
                                          "starts_at": alert.get("startsAt", "N/A"),
                                      }
                                  )
  
                  if not active_alerts:
                      return "No active alerts found."
  
                  # Format output
                  result = "Active Alerts (Non-Suppressed)\n"
                  result += "=" * 50 + "\n\n"
  
                  # Group by alert name
                  alert_groups = {}
                  for alert in active_alerts:
                      name = alert["name"]
                      if name not in alert_groups:
                          alert_groups[name] = []
                      alert_groups[name].append(alert)
  
                  for alertname, alerts in sorted(alert_groups.items()):
                      result += f"🔥 {alertname} ({len(alerts)} instance{'s' if len(alerts) > 1 else ''})\n"
                      result += f"   Severity: {alerts[0]['severity']}\n"
  
                      # Show details for each instance
                      for alert in alerts[:5]:  # Limit to 5 instances to avoid clutter
                          result += f"   • {alert['instance']} ({alert['namespace']})\n"
  
                      if len(alerts) > 5:
                          result += f"   • ... and {len(alerts) - 5} more\n"
  
                      result += "\n"
  
                  result += f"Total Active Alerts: {len(active_alerts)}"
                  return result
              else:
                  return f"Error fetching alerts: code {response.status_code}"
  
      except Exception as e:
          return f"Error connecting to Karma: {str(e)}"