diff --git a/README.md b/README.md index 53dd455..a54be3c 100644 --- a/README.md +++ b/README.md @@ -178,6 +178,7 @@ This section describes the tools provided by the PagerDuty MCP server. They are | add_responders | Incidents | Adds responders to an incident | ❌ | | create_incident | Incidents | Creates a new incident | ❌ | | get_incident | Incidents | Retrieves a specific incident | ✅ | +| get_incident_log_entries | Incidents | Retrieves incident timeline/log entries | ✅ | | list_incidents | Incidents | Lists incidents | ✅ | | manage_incidents | Incidents | Updates status, urgency, assignment, or escalation level | ❌ | | add_team_member | Teams | Adds a user to a team with a specific role | ❌ | diff --git a/pagerduty_mcp/__main__.py b/pagerduty_mcp/__main__.py index c42a2bb..4d19ab1 100644 --- a/pagerduty_mcp/__main__.py +++ b/pagerduty_mcp/__main__.py @@ -6,5 +6,6 @@ def main(): print("Starting PagerDuty MCP Server. Use the --enable-write-tools flag to enable write tools.") app() + if __name__ == "__main__": main() diff --git a/pagerduty_mcp/models/__init__.py b/pagerduty_mcp/models/__init__.py index 460073a..a6405d0 100644 --- a/pagerduty_mcp/models/__init__.py +++ b/pagerduty_mcp/models/__init__.py @@ -10,6 +10,7 @@ IncidentQuery, IncidentResponderRequest, IncidentResponderRequestResponse, + LogEntry, ResponderRequest, ResponderRequestTarget, ) @@ -40,6 +41,7 @@ "IncidentResponderRequest", "IncidentResponderRequestResponse", "ListResponseModel", + "LogEntry", "MCPContext", "Oncall", "OncallQuery", diff --git a/pagerduty_mcp/models/incidents.py b/pagerduty_mcp/models/incidents.py index c956ca9..4522f08 100644 --- a/pagerduty_mcp/models/incidents.py +++ b/pagerduty_mcp/models/incidents.py @@ -206,8 +206,48 @@ class IncidentResponderRequestResponse(BaseModel): message: str | None = Field(default=None, description="The message included with the request") responder_request_targets: list[dict[str, Any]] = Field(description="The users requested to respond") + class IncidentNote(BaseModel): id: str | None = Field(description="The ID of the note", default=None) content: str = Field(description="The content of the note") created_at: datetime = Field(description="The time the note was created") user: UserReference = Field(description="The user who created the note") + + +# Log Entry Models +class LogEntryAgent(BaseModel): + id: str = Field(description="The ID of the agent") + type: str = Field(description="The type of the agent") + summary: str = Field(description="A summary of the agent") + self: str | None = Field(default=None, description="The API URL of the agent") + html_url: str | None = Field(default=None, description="The web URL of the agent") + + +class LogEntryChannel(BaseModel): + type: str = Field(description="The type of channel") + summary: str | None = Field(default=None, description="Summary of the channel") + notification: dict[str, Any] | None = Field(default=None, description="Notification details") + client: str | None = Field(default=None, description="Client name") + client_url: str | None = Field(default=None, description="Client URL") + + +class LogEntry(BaseModel): + id: str = Field(description="The ID of the log entry") + type: str = Field(description="The type of log entry") + summary: str = Field(description="A summary of the log entry") + self: str = Field(description="The API URL of the log entry") + html_url: str | None = Field(default=None, description="The web URL of the log entry") + created_at: datetime = Field(description="The time the log entry was created") + agent: LogEntryAgent | None = Field(default=None, description="The agent that created the log entry") + channel: LogEntryChannel | None = Field(default=None, description="The channel information") + service: ServiceReference = Field(description="The service associated with the log entry") + incident: dict[str, Any] = Field(description="The incident reference") + teams: list[dict[str, Any]] = Field(default=[], description="The teams associated with the log entry") + contexts: list[dict[str, Any]] = Field(default=[], description="Additional context information") + # Optional fields specific to certain log entry types + acknowledgement_timeout: int | None = Field(default=None, description="Acknowledgement timeout in seconds") + assignees: list[UserReference] | None = Field(default=None, description="Users assigned in this log entry") + user: UserReference | None = Field(default=None, description="User associated with the log entry") + linked_incident: dict[str, Any] | None = Field(default=None, description="Linked incident reference") + event_details: dict[str, Any] = Field(default={}, description="Event details") + event_rule_action: dict[str, Any] | None = Field(default=None, description="Event rule action details") diff --git a/pagerduty_mcp/tools/__init__.py b/pagerduty_mcp/tools/__init__.py index 03ea84d..8258afc 100644 --- a/pagerduty_mcp/tools/__init__.py +++ b/pagerduty_mcp/tools/__init__.py @@ -11,6 +11,7 @@ add_responders, create_incident, get_incident, + get_incident_log_entries, list_incidents, manage_incidents, ) @@ -44,6 +45,7 @@ # Incidents list_incidents, get_incident, + get_incident_log_entries, # Services list_services, get_service, diff --git a/pagerduty_mcp/tools/incidents.py b/pagerduty_mcp/tools/incidents.py index ac22286..aefb910 100644 --- a/pagerduty_mcp/tools/incidents.py +++ b/pagerduty_mcp/tools/incidents.py @@ -13,6 +13,7 @@ IncidentResponderRequest, IncidentResponderRequestResponse, ListResponseModel, + LogEntry, MCPContext, UserReference, ) @@ -70,6 +71,24 @@ def get_incident(incident_id: str) -> Incident: return Incident.model_validate(response) +def get_incident_log_entries(incident_id: str) -> list[LogEntry]: + """Get log entries (timeline) for a specific incident. + + This function retrieves the complete timeline of activities for an incident, + including acknowledgments, assignments, status changes, and other actions. + This provides the same information visible in the PagerDuty UI Timeline tab. + + Args: + incident_id: The ID of the incident to retrieve log entries for. + + Returns: + List of log entries representing the incident timeline + """ + response = get_client().rget(f"/incidents/{incident_id}/log_entries") + # The response is a direct list of log entries + return [LogEntry.model_validate(entry) for entry in response] + + def create_incident(create_model: IncidentCreateRequest) -> Incident: """Create an incident. diff --git a/tests/test_incidents.py b/tests/test_incidents.py index b565e5a..5ebce2a 100644 --- a/tests/test_incidents.py +++ b/tests/test_incidents.py @@ -17,6 +17,7 @@ IncidentResponderRequest, IncidentResponderRequestResponse, ListResponseModel, + LogEntry, MCPContext, ServiceReference, UserReference, @@ -32,6 +33,7 @@ add_responders, create_incident, get_incident, + get_incident_log_entries, list_incidents, manage_incidents, ) @@ -61,6 +63,80 @@ def setUpClass(cls): "html_url": "https://test.pagerduty.com/incidents/PINCIDENT123", } + cls.sample_log_entry_data = [ + { + "id": "RO52CY6QBZKDN9PEGS3YPN35TH", + "type": "resolve_log_entry", + "summary": "Resolved through the API.", + "self": "https://api.pagerduty.com/log_entries/RO52CY6QBZKDN9PEGS3YPN35TH", + "html_url": None, + "created_at": "2025-09-18T22:59:28Z", + "agent": { + "id": "PWB6A94", + "type": "events_api_v2_inbound_integration_reference", + "summary": "AlertManager", + "self": "https://api.pagerduty.com/services/P9TYXZJ/integrations/PWB6A94", + "html_url": "https://thousandeyesops.pagerduty.com/services/P9TYXZJ/integrations/PWB6A94", + }, + "channel": { + "type": "integration", + "client": "Alertmanager", + "client_url": "https://alertmanager.example.com", + }, + "service": { + "id": "P9TYXZJ", + "type": "service_reference", + "summary": "[PROD] Webapps", + "self": "https://api.pagerduty.com/services/P9TYXZJ", + "html_url": "https://thousandeyesops.pagerduty.com/service-directory/P9TYXZJ", + }, + "incident": { + "id": "Q0Y9H9AUGZHAHS", + "type": "incident_reference", + "summary": "[#568677] Test incident", + "self": "https://api.pagerduty.com/incidents/Q0Y9H9AUGZHAHS", + "html_url": "https://thousandeyesops.pagerduty.com/incidents/Q0Y9H9AUGZHAHS", + }, + "teams": [], + "contexts": [], + "event_details": {}, + }, + { + "id": "RONF3S8XQ8TVEA2DXQS6FRPBEV", + "type": "acknowledge_log_entry", + "summary": "Acknowledged by Test User.", + "self": "https://api.pagerduty.com/log_entries/RONF3S8XQ8TVEA2DXQS6FRPBEV", + "html_url": None, + "created_at": "2025-09-18T22:54:38Z", + "agent": { + "id": "PE9XZ7K", + "type": "user_reference", + "summary": "Test User", + "self": "https://api.pagerduty.com/users/PE9XZ7K", + "html_url": "https://thousandeyesops.pagerduty.com/users/PE9XZ7K", + }, + "channel": {"type": "mobile"}, + "service": { + "id": "P9TYXZJ", + "type": "service_reference", + "summary": "[PROD] Webapps", + "self": "https://api.pagerduty.com/services/P9TYXZJ", + "html_url": "https://thousandeyesops.pagerduty.com/service-directory/P9TYXZJ", + }, + "incident": { + "id": "Q0Y9H9AUGZHAHS", + "type": "incident_reference", + "summary": "[#568677] Test incident", + "self": "https://api.pagerduty.com/incidents/Q0Y9H9AUGZHAHS", + "html_url": "https://thousandeyesops.pagerduty.com/incidents/Q0Y9H9AUGZHAHS", + }, + "teams": [], + "contexts": [], + "acknowledgement_timeout": 3600, + "event_details": {}, + }, + ] + cls.sample_user_data = Mock() cls.sample_user_data.id = "PUSER123" cls.sample_user_data.teams = [Mock(id="PTEAM123")] @@ -182,6 +258,68 @@ def test_get_incident_api_error(self, mock_get_client): self.assertIn("API Error", str(context.exception)) + @patch("pagerduty_mcp.tools.incidents.get_client") + def test_get_incident_log_entries_success(self, mock_get_client): + """Test getting incident log entries successfully.""" + # Setup mock + mock_client = Mock() + mock_client.rget.return_value = self.sample_log_entry_data + mock_get_client.return_value = mock_client + + # Test + result = get_incident_log_entries("PINCIDENT123") + + # Assertions + self.assertIsInstance(result, list) + self.assertEqual(len(result), 2) + self.assertIsInstance(result[0], LogEntry) + self.assertIsInstance(result[1], LogEntry) + + # Check first log entry (resolve) + self.assertEqual(result[0].id, "RO52CY6QBZKDN9PEGS3YPN35TH") + self.assertEqual(result[0].type, "resolve_log_entry") + self.assertEqual(result[0].summary, "Resolved through the API.") + self.assertIsNotNone(result[0].agent) + self.assertEqual(result[0].agent.summary, "AlertManager") + + # Check second log entry (acknowledge) + self.assertEqual(result[1].id, "RONF3S8XQ8TVEA2DXQS6FRPBEV") + self.assertEqual(result[1].type, "acknowledge_log_entry") + self.assertEqual(result[1].summary, "Acknowledged by Test User.") + self.assertEqual(result[1].acknowledgement_timeout, 3600) + + mock_client.rget.assert_called_once_with("/incidents/PINCIDENT123/log_entries") + + @patch("pagerduty_mcp.tools.incidents.get_client") + def test_get_incident_log_entries_empty(self, mock_get_client): + """Test getting incident log entries with empty response.""" + # Setup mock + mock_client = Mock() + mock_client.rget.return_value = [] + mock_get_client.return_value = mock_client + + # Test + result = get_incident_log_entries("PINCIDENT123") + + # Assertions + self.assertIsInstance(result, list) + self.assertEqual(len(result), 0) + mock_client.rget.assert_called_once_with("/incidents/PINCIDENT123/log_entries") + + @patch("pagerduty_mcp.tools.incidents.get_client") + def test_get_incident_log_entries_api_error(self, mock_get_client): + """Test get_incident_log_entries with API error.""" + # Setup mock to raise exception + mock_client = Mock() + mock_client.rget.side_effect = Exception("API Error") + mock_get_client.return_value = mock_client + + # Test that exception is raised + with self.assertRaises(Exception) as context: + get_incident_log_entries("PINCIDENT123") + + self.assertIn("API Error", str(context.exception)) + @patch("pagerduty_mcp.tools.incidents.get_client") def test_create_incident_success(self, mock_get_client): """Test creating an incident successfully.""" @@ -527,10 +665,7 @@ def test_add_note_to_incident_success(self, mock_get_client): "id": "PNOTE123", "content": "This is a test note", "created_at": "2023-01-01T10:00:00Z", - "user": { - "id": "PUSER123", - "summary": "Test User" - } + "user": {"id": "PUSER123", "summary": "Test User"}, } mock_client = Mock() @@ -548,8 +683,7 @@ def test_add_note_to_incident_success(self, mock_get_client): # Verify API call mock_client.rpost.assert_called_once_with( - "/incidents/PINC123/notes", - json={"note": {"content": "This is a test note"}} + "/incidents/PINC123/notes", json={"note": {"content": "This is a test note"}} ) def test_incidentquery_reject_statuses_param(self):