GitLab Pipeline MCP Server

import axios, { AxiosInstance } from 'axios'; import { logger } from '../utils/logger.js'; export interface PipelineVariable { [key: string]: string; } export interface Pipeline { id: number; status: string; ref: string; sha: string; web_url: string; created_at: string; updated_at: string; source: string; } export interface Job { id: number; name: string; status: string; stage: string; created_at: string; started_at?: string; finished_at?: string; web_url: string; } export interface ListPipelinesOptions { ref?: string; status?: string; per_page?: number; page?: number; } export class GitLabService { private createAxiosInstance(gitlabUrl: string, token: string): AxiosInstance { return axios.create({ baseURL: `${gitlabUrl}/api/v4`, headers: { 'Private-Token': token, 'Content-Type': 'application/json', }, timeout: 30000, }); } /** * 触发GitLab流水线 */ async triggerPipeline( gitlabUrl: string, projectId: string, ref: string, token: string, variables?: PipelineVariable ): Promise<Pipeline> { try { const api = this.createAxiosInstance(gitlabUrl, token); const data: any = { ref }; if (variables && Object.keys(variables).length > 0) { data.variables = Object.entries(variables).map(([key, value]) => ({ key, value, })); } logger.info(`触发流水线: ${projectId}@${ref}`); const response = await api.post(`/projects/${encodeURIComponent(projectId)}/pipeline`, data); logger.info(`流水线触发成功: ${response.data.id}`); return response.data; } catch (error) { logger.error('触发流水线失败:', error); if (axios.isAxiosError(error)) { throw new Error(`GitLab API 错误: ${error.response?.status} - ${error.response?.data?.message || error.message}`); } throw error; } } /** * 获取流水线状态 */ async getPipelineStatus( gitlabUrl: string, projectId: string, pipelineId: string, token: string ): Promise<Pipeline> { try { const api = this.createAxiosInstance(gitlabUrl, token); logger.info(`获取流水线状态: ${pipelineId}`); const response = await api.get(`/projects/${encodeURIComponent(projectId)}/pipelines/${pipelineId}`); return response.data; } catch (error) { logger.error('获取流水线状态失败:', error); if (axios.isAxiosError(error)) { throw new Error(`GitLab API 错误: ${error.response?.status} - ${error.response?.data?.message || error.message}`); } throw error; } } /** * 列出项目的流水线 */ async listPipelines( gitlabUrl: string, projectId: string, token: string, options: ListPipelinesOptions = {} ): Promise<Pipeline[]> { try { const api = this.createAxiosInstance(gitlabUrl, token); logger.info(`获取项目流水线列表: ${projectId}`); const params: any = { per_page: options.per_page || 20, page: options.page || 1, }; if (options.ref) params.ref = options.ref; if (options.status) params.status = options.status; const response = await api.get(`/projects/${encodeURIComponent(projectId)}/pipelines`, { params }); return response.data; } catch (error) { logger.error('获取流水线列表失败:', error); if (axios.isAxiosError(error)) { throw new Error(`GitLab API 错误: ${error.response?.status} - ${error.response?.data?.message || error.message}`); } throw error; } } /** * 获取流水线的作业列表 */ async getPipelineJobs( gitlabUrl: string, projectId: string, pipelineId: string, token: string ): Promise<Job[]> { try { const api = this.createAxiosInstance(gitlabUrl, token); logger.info(`获取流水线作业: ${pipelineId}`); const response = await api.get(`/projects/${encodeURIComponent(projectId)}/pipelines/${pipelineId}/jobs`); return response.data; } catch (error) { logger.error('获取流水线作业失败:', error); if (axios.isAxiosError(error)) { throw new Error(`GitLab API 错误: ${error.response?.status} - ${error.response?.data?.message || error.message}`); } throw error; } } /** * 取消流水线 */ async cancelPipeline( gitlabUrl: string, projectId: string, pipelineId: string, token: string ): Promise<Pipeline> { try { const api = this.createAxiosInstance(gitlabUrl, token); logger.info(`取消流水线: ${pipelineId}`); const response = await api.post(`/projects/${encodeURIComponent(projectId)}/pipelines/${pipelineId}/cancel`); return response.data; } catch (error) { logger.error('取消流水线失败:', error); if (axios.isAxiosError(error)) { throw new Error(`GitLab API 错误: ${error.response?.status} - ${error.response?.data?.message || error.message}`); } throw error; } } /** * 重试流水线 */ async retryPipeline( gitlabUrl: string, projectId: string, pipelineId: string, token: string ): Promise<Pipeline> { try { const api = this.createAxiosInstance(gitlabUrl, token); logger.info(`重试流水线: ${pipelineId}`); const response = await api.post(`/projects/${encodeURIComponent(projectId)}/pipelines/${pipelineId}/retry`); return response.data; } catch (error) { logger.error('重试流水线失败:', error); if (axios.isAxiosError(error)) { throw new Error(`GitLab API 错误: ${error.response?.status} - ${error.response?.data?.message || error.message}`); } throw error; } } /** * 获取作业日志 */ async getJobLog( gitlabUrl: string, projectId: string, jobId: string, token: string ): Promise<string> { try { const api = this.createAxiosInstance(gitlabUrl, token); logger.info(`获取作业日志: ${jobId}`); const response = await api.get(`/projects/${encodeURIComponent(projectId)}/jobs/${jobId}/trace`); return response.data; } catch (error) { logger.error('获取作业日志失败:', error); if (axios.isAxiosError(error)) { throw new Error(`GitLab API 错误: ${error.response?.status} - ${error.response?.data?.message || error.message}`); } throw error; } } }