Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 | ❌ |
Expand Down
1 change: 1 addition & 0 deletions pagerduty_mcp/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
2 changes: 2 additions & 0 deletions pagerduty_mcp/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
IncidentQuery,
IncidentResponderRequest,
IncidentResponderRequestResponse,
LogEntry,
ResponderRequest,
ResponderRequestTarget,
)
Expand Down Expand Up @@ -40,6 +41,7 @@
"IncidentResponderRequest",
"IncidentResponderRequestResponse",
"ListResponseModel",
"LogEntry",
"MCPContext",
"Oncall",
"OncallQuery",
Expand Down
40 changes: 40 additions & 0 deletions pagerduty_mcp/models/incidents.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
2 changes: 2 additions & 0 deletions pagerduty_mcp/tools/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
add_responders,
create_incident,
get_incident,
get_incident_log_entries,
list_incidents,
manage_incidents,
)
Expand Down Expand Up @@ -44,6 +45,7 @@
# Incidents
list_incidents,
get_incident,
get_incident_log_entries,
# Services
list_services,
get_service,
Expand Down
19 changes: 19 additions & 0 deletions pagerduty_mcp/tools/incidents.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
IncidentResponderRequest,
IncidentResponderRequestResponse,
ListResponseModel,
LogEntry,
MCPContext,
UserReference,
)
Expand Down Expand Up @@ -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.

Expand Down
146 changes: 140 additions & 6 deletions tests/test_incidents.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
IncidentResponderRequest,
IncidentResponderRequestResponse,
ListResponseModel,
LogEntry,
MCPContext,
ServiceReference,
UserReference,
Expand All @@ -32,6 +33,7 @@
add_responders,
create_incident,
get_incident,
get_incident_log_entries,
list_incidents,
manage_incidents,
)
Expand Down Expand Up @@ -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")]
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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()
Expand All @@ -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):
Expand Down