Karma MCP Server

Instructions

Implementation Reference

• src/karma_mcp/server.py:626-717 (handler)

  The `list_suppressed_alerts` function is the handler that fetches alerts from the Karma API, filters them to find suppressed instances, groups them by alert name, and formats the result.

  async def list_suppressed_alerts() -> str:
      """List only 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()
  
                  suppressed_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 suppressed alerts
                              if alert_state.lower() == "suppressed":
                                  # Convert alert labels to dict
                                  alert_labels_dict = {}
                                  for label in alert.get("labels", []):
                                      alert_labels_dict[label.get("name", "")] = (
                                          label.get("value", "")
                                      )
  
                                  suppressed_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 suppressed_alerts:
                      return "No suppressed alerts found."
  
                  # Format output
                  result = "Suppressed Alerts\n"
                  result += "=" * 50 + "\n\n"
  
                  # Group by alert name
                  alert_groups = {}
                  for alert in suppressed_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 Suppressed Alerts: {len(suppressed_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)}"