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