|
| 1 | +from datetime import datetime |
| 2 | +from typing import Literal |
| 3 | + |
| 4 | +from mcp.server import FastMCP |
| 5 | +from pydantic import BaseModel, Field |
| 6 | + |
| 7 | +from ..client import ( |
| 8 | + grafana_client, |
| 9 | + AddActivityToIncidentArguments, |
| 10 | + CreateIncidentArguments, |
| 11 | + IncidentPreviewBody, |
| 12 | + IncidentPreviewQuery, |
| 13 | +) |
| 14 | + |
| 15 | + |
| 16 | +class ListIncidentsArguments(BaseModel): |
| 17 | + """ |
| 18 | + Arguments for the list_incidents tool. |
| 19 | + """ |
| 20 | + |
| 21 | + # Note: this differs from what the client expects (which is a closer match |
| 22 | + # to the API) so that we can make it easier for LLM clients to use sensible |
| 23 | + # arguments without them having to know the query syntax. |
| 24 | + |
| 25 | + limit: int = Field(default=10, ge=1, le=100) |
| 26 | + drill: bool = Field( |
| 27 | + default=False, description="Whether to include drill incidents." |
| 28 | + ) |
| 29 | + status: Literal["resolved", "active"] | None = Field( |
| 30 | + default=None, |
| 31 | + description="The status of the incidents to include. If not provided, all incidents will be included.", |
| 32 | + ) |
| 33 | + |
| 34 | + |
| 35 | +async def list_incidents(arguments: ListIncidentsArguments) -> bytes: |
| 36 | + """ |
| 37 | + List incidents from the Grafana Incident incident management tool. |
| 38 | +
|
| 39 | + Incidents will be returned in descending order by creation time. |
| 40 | + """ |
| 41 | + query_string = "isdrill:false" if not arguments.drill else "" |
| 42 | + if arguments.status is not None: |
| 43 | + query_string += f" and status:{arguments.status}" |
| 44 | + body = IncidentPreviewBody( |
| 45 | + query=IncidentPreviewQuery( |
| 46 | + query_string=query_string, |
| 47 | + order_direction="DESC", |
| 48 | + order_field="createdTime", |
| 49 | + limit=arguments.limit, |
| 50 | + ), |
| 51 | + include_custom_field_values=True, |
| 52 | + ) |
| 53 | + return await grafana_client.list_incidents(body) |
| 54 | + |
| 55 | + |
| 56 | +async def create_incident(arguments: CreateIncidentArguments) -> bytes: |
| 57 | + """ |
| 58 | + Create an incident in the Grafana Incident incident management tool. |
| 59 | + """ |
| 60 | + return await grafana_client.create_incident(arguments) |
| 61 | + |
| 62 | + |
| 63 | +async def add_activity_to_incident( |
| 64 | + incident_id: str, body: str, event_time: datetime | None = None |
| 65 | +) -> bytes: |
| 66 | + """ |
| 67 | + Add an activity to an incident in the Grafana Incident incident management tool. |
| 68 | +
|
| 69 | + :param incident_id: The ID of the incident to add the activity to. |
| 70 | + :param body: The body of the activity. URLs will be parsed and attached as context. |
| 71 | + :param event_time: The time that the activity occurred. If not provided, the current time will be used. |
| 72 | + If provided, it must be in RFC3339 format. |
| 73 | + """ |
| 74 | + return await grafana_client.add_activity_to_incident( |
| 75 | + AddActivityToIncidentArguments( |
| 76 | + incident_id=incident_id, |
| 77 | + body=body, |
| 78 | + event_time=event_time, |
| 79 | + ) |
| 80 | + ) |
| 81 | + |
| 82 | + |
| 83 | +async def resolve_incident(incident_id: str, summary: str) -> bytes: |
| 84 | + """ |
| 85 | + Resolve an incident in the Grafana Incident incident management tool. |
| 86 | +
|
| 87 | + :param incident_id: The ID of the incident to resolve. |
| 88 | + :param summary: The summary of the incident now it has been resolved. |
| 89 | + This should be succint but thorough and informative, |
| 90 | + enough to serve as a mini post-incident-report. |
| 91 | + """ |
| 92 | + return await grafana_client.close_incident(incident_id, summary) |
| 93 | + |
| 94 | + |
| 95 | +def add_tools(mcp: FastMCP): |
| 96 | + mcp.add_tool(list_incidents) |
| 97 | + mcp.add_tool(create_incident) |
| 98 | + mcp.add_tool(add_activity_to_incident) |
| 99 | + mcp.add_tool(resolve_incident) |
0 commit comments