create_silence

Instructions

Implementation Reference

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

  Main tool implementation: `create_silence` MCP tool decorated with @mcp.tool(). Fetches cluster info from Karma API, parses duration, builds silence matchers, and returns a prepared silence creation request (placeholder awaiting direct Alertmanager API integration).

  @mcp.tool()
  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)}"

• src/karma_mcp/server.py:1234-1234 (registration)

  The @mcp.tool() decorator registers `create_silence` as an MCP tool on the FastMCP server instance.

  @mcp.tool()

• src/karma_mcp/server.py:1236-1253 (schema)

  Input parameters/schema for create_silence: cluster (str), alertname (str), duration (str, default '2h'), comment (str, default 'Silenced via MCP'), matchers (str, default '').

      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
      """

• src/karma_mcp/http_server.py:20-22 (registration)

  Import of create_silence from .server into the HTTP server module for REST endpoint exposure.

  from .server import (
      check_karma,
      create_silence,

• src/karma_mcp/http_server.py:225-239 (handler)

  HTTP REST endpoint wrapper: POST /silences that calls the create_silence MCP tool via the create_silence_endpoint function.

  @app.post("/silences", response_model=MCPResponse)
  async def create_silence_endpoint(request: SilenceRequest):
      """Create a new silence for specific alerts"""
      try:
          result = await create_silence(
              cluster=request.cluster,
              alertname=request.alertname,
              duration=request.duration,
              comment=request.comment,
              matchers=request.matchers,
          )
          return MCPResponse(success=True, data=result)
      except Exception as e:
          logger.error(f"Error creating silence: {e}")
          return MCPResponse(success=False, data="", error=str(e))