GitLab MCP Server

Instructions

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"}]'）
      """