alertmanager-mcp-server

Instructions

Implementation Reference

• src/alertmanager_mcp_server/server.py:494-516 (handler)

  The async handler function for the 'post_alerts' tool. It is registered via the @mcp.tool decorator and forwards the alerts list as JSON to the Alertmanager /api/v2/alerts POST endpoint using the make_request helper. Input schema is defined by the List[Dict] type hint and docstring.

  @mcp.tool(description="Create new alerts")
  async def post_alerts(alerts: List[Dict]):
      """Create new alerts
  
      Parameters
      ----------
      alerts
          A list of Alert object.
          [
              {
                  "startsAt": datetime,
                  "endsAt": datetime,
                  "annotations": labelSet
              }
          ]
  
      Returns
      -------
      dict:
          Create alert response from Alertmanager API.
      """
      return make_request(method="POST", route="/api/v2/alerts", json=alerts)

• src/alertmanager_mcp_server/server.py:183-234 (helper)

  Core helper function that performs HTTP requests to the Alertmanager API. Handles authentication, tenant headers, URL joining, error handling, and JSON parsing. Directly called by post_alerts to POST the alerts.

  def make_request(method="GET", route="/", **kwargs):
      """Make HTTP request and return a requests.Response object.
  
      Parameters
      ----------
      method : str
          HTTP method to use for the request.
      route : str
          (Default value = "/")
          This is the url we are making our request to.
      **kwargs : dict
          Arbitrary keyword arguments.
  
  
      Returns
      -------
      dict:
          The response from the Alertmanager API. This is a dictionary
          containing the response data.
      """
      try:
          route = url_join(config.url, route)
          auth = (
              requests.auth.HTTPBasicAuth(config.username, config.password)
              if config.username and config.password
              else None
          )
  
          # Add X-Scope-OrgId header for multi-tenant setups
          # Priority: 1) Request header from caller (via ContextVar), 2) Static config tenant
          headers = kwargs.get("headers", {})
  
          tenant_id = _current_scope_org_id.get() or config.tenant_id
  
          if tenant_id:
              headers["X-Scope-OrgId"] = tenant_id
          if headers:
              kwargs["headers"] = headers
  
          response = requests.request(
              method=method.upper(), url=route, auth=auth, timeout=60, **kwargs
          )
          response.raise_for_status()
          result = response.json()
  
          # Ensure we always return something (empty list is valid but might cause issues)
          if result is None:
              return {"message": "No data returned"}
          return result
      except requests.exceptions.RequestException as e:
          return {"error": str(e)}

• src/alertmanager_mcp_server/server.py:494-494 (registration)

  The @mcp.tool decorator registers the post_alerts function as an MCP tool with the name 'post_alerts' (inferred from function name) on the FastMCP instance.

  @mcp.tool(description="Create new alerts")