post_alerts

Instructions

Implementation Reference

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

  The tool handler for post_alerts, decorated with @mcp.tool, that takes a list of alert dicts and sends them as a POST request to the Alertmanager API at /api/v2/alerts.

  @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:494-494 (registration)

  Registration of post_alerts as an MCP tool via the @mcp.tool decorator on FastMCP instance 'mcp'.

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

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

  The make_request helper function that performs the actual HTTP request to the Alertmanager API, used by post_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:60-60 (registration)

  The FastMCP instance on which the @mcp.tool decorator registers tools like post_alerts.

  mcp = FastMCP("Alertmanager MCP")

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

  The input schema for post_alerts is defined by the type hint List[Dict] on the alerts parameter.

  @mcp.tool(description="Create new alerts")
  async def post_alerts(alerts: List[Dict]):