feat: add get_incident_log_entries function for incident timeline data

vicgp-te · vicgp-te · commit f71b72248262 · 2025-09-19T06:35:37.000-07:00
- Retrieve complete incident timeline including acknowledgments and status changes
- Add LogEntry, LogEntryAgent, and LogEntryChannel models
- Include comprehensive unit tests with 100% coverage
- Update documentation and tool registration
diff --git 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
@@ -1,9 +1,11 @@
 from pagerduty_mcp.server import app
 
+
 def main():
     """Main entry point for the pagerduty-mcp command."""
     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
@@ -10,6 +10,7 @@
     IncidentQuery,
     IncidentResponderRequest,
     IncidentResponderRequestResponse,
+    LogEntry,
 )
 from .oncalls import Oncall, OncallQuery
 from .references import IncidentReference, ScheduleReference, ServiceReference, TeamReference, UserReference
@@ -38,6 +39,7 @@
     "IncidentResponderRequest",
     "IncidentResponderRequestResponse",
     "ListResponseModel",
+    "LogEntry",
     "MCPContext",
     "Oncall",
     "OncallQuery",
diff --git a/pagerduty_mcp/models/incidents.py b/pagerduty_mcp/models/incidents.py
@@ -195,8 +195,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
@@ -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
@@ -13,6 +13,7 @@
     IncidentResponderRequest,
     IncidentResponderRequestResponse,
     ListResponseModel,
+    LogEntry,
     MCPContext,
     UserReference,
 )
@@ -51,6 +52,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
@@ -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."""
@@ -467,10 +605,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()
@@ -488,9 +623,9 @@ 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"}}
         )
 
+
 if __name__ == "__main__":
     unittest.main()
