PagerDuty MCP CLI Client - minimal CLI for PagerDuty hosted MCP server

README.md

PagerDuty MCP CLI Client

Minimal CLI client for PagerDuty's hosted MCP server — no LLM required. Connects to mcp.pagerduty.com via streamable HTTP transport with token auth. Lists available MCP tools and calls them with JSON arguments.

Setup

pip install mcp
export PAGERDUTY_USER_API_KEY="u+your-token-here"

Usage

# List all available tools
python3 getdata.py

# Call a tool with default arguments
python3 getdata.py list_incidents '{"query_model": {}}'
python3 getdata.py list_services '{"query_model": {}}'

# Call a tool with specific arguments
python3 getdata.py get_incident '{"incident_id": "Q1EBPJVK2KASMY"}'
python3 getdata.py list_oncalls '{"query_model": {}}'

Environment Variables

Variable	Required	Default	Description
PAGERDUTY_USER_API_KEY	Yes	—	PagerDuty user API token (format: u+...)
PAGERDUTY_MCP_URL	No	https://mcp.pagerduty.com/mcp	MCP server endpoint

MCP Endpoints

Endpoint	Tools	Purpose
https://mcp.pagerduty.com/mcp (default)	64 CRUD tools	Direct PagerDuty API operations — incidents, services, schedules, teams, escalation policies, event orchestrations, status pages
https://mcp.pagerduty.com/pagerduty-advance-mcp	4 AI agent tools	SRE agent, insights/analytics, scheduling/on-call agents

PagerDuty Advance MCP

The Advance endpoint exposes four AI-powered agent tools:

• sre_agent_tool — Incident response and root cause analysis
• insights_agent_tool — Historical reporting and trend analysis
• shift_agent_find_coverage_tool — Find replacement on-call coverage
• shift_agent_on_call_questions_and_conflicts_tool — Scheduling and on-call queries

SRE Agent Example

export PAGERDUTY_MCP_URL='https://mcp.pagerduty.com/pagerduty-advance-mcp'

# List the 4 Advance tools
python3 getdata.py

# Analyze an incident (requires incident_id + message)
python3 getdata.py sre_agent_tool \
  '{"message": "What happened and what was the root cause?", "incident_id": "Q1EBPJVK2KASMY"}'

The SRE agent returns structured markdown with:

• Incident description and categorization
• Root cause analysis
• Pattern detection across past incidents
• Recommended next steps

How It Works

Single Python script (getdata.py) using:

• mcp.client.streamable_http.streamablehttp_client for transport
• mcp.ClientSession for the MCP protocol layer
• Header-based auth: Authorization: Token token={key}

Each operation (list tools, call tool) creates a fresh session.

getdata.py

#!/usr/bin/env python3
"""
Minimal PagerDuty hosted MCP client - no LLM required.

Connects to the PagerDuty hosted MCP service at mcp.pagerduty.com
using streamable HTTP transport with token auth.

Requirements:
    pip install mcp

Usage:
    export PAGERDUTY_USER_API_KEY="u+your-token-here"

    # List available tools
    python pd_mcp_client.py

    # Call a tool
    python pd_mcp_client.py get_incidents
    python pd_mcp_client.py get_incidents '{"status": "triggered"}'
    python pd_mcp_client.py list_services
    python pd_mcp_client.py get_oncalls '{"schedule_id": "PXXXXX"}'
"""

import asyncio
import json
import os
import sys

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

PD_MCP_URL = os.environ.get("PAGERDUTY_MCP_URL", "https://mcp.pagerduty.com/mcp")
PD_API_KEY = os.environ.get("PAGERDUTY_USER_API_KEY", "")


async def get_session():
    """Create an authenticated MCP session"""
    headers = {"Authorization": f"Token token={PD_API_KEY}"}
    return streamablehttp_client(PD_MCP_URL, headers=headers)


async def list_tools():
    async with streamablehttp_client(
        PD_MCP_URL,
        headers={"Authorization": f"Token token={PD_API_KEY}"}
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            return await session.list_tools()


async def call_tool(tool_name, arguments=None):
    async with streamablehttp_client(
        PD_MCP_URL,
        headers={"Authorization": f"Token token={PD_API_KEY}"}
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            return await session.call_tool(tool_name, arguments or {})


def print_result(result):
    """Pretty print tool results"""
    for content in result.content:
        if content.type == "text":
            try:
                print(json.dumps(json.loads(content.text), indent=2))
            except json.JSONDecodeError:
                print(content.text)


if __name__ == "__main__":
    if not PD_API_KEY:
        print("ERROR: Set PAGERDUTY_USER_API_KEY environment variable")
        print("  export PAGERDUTY_USER_API_KEY='u+your-token-here'")
        sys.exit(1)

    if len(sys.argv) < 2:
        try:
            tools = asyncio.run(list_tools())
            print(f"Found {len(tools.tools)} tools:\n")
            for t in tools.tools:
                desc = t.description[:100] if t.description else "No description"
                print(f"  {t.name}")
                print(f"    {desc}...\n")
        except Exception as e:
            print(f"Connection error: {e}")
            sys.exit(1)
    else:
        tool_name = sys.argv[1]
        args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}

        try:
            result = asyncio.run(call_tool(tool_name, args))
            print_result(result)
        except Exception as e:
            print(f"Error calling {tool_name}: {e}")
            sys.exit(1)