Rootly MCP server

Instructions

{
  "key": "value"
}

Implementation Reference

• src/rootly_mcp_server/server.py:337-344 (registration)

  Registers the listIncidentAlerts tool (and all other allowed API endpoints) dynamically from the filtered OpenAPI specification using FastMCP.from_openapi. The operationId 'listIncidentAlerts' from the spec becomes the MCP tool name.

  # By default, all routes become tools which is what we want
  mcp = FastMCP.from_openapi(
      openapi_spec=filtered_spec,
      client=http_client.client,
      name=name,
      timeout=30.0,
      tags={"rootly", "incident-management"},
  )

• src/rootly_mcp_server/server.py:175-277 (handler)

  The AuthenticatedHTTPXClient class provides the HTTP proxy handler logic used by all dynamically generated OpenAPI tools, including listIncidentAlerts. It transforms parameters, handles authentication, and makes requests to the Rootly API endpoint /v1/incidents/{incident_id}/alerts.

  class AuthenticatedHTTPXClient:
      """An HTTPX client wrapper that handles Rootly API authentication and parameter transformation."""
  
      def __init__(self, base_url: str = "https://api.rootly.com", hosted: bool = False, parameter_mapping: Optional[Dict[str, str]] = None):
          self._base_url = base_url
          self.hosted = hosted
          self._api_token = None
          self.parameter_mapping = parameter_mapping or {}
  
          if not self.hosted:
              self._api_token = self._get_api_token()
  
          # Create the HTTPX client  
          headers = {
              "Content-Type": "application/vnd.api+json", 
              "Accept": "application/vnd.api+json"
              # Let httpx handle Accept-Encoding automatically with all supported formats
          }
          if self._api_token:
              headers["Authorization"] = f"Bearer {self._api_token}"
  
          self.client = httpx.AsyncClient(
              base_url=base_url,
              headers=headers,
              timeout=30.0,
              follow_redirects=True,
              # Ensure proper handling of compressed responses
              limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
          )
  
      def _get_api_token(self) -> Optional[str]:
          """Get the API token from environment variables."""
          api_token = os.getenv("ROOTLY_API_TOKEN")
          if not api_token:
              logger.warning("ROOTLY_API_TOKEN environment variable is not set")
              return None
          return api_token
  
      def _transform_params(self, params: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
          """Transform sanitized parameter names back to original names."""
          if not params or not self.parameter_mapping:
              return params
  
          transformed = {}
          for key, value in params.items():
              # Use the original name if we have a mapping, otherwise keep the sanitized name
              original_key = self.parameter_mapping.get(key, key)
              transformed[original_key] = value
              if original_key != key:
                  logger.debug(f"Transformed parameter: '{key}' -> '{original_key}'")
          return transformed
  
      async def request(self, method: str, url: str, **kwargs):
          """Override request to transform parameters."""
          # Transform query parameters
          if 'params' in kwargs:
              kwargs['params'] = self._transform_params(kwargs['params'])
  
          # Call the underlying client's request method and let it handle everything
          return await self.client.request(method, url, **kwargs)
  
      async def get(self, url: str, **kwargs):
          """Proxy to request with GET method."""
          return await self.request('GET', url, **kwargs)
  
      async def post(self, url: str, **kwargs):
          """Proxy to request with POST method."""
          return await self.request('POST', url, **kwargs)
  
      async def put(self, url: str, **kwargs):
          """Proxy to request with PUT method."""
          return await self.request('PUT', url, **kwargs)
  
      async def patch(self, url: str, **kwargs):
          """Proxy to request with PATCH method."""
          return await self.request('PATCH', url, **kwargs)
  
      async def delete(self, url: str, **kwargs):
          """Proxy to request with DELETE method."""
          return await self.request('DELETE', url, **kwargs)
  
      async def __aenter__(self):
          return self
  
      async def __aexit__(self, exc_type, exc_val, exc_tb):
          pass
  
      def __getattr__(self, name):
          # Delegate all other attributes to the underlying client, except for request methods
          if name in ['request', 'get', 'post', 'put', 'patch', 'delete']:
              # Use our overridden methods instead
              return getattr(self, name)
          return getattr(self.client, name)
      
      @property 
      def base_url(self):
          return self._base_url
          
      @property
      def headers(self):
          return self.client.headers

• src/rootly_mcp_server/server.py:121-172 (registration)

  Includes the path "/incidents/{incident_id}/alerts" in DEFAULT_ALLOWED_PATHS, which is filtered into the OpenAPI spec and generates the listIncidentAlerts tool.

  DEFAULT_ALLOWED_PATHS = [
      "/incidents/{incident_id}/alerts",
      "/alerts",
      "/alerts/{alert_id}",
      "/severities",
      "/severities/{severity_id}",
      "/teams",
      "/teams/{team_id}",
      "/services",
      "/services/{service_id}",
      "/functionalities",
      "/functionalities/{functionality_id}",
      # Incident types
      "/incident_types",
      "/incident_types/{incident_type_id}",
      # Action items (all, by id, by incident)
      "/incident_action_items",
      "/incident_action_items/{incident_action_item_id}",
      "/incidents/{incident_id}/action_items",
      # Workflows
      "/workflows",
      "/workflows/{workflow_id}",
      # Workflow runs
      "/workflow_runs",
      "/workflow_runs/{workflow_run_id}",
      # Environments
      "/environments",
      "/environments/{environment_id}",
      # Users
      "/users",
      "/users/{user_id}",
      "/users/me",
      # Status pages
      "/status_pages",
      "/status_pages/{status_page_id}",
      # On-call schedules and shifts
      "/schedules",
      "/schedules/{schedule_id}",
      "/schedules/{schedule_id}/shifts",
      "/shifts",
      "/schedule_rotations/{schedule_rotation_id}",
      "/schedule_rotations/{schedule_rotation_id}/schedule_rotation_users",
      "/schedule_rotations/{schedule_rotation_id}/schedule_rotation_active_days",
      # On-call overrides
      "/schedules/{schedule_id}/override_shifts",
      "/override_shifts/{override_shift_id}",
      # On-call shadows and roles
      "/schedules/{schedule_id}/on_call_shadows",
      "/on_call_shadows/{on_call_shadow_id}",
      "/on_call_roles",
      "/on_call_roles/{on_call_role_id}",
  ]

• src/rootly_mcp_server/utils.py:45-105 (helper)

  Sanitizes parameter names in the OpenAPI spec to MCP-compliant format and creates mapping for request transformation, used for all tools including listIncidentAlerts.

  def sanitize_parameters_in_spec(spec: Dict[str, Any]) -> Dict[str, str]:
      """
      Sanitize all parameter names in an OpenAPI specification.
      
      This function modifies the spec in-place and builds a mapping 
      of sanitized names to original names.
      
      Args:
          spec: OpenAPI specification dictionary
          
      Returns:
          Dictionary mapping sanitized names to original names
      """
      parameter_mapping = {}
      
      # Sanitize parameters in paths
      if "paths" in spec:
          for path, path_item in spec["paths"].items():
              if not isinstance(path_item, dict):
                  continue
                  
              # Sanitize path-level parameters
              if "parameters" in path_item:
                  for param in path_item["parameters"]:
                      if "name" in param:
                          original_name = param["name"]
                          sanitized_name = sanitize_parameter_name(original_name)
                          if sanitized_name != original_name:
                              logger.debug(f"Sanitized path-level parameter: '{original_name}' -> '{sanitized_name}'")
                              param["name"] = sanitized_name
                              parameter_mapping[sanitized_name] = original_name
              
              # Sanitize operation-level parameters
              for method, operation in path_item.items():
                  if method.lower() not in ["get", "post", "put", "delete", "patch", "options", "head", "trace"]:
                      continue
                  if not isinstance(operation, dict):
                      continue
                      
                  if "parameters" in operation:
                      for param in operation["parameters"]:
                          if "name" in param:
                              original_name = param["name"]
                              sanitized_name = sanitize_parameter_name(original_name)
                              if sanitized_name != original_name:
                                  logger.debug(f"Sanitized operation parameter: '{original_name}' -> '{sanitized_name}'")
                                  param["name"] = sanitized_name
                                  parameter_mapping[sanitized_name] = original_name
      
      # Sanitize parameters in components (OpenAPI 3.0)
      if "components" in spec and "parameters" in spec["components"]:
          for param_name, param_def in spec["components"]["parameters"].items():
              if isinstance(param_def, dict) and "name" in param_def:
                  original_name = param_def["name"]
                  sanitized_name = sanitize_parameter_name(original_name)
                  if sanitized_name != original_name:
                      logger.debug(f"Sanitized component parameter: '{original_name}' -> '{sanitized_name}'")
                      param_def["name"] = sanitized_name
                      parameter_mapping[sanitized_name] = original_name
      
      return parameter_mapping

• src/rootly_mcp_server/server.py:1981-1993 (schema)

  Modifies the OpenAPI schema for alert endpoints (including /incidents/{incident_id}/alerts) by adding pagination parameters (page[size], page[number]), fields[alerts] sparse fieldset, and include parameter to optimize responses for the listIncidentAlerts tool.

  if "alert" in path.lower():
      # Add fields[alerts] parameter with essential fields only - make it required with default
      operation["parameters"].append({
          "name": "fields[alerts]",
          "in": "query",
          "required": True,
          "schema": {
              "type": "string",
              "default": "id,summary,status,started_at,ended_at,short_id,alert_urgency_id,source,noise",
              "description": "Comma-separated list of alert fields to include (reduces payload size)"
          }
      })