Karma MCP Server

Instructions

Implementation Reference

• src/karma_mcp/server.py:1235-1359 (handler)

  The 'create_silence' tool handler implementation. Note: It is currently a placeholder for the actual API interaction.

  async def create_silence(
      cluster: str,
      alertname: str,
      duration: str = "2h",
      comment: str = "Silenced via MCP",
      matchers: str = "",
  ) -> str:
      """Create a new silence for specific alerts
  
      Args:
          cluster: Target cluster name (e.g., 'teddy-prod')
          alertname: Name of the alert to silence
          duration: Duration of silence (e.g., '2h', '30m', '1d')
          comment: Comment explaining why the alert is being silenced
          matchers: Additional matchers in format 'key=value,key2=value2' (optional)
  
      Returns:
          Silence creation result with silence ID
      """
      try:
          # First, get the Alertmanager URL for this cluster
          async with httpx.AsyncClient() as client:
              response = await client.post(
                  f"{KARMA_URL}/alerts.json",
                  json={},
                  headers={"Content-Type": "application/json"},
                  timeout=10.0,
              )
  
              if response.status_code != 200:
                  return (
                      f"Error fetching cluster information: code {response.status_code}"
                  )
  
              data = response.json()
              upstreams = data.get("upstreams", {}).get("instances", [])
  
              # Find the Alertmanager for this cluster
              alertmanager_url = None
              for upstream in upstreams:
                  if cluster.lower() in upstream.get("cluster", "").lower():
                      # For now, we'll need to use the publicURI
                      # In production, you might need to handle authentication
                      alertmanager_url = upstream.get("publicURI")
                      break
  
              if not alertmanager_url:
                  return f"Could not find Alertmanager for cluster: {cluster}"
  
              # Parse duration
              import re
              from datetime import datetime, timedelta
  
              # Parse duration string (e.g., '2h', '30m', '1d')
              duration_match = re.match(r"(\d+)([hdm])", duration.lower())
              if not duration_match:
                  return "Invalid duration format. Use format like '2h', '30m', or '1d'"
  
              amount = int(duration_match.group(1))
              unit = duration_match.group(2)
  
              if unit == "h":
                  delta = timedelta(hours=amount)
              elif unit == "m":
                  delta = timedelta(minutes=amount)
              elif unit == "d":
                  delta = timedelta(days=amount)
              else:
                  return f"Unsupported time unit: {unit}"
  
              starts_at = datetime.utcnow()
              ends_at = starts_at + delta
  
              # Build matchers
              silence_matchers = [
                  {
                      "name": "alertname",
                      "value": alertname,
                      "isRegex": False,
                      "isEqual": True,
                  }
              ]
  
              # Add additional matchers if provided
              if matchers:
                  for matcher_str in matchers.split(","):
                      if "=" in matcher_str:
                          key, value = matcher_str.split("=", 1)
                          silence_matchers.append(
                              {
                                  "name": key.strip(),
                                  "value": value.strip(),
                                  "isRegex": False,
                                  "isEqual": True,
                              }
                          )
  
              # Create silence request
              {
                  "matchers": silence_matchers,
                  "startsAt": starts_at.isoformat() + "Z",
                  "endsAt": ends_at.isoformat() + "Z",
                  "createdBy": "karma-mcp",
                  "comment": comment,
              }
  
              # Note: Actually creating the silence requires direct Alertmanager API access
              # which may need authentication. This is a placeholder for the actual implementation
              return f"""
  ⚠️ Silence Creation Request Prepared:
  
  📍 Cluster: {cluster}
  🔕 Alert: {alertname}
  ⏱️ Duration: {duration} (until {ends_at.strftime("%Y-%m-%d %H:%M UTC")})
  💬 Comment: {comment}
  🏷️ Matchers: {len(silence_matchers)} configured
  
  Note: Direct Alertmanager API integration is required to complete this action.
  The Alertmanager endpoint would be: {alertmanager_url}/api/v2/silences
  
  To manually create this silence, you can use the Karma UI or Alertmanager API directly.
  """
  
      except Exception as e:
          return f"Error creating silence: {str(e)}"