Rootly MCP server

Instructions

{
  "key": "value"
}

{
  "key": "value"
}

{
  "key": "value"
}

Implementation Reference

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

  Registers the createFunctionality tool via FastMCP.from_openapi using the filtered OpenAPI specification which includes the /functionalities path.

  # 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:319-321 (schema)

  Sanitizes parameter names in the OpenAPI spec for the createFunctionality tool (and others) to ensure MCP compliance.

  parameter_mapping = sanitize_parameters_in_spec(filtered_spec)
  logger.info(f"Sanitized parameter names for MCP compatibility (mapped {len(parameter_mapping)} parameters)")

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

  The AuthenticatedHTTPXClient class provides the core request handling logic used by all generated MCP tools, including createFunctionality, to proxy requests to the Rootly API.

  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 (helper)

  Defines the default allowed API paths, including /functionalities and /functionalities/{functionality_id}, which enables generation of createFunctionality (POST /v1/functionalities) and related tools.

  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)

  Utility function that sanitizes OpenAPI parameter names across the entire spec for MCP tool schema compatibility.

  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