post_alerts
Create new alerts in Alertmanager to notify teams about system issues or events requiring attention.
Instructions
Create new alerts
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| alerts | Yes |
Implementation Reference
- 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) - 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")