diff options
Diffstat (limited to 'servers/gitlab_glab/src/mcp_server_gitlab_glab')
| -rw-r--r-- | servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py | 88 |
1 files changed, 55 insertions, 33 deletions
diff --git a/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py b/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py index 38a9eb6..e6c4d47 100644 --- a/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py +++ b/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py @@ -32,11 +32,14 @@ class GitLabServer: "Authentication required. Please run 'glab auth login' to authenticate." ) - def execute_glab_command(self, args: list[str]) -> tuple[bool, Any]: + def execute_glab_command( + self, args: list[str], working_directory: str + ) -> tuple[bool, Any]: """Execute a glab command and return the result. Args: args: List of command arguments to pass to glab. + working_directory: The directory to execute the command in. Returns: A tuple containing: @@ -49,16 +52,17 @@ class GitLabServer: capture_output=True, text=True, check=False, + cwd=working_directory, ) if result.returncode != 0: error_msg = result.stderr.strip() logger.error(f"glab command failed: {error_msg}") - + # Check for authentication errors if "authentication required" in error_msg.lower(): return False, {"error": self.auth_message} - + return False, {"error": error_msg} # For API commands, parse JSON output @@ -79,14 +83,17 @@ class GitLabServer: logger.error(f"Command execution failed: {str(e)}") return False, {"error": f"Command execution failed: {str(e)}"} - def check_availability(self) -> dict[str, Any]: + def check_availability(self, working_directory: str) -> dict[str, Any]: """Check if the glab CLI tool is available and accessible. + Args: + working_directory: The directory to execute the command in. + Returns: A dictionary containing availability status and version information. """ - success, result = self.execute_glab_command(["--version"]) - + success, result = self.execute_glab_command(["--version"], working_directory) + if success: return { "available": True, @@ -98,33 +105,41 @@ class GitLabServer: "error": result.get("error", "Unknown error"), } - def find_project(self, project_name: str) -> dict[str, Any]: - """Find a GitLab project by name. + def find_project( + self, project_name: str, working_directory: str + ) -> dict[str, Any] | list[dict[str, Any]]: + """Find GitLab projects by name. Args: project_name: The name of the project to search for. + working_directory: The directory to execute the command in. Returns: - A dictionary containing project information if found, or an error message. + A list of dictionaries containing project information if found, + or an error message. """ success, result = self.execute_glab_command( - ["api", f"/projects?search={project_name}"] + ["api", f"/projects?search_namespaces=true&search={project_name}"], working_directory ) - + if not success: return result - + # Check if any projects were found if isinstance(result, list) and len(result) > 0: - # Return the first matching project - project = result[0] - return { - "id": project.get("id"), - "name": project.get("name"), - "path_with_namespace": project.get("path_with_namespace"), - "web_url": project.get("web_url"), - "description": project.get("description"), - } + # Return all matching projects + projects = [] + for project in result: + projects.append( + { + "id": project.get("id"), + "name": project.get("name"), + "path_with_namespace": project.get("path_with_namespace"), + "web_url": project.get("web_url"), + "description": project.get("description"), + } + ) + return projects else: return {"error": f"Project '{project_name}' not found"} @@ -141,33 +156,40 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP: """ # Create a FastMCP server with host and port settings mcp = FastMCP("GitLab CLI", host=host, port=port) - + # Create a GitLabServer instance gitlab = GitLabServer() - + # Add check_glab_availability tool @mcp.tool() - def check_glab_availability() -> dict[str, Any]: + def check_glab_availability(working_directory: str) -> dict[str, Any]: """Check if the glab CLI tool is available and accessible. + Args: + working_directory: The directory to execute the command in. + Returns: A dictionary containing availability status and version information. """ - return gitlab.check_availability() - + return gitlab.check_availability(working_directory) + # Add find_project tool @mcp.tool() - def find_project(project_name: str) -> dict[str, Any]: - """Find a GitLab project by name and return its ID. + def find_project( + project_name: str, working_directory: str + ) -> dict[str, Any] | list[dict[str, Any]]: + """Find GitLab projects by name. Args: project_name: The name of the project to search for. + working_directory: The directory to execute the command in. Returns: - A dictionary containing project information if found, or an error message. + A list of dictionaries containing project information if found, + or an error message. """ - return gitlab.find_project(project_name) - + return gitlab.find_project(project_name, working_directory) + return mcp @@ -181,10 +203,10 @@ async def main(transport_type: str, host: str, port: int) -> None: """ logger.info("Starting MCP GitLab CLI Server") logger.info(f"Starting GitLab CLI MCP Server with {transport_type} transport") - + # Create the server with host and port mcp = create_server(host=host, port=port) - + # Run the server with the appropriate transport if transport_type == "stdio": logger.info("Server running with stdio transport") |