Rootly MCP server

Instructions

Implementation Reference

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

  Registers the 'createService' tool (and all other OpenAPI-derived tools) using FastMCP.from_openapi with the filtered Rootly API spec containing the /services POST endpoint.
  # Create the MCP server using OpenAPI integration # 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:121-172 (registration)

  Defines the allowed API paths including '/services' (for createService POST) and '/services/{service_id}' which are filtered into the OpenAPI spec for tool generation.
  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/server.py:175-277 (helper)

  Custom HTTP client used by all generated tools (including createService) to make authenticated requests to the Rootly API, with parameter transformation.
  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/utils.py:45-105 (helper)

  Sanitizes parameter names in the OpenAPI spec to MCP-compliant format before tool generation; used for createService schema.
  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:376-392 (handler)

  Internal async request function used by custom tools, but generated tools use the http_client directly via FastMCP.
  """Make an authenticated request, extracting token from MCP headers in hosted mode.""" # In hosted mode, get token from MCP request headers if hosted: try: from fastmcp.server.dependencies import get_http_headers request_headers = get_http_headers() auth_header = request_headers.get("authorization", "") if auth_header: # Add authorization header to the request if "headers" not in kwargs: kwargs["headers"] = {} kwargs["headers"]["Authorization"] = auth_header except Exception: pass # Fallback to default client behavior # Use our custom client with proper error handling instead of bypassing it return await http_client.request(method, url, **kwargs)