create_pipeline
Trigger new CI/CD pipelines in GitLab by specifying project, branch, and optional variables to automate software deployment and testing workflows.
Instructions
觸發新的 CI/CD Pipeline
Args: project_id: 專案 ID 或路徑 ref: 分支名稱或標籤 variables: Pipeline 變數（JSON 格式，如 '[{"key":"VAR1","value":"val1"}]'）
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes | ||
| ref | Yes | ||
| variables | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/gitlab_mcp/gitlab_client.py:319-604 (handler)The actual low-level GitLab API call implementation for creating a pipeline.
def create_pipeline( self, project_id: int | str, ref: str, variables: list[dict] = None ) -> dict: """POST /projects/:id/pipeline""" pid = self._resolve_project_id(project_id) data = {"ref": ref} if variables: data["variables"] = variables return self._post_json(f"/projects/{pid}/pipeline", data=data) def retry_pipeline(self, project_id: int | str, pipeline_id: int) -> dict: """POST /projects/:id/pipelines/:pipeline_id/retry""" pid = self._resolve_project_id(project_id) return self._post_json(f"/projects/{pid}/pipelines/{pipeline_id}/retry") def cancel_pipeline(self, project_id: int | str, pipeline_id: int) -> dict: """POST /projects/:id/pipelines/:pipeline_id/cancel""" pid = self._resolve_project_id(project_id) return self._post_json(f"/projects/{pid}/pipelines/{pipeline_id}/cancel") def list_pipeline_jobs( self, project_id: int | str, pipeline_id: int, page: int = 1, per_page: int = 20, ) -> list[dict]: """GET /projects/:id/pipelines/:pipeline_id/jobs""" pid = self._resolve_project_id(project_id) params = {"page": page, "per_page": per_page} return self._get_json( f"/projects/{pid}/pipelines/{pipeline_id}/jobs", params=params ) def get_job(self, project_id: int | str, job_id: int) -> dict: """GET /projects/:id/jobs/:job_id""" pid = self._resolve_project_id(project_id) return self._get_json(f"/projects/{pid}/jobs/{job_id}") def get_job_log(self, project_id: int | str, job_id: int) -> str: """GET /projects/:id/jobs/:job_id/trace — 回傳純文字日誌""" pid = self._resolve_project_id(project_id) return self._get_text(f"/projects/{pid}/jobs/{job_id}/trace") def retry_job(self, project_id: int | str, job_id: int) -> dict: """POST /projects/:id/jobs/:job_id/retry""" pid = self._resolve_project_id(project_id) return self._post_json(f"/projects/{pid}/jobs/{job_id}/retry") # ------------------------------------------------------------------ # Repository API # ------------------------------------------------------------------ def list_branches( self, project_id: int | str, search: str = None, page: int = 1, per_page: int = 20, ) -> list[dict]: """GET /projects/:id/repository/branches""" pid = self._resolve_project_id(project_id) params = {"page": page, "per_page": per_page} if search: params["search"] = search return self._get_json(f"/projects/{pid}/repository/branches", params=params) def get_branch(self, project_id: int | str, branch_name: str) -> dict: """GET /projects/:id/repository/branches/:branch""" pid = self._resolve_project_id(project_id) encoded_branch = quote(branch_name, safe="") return self._get_json( f"/projects/{pid}/repository/branches/{encoded_branch}" ) def list_commits( self, project_id: int | str, ref_name: str = None, path: str = None, page: int = 1, per_page: int = 20, ) -> list[dict]: """GET /projects/:id/repository/commits""" pid = self._resolve_project_id(project_id) params = {"page": page, "per_page": per_page} if ref_name: params["ref_name"] = ref_name if path: params["path"] = path return self._get_json(f"/projects/{pid}/repository/commits", params=params) def get_commit(self, project_id: int | str, sha: str) -> dict: """GET /projects/:id/repository/commits/:sha""" pid = self._resolve_project_id(project_id) return self._get_json(f"/projects/{pid}/repository/commits/{sha}") def compare_branches( self, project_id: int | str, from_ref: str, to_ref: str ) -> dict: """GET /projects/:id/repository/compare""" pid = self._resolve_project_id(project_id) params = {"from": from_ref, "to": to_ref} return self._get_json(f"/projects/{pid}/repository/compare", params=params) def list_repository_tree( self, project_id: int | str, path: str = "", ref: str = None, recursive: bool = False, page: int = 1, per_page: int = 20, ) -> list[dict]: """GET /projects/:id/repository/tree""" pid = self._resolve_project_id(project_id) params = {"page": page, "per_page": per_page} if path: params["path"] = path if ref: params["ref"] = ref if recursive: params["recursive"] = True return self._get_json(f"/projects/{pid}/repository/tree", params=params) # ------------------------------------------------------------------ # 全域客戶端實例 # ------------------------------------------------------------------ _client: Optional[GitLabClient] = None def get_client() -> GitLabClient: """取得全域客戶端實例（單例模式）""" global _client if _client is None: _client = GitLabClient() return _client def reload_client() -> GitLabClient: """重新載入客戶端（主要用於測試）""" global _client _client = None return get_client() - src/gitlab_mcp/server.py:578-587 (registration)The MCP tool registration and wrapper function in the server implementation.
@mcp.tool() def create_pipeline(project_id: int | str, ref: str, variables: str = None) -> str: """觸發新的 CI/CD Pipeline Args: project_id: 專案 ID 或路徑 ref: 分支名稱或標籤 variables: Pipeline 變數（JSON 格式，如 '[{"key":"VAR1","value":"val1"}]'） """