Add package solidlsp with copies of classes from multilspy (unmodified)

pyproject.toml

@@ -112,10 +112,10 @@ PYDEVD_DISABLE_FILE_VALIDATION = "1"
 # For custom markers, one can either adjust the env var or just use -m option in the command line,
 # as the second -m option will override the first one.
 test = "pytest test -vv -m \"${PYTEST_MARKERS:-not java and not rust and not isolated_process}\""
-_black_check = "black --check --exclude src/multilspy/ src scripts test"
-_ruff_check = "ruff check --exclude .venv/ --exclude src/multilspy/ src scripts test"
-_black_format = "black --exclude .venv/|src/multilspy/ src scripts test"
-_ruff_format = "ruff check --exclude src/multilspy/ --fix src scripts test"
+_black_check = "black --check --exclude src/multilspy/ --exclude src/solidlsp/ src scripts test"
+_ruff_check = "ruff check --exclude .venv/ --exclude src/multilspy/ --exclude src/solidlsp/ src scripts test"
+_black_format = "black --exclude .venv/|src/multilspy/|src/solidlsp/ src scripts test"
+_ruff_format = "ruff check --exclude src/multilspy/ --exclude src/solidlsp/ --fix src scripts test"
 lint = [
   "_black_check",
   "_ruff_check",

src/solidlsp/ls_handler.py

@@ -0,0 +1,531 @@
+import asyncio
+import dataclasses
+import json
+import logging
+import os
+import subprocess
+import threading
+
+import psutil
+from typing import Any, Callable, Dict, List, Optional, Union
+
+from multilspy.lsp_protocol_handler.lsp_requests import LspNotification, LspRequest
+from multilspy.lsp_protocol_handler.lsp_types import ErrorCodes
+from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo, StringDict, Request, content_length, ENCODING, Error, MessageType, \
+    make_notification, PayloadLike, make_response, make_error_response, make_request, create_message
+from multilspy.multilspy_exceptions import MultilspyException
+
+
+log = logging.getLogger(__name__)
+
+
+class LanguageServerHandler:
+    """
+    This class provides the implementation of Python client for the Language Server Protocol.
+    A class that launches the language server and communicates with it
+    using the Language Server Protocol (LSP).
+
+    It provides methods for sending requests, responses, and notifications to the server
+    and for registering handlers for requests and notifications from the server.
+
+    Uses JSON-RPC 2.0 for communication with the server over stdin/stdout.
+
+    Attributes:
+        send: A LspRequest object that can be used to send requests to the server and
+            await for the responses.
+        notify: A LspNotification object that can be used to send notifications to the server.
+        cmd: A string that represents the command to launch the language server process.
+        process: A subprocess.Popen object that represents the language server process.
+        _received_shutdown: A boolean flag that indicates whether the client has received
+            a shutdown request from the server.
+        request_id: An integer that represents the next available request id for the client.
+        _response_handlers: A dictionary that maps request ids to Request objects that
+            store the results or errors of the requests.
+        on_request_handlers: A dictionary that maps method names to callback functions
+            that handle requests from the server.
+        on_notification_handlers: A dictionary that maps method names to callback functions
+            that handle notifications from the server.
+        logger: An optional function that takes two strings (source and destination) and
+            a payload dictionary, and logs the communication between the client and the server.
+        tasks: A dictionary that maps task ids to asyncio.Task objects that represent
+            the asynchronous tasks created by the handler.
+        task_counter: An integer that represents the next available task id for the handler.
+        loop: An asyncio.AbstractEventLoop object that represents the event loop used by the handler.
+        start_independent_lsp_process: An optional boolean flag that indicates whether to start the
+        language server process in an independent process group. Default is `True`. Setting it to
+        `False` means that the language server process will be in the same process group as the
+        the current process, and any SIGINT and SIGTERM signals will be sent to both processes.
+    """
+
+    def __init__(
+            self,
+            process_launch_info: ProcessLaunchInfo,
+            logger: Optional[Callable[[str, str, StringDict | str], None]] = None,
+            start_independent_lsp_process=True,
+    ) -> None:
+        """
+        Params:
+            cmd: A string that represents the command to launch the language server process.
+            logger: An optional function that takes two strings (source and destination) and
+                a payload dictionary, and logs the communication between the client and the server.
+        """
+        self.send = LspRequest(self.send_request)
+        self.notify = LspNotification(self.send_notification)
+
+        self.process_launch_info = process_launch_info
+        self.process = None
+        self._received_shutdown = False
+
+        self.request_id = 1
+        self._response_handlers: Dict[Any, Request] = {}
+        self.on_request_handlers = {}
+        self.on_notification_handlers = {}
+        self.logger = logger
+        self.tasks = {}
+        self.task_counter = 0
+        self.loop = None
+        self.start_independent_lsp_process = start_independent_lsp_process
+
+        # Add thread locks for shared resources to prevent race conditions
+        self._stdin_lock = threading.Lock()
+        self._request_id_lock = threading.Lock()
+        self._response_handlers_lock = threading.Lock()
+        self._tasks_lock = threading.Lock()
+
+
+
+    def is_running(self) -> bool:
+        """
+        Checks if the language server process is currently running.
+        """
+        return self.process is not None and self.process.returncode is None
+
+    async def start(self) -> None:
+        """
+        Starts the language server process and creates a task to continuously read from its stdout to handle communications
+        from the server to the client
+        """
+        child_proc_env = os.environ.copy()
+        child_proc_env.update(self.process_launch_info.env)
+
+        log.info("Starting language server process via command: %s", self.process_launch_info.cmd)
+        self.process = await asyncio.create_subprocess_shell(
+            self.process_launch_info.cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stdin=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            env=child_proc_env,
+            cwd=self.process_launch_info.cwd,
+            start_new_session=self.start_independent_lsp_process,
+        )
+
+        # Check if process terminated immediately
+        if self.process.returncode is not None:
+            log.error("Language server has already terminated/could not be started")
+            # Process has already terminated
+            stderr_data = await self.process.stderr.read()
+            error_message = stderr_data.decode('utf-8', errors='replace')
+            raise RuntimeError(f"Process terminated immediately with code {self.process.returncode}. Error: {error_message}")
+
+        self.loop = asyncio.get_event_loop()
+
+        # Use lock to prevent race conditions on tasks and task_counter during startup
+        with self._tasks_lock:
+            self.tasks[self.task_counter] = self.loop.create_task(self.run_forever())
+            self.task_counter += 1
+            self.tasks[self.task_counter] = self.loop.create_task(self.run_forever_stderr())
+            self.task_counter += 1
+
+
+
+    async def stop(self) -> None:
+        """
+        Sends the terminate signal to the language server process and waits for it to exit, with a timeout, killing it if necessary
+        """
+        # First cancel all tasks
+        await self._cancel_pending_tasks()
+
+        process = self.process
+        self.process = None
+
+        if not process:
+            return
+
+        # Clean up the process
+        await self._cleanup_process(process)
+
+    async def _cancel_pending_tasks(self):
+        """Cancel all pending tasks and wait for them to complete or timeout."""
+        pending_tasks = []
+
+        # Use lock to safely access tasks dictionary
+        with self._tasks_lock:
+            for task in self.tasks.values():
+                if not task.done():
+                    task.cancel()
+                    pending_tasks.append(task)
+
+        if pending_tasks:
+            try:
+                await asyncio.wait_for(asyncio.gather(*pending_tasks, return_exceptions=True), timeout=5.0)
+            except (asyncio.TimeoutError, Exception):
+                pass
+
+        # Clear tasks dictionary under lock
+        with self._tasks_lock:
+            self.tasks = {}
+
+    async def _cleanup_process(self, process):
+        """Clean up a process: close stdin, terminate/kill process, close stdout/stderr."""
+        # Close stdin first to prevent deadlocks
+        # See: https://bugs.python.org/issue35539
+        self._safely_close_pipe(process.stdin)
+
+        # Terminate/kill the process if it's still running
+        if process.returncode is None:
+            await self._terminate_or_kill_process(process)
+
+        # Close stdout and stderr pipes after process has exited
+        # This is essential to prevent "I/O operation on closed pipe" errors and
+        # "Event loop is closed" errors during garbage collection
+        # See: https://bugs.python.org/issue41320 and https://github.com/python/cpython/issues/88050
+        self._safely_close_pipe(process.stdout)
+        self._safely_close_pipe(process.stderr)
+
+        # Small delay to ensure OS has released file handles
+        await asyncio.sleep(0.5)
+
+    def _safely_close_pipe(self, pipe):
+        """Safely close a pipe, ignoring any exceptions."""
+        if pipe:
+            try:
+                pipe.close()
+            except Exception:
+                pass
+
+    async def _terminate_or_kill_process(self, process):
+        """Try to terminate the process gracefully, then forcefully if necessary."""
+        # First try to terminate the process tree gracefully
+        self._signal_process_tree(process, terminate=True)
+
+        # Wait for the process to exit (with timeout)
+        try:
+            await asyncio.wait_for(process.wait(), timeout=10)
+        except (asyncio.TimeoutError, Exception):
+            # If termination failed, forcefully kill the process tree
+            self._signal_process_tree(process, terminate=False)
+            try:
+                # Give it one more chance to exit
+                await asyncio.wait_for(process.wait(), timeout=2)
+            except Exception:
+                pass
+
+    def _signal_process_tree(self, process, terminate=True):
+        """Send signal (terminate or kill) to the process and all its children."""
+        signal_method = "terminate" if terminate else "kill"
+
+        # Try to get the parent process
+        parent = None
+        try:
+            parent = psutil.Process(process.pid)
+        except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
+            pass
+
+        # If we have the parent process and it's running, signal the entire tree
+        if parent and parent.is_running():
+            # Signal children first
+            for child in parent.children(recursive=True):
+                try:
+                    getattr(child, signal_method)()
+                except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
+                    pass
+
+            # Then signal the parent
+            try:
+                getattr(parent, signal_method)()
+            except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
+                pass
+        else:
+            # Fall back to direct process signaling
+            try:
+                getattr(process, signal_method)()
+            except Exception:
+                pass
+
+
+    async def shutdown(self) -> None:
+        """
+        Perform the shutdown sequence for the client, including sending the shutdown request to the server and notifying it of exit
+        """
+        self._log("Sending shutdown request to server")
+        await self.send.shutdown()
+        self._log("Received shutdown response from server")
+        self._received_shutdown = True
+        self._log("Sending exit notification to server")
+        self.notify.exit()
+        self._log("Sent exit notification to server")
+        if self.process and self.process.stdout:
+            self.process.stdout.set_exception(StopLoopException())
+            # This yields the control to the event loop to allow the exception to be handled
+            # in the run_forever and run_forever_stderr methods
+            await asyncio.sleep(0)
+
+    def _log(self, message: str | StringDict) -> None:
+        """
+        Create a log message
+        """
+        if self.logger is not None:
+            self.logger("client", "logger", message)
+
+    async def run_forever(self) -> bool:
+        """
+        Continuously read from the language server process stdout and handle the messages
+        invoking the registered response and notification handlers
+        """
+        try:
+            while self.process and self.process.stdout and not self.process.stdout.at_eof():
+                line = await self.process.stdout.readline()
+                if not line:
+                    continue
+                try:
+                    num_bytes = content_length(line)
+                except ValueError:
+                    continue
+                if num_bytes is None:
+                    continue
+                while line and line.strip():
+                    line = await self.process.stdout.readline()
+                if not line:
+                    continue
+                body = await self.process.stdout.readexactly(num_bytes)
+
+                # Use lock to prevent race conditions on tasks and task_counter
+                with self._tasks_lock:
+                    self.tasks[self.task_counter] = asyncio.get_event_loop().create_task(self._handle_body(body))
+                    self.task_counter += 1
+        except (BrokenPipeError, ConnectionResetError, StopLoopException):
+            pass
+        return self._received_shutdown
+
+
+    async def run_forever_stderr(self) -> None:
+        """
+        Continuously read from the language server process stderr and log the messages
+        """
+        try:
+            while self.process and self.process.stderr and not self.process.stderr.at_eof():
+                line = await self.process.stderr.readline()
+                if not line:
+                    continue
+                self._log("LSP stderr: " + line.decode(ENCODING, errors='replace'))
+        except (BrokenPipeError, ConnectionResetError, StopLoopException):
+            pass
+
+    async def _handle_body(self, body: bytes) -> None:
+        """
+        Parse the body text received from the language server process and invoke the appropriate handler
+        """
+        try:
+            await self._receive_payload(json.loads(body))
+        except IOError as ex:
+            self._log(f"malformed {ENCODING}: {ex}")
+        except UnicodeDecodeError as ex:
+            self._log(f"malformed {ENCODING}: {ex}")
+        except json.JSONDecodeError as ex:
+            self._log(f"malformed JSON: {ex}")
+
+    async def _receive_payload(self, payload: StringDict) -> None:
+        """
+        Determine if the payload received from server is for a request, response, or notification and invoke the appropriate handler
+        """
+        if self.logger:
+            self.logger("server", "client", payload)
+        try:
+            if "method" in payload:
+                if "id" in payload:
+                    await self._request_handler(payload)
+                else:
+                    await self._notification_handler(payload)
+            elif "id" in payload:
+                await self._response_handler(payload)
+            else:
+                self._log(f"Unknown payload type: {payload}")
+        except Exception as err:
+            self._log(f"Error handling server payload: {err}")
+
+    def send_notification(self, method: str, params: Optional[dict] = None) -> None:
+        """
+        Send notification pertaining to the given method to the server with the given parameters
+        """
+        self._send_payload_sync(make_notification(method, params))
+
+    def send_response(self, request_id: Any, params: PayloadLike) -> None:
+        """
+        Send response to the given request id to the server with the given parameters
+        """
+        # Use lock to prevent race conditions on tasks and task_counter
+        with self._tasks_lock:
+            self.tasks[self.task_counter] = asyncio.get_event_loop().create_task(
+                self._send_payload(make_response(request_id, params))
+            )
+            self.task_counter += 1
+
+
+    def send_error_response(self, request_id: Any, err: Error) -> None:
+        """
+        Send error response to the given request id to the server with the given error
+        """
+        # Use lock to prevent race conditions on tasks and task_counter
+        with self._tasks_lock:
+            self.tasks[self.task_counter] = asyncio.get_event_loop().create_task(
+                self._send_payload(make_error_response(request_id, err))
+            )
+            self.task_counter += 1
+
+
+    async def send_request(self, method: str, params: Optional[dict] = None) -> PayloadLike:
+        """
+        Send request to the server, register the request id, and wait for the response
+        """
+        request = Request()
+
+        # Use lock to prevent race conditions on request_id and _response_handlers
+        with self._request_id_lock:
+            request_id = self.request_id
+            self.request_id += 1
+
+        with self._response_handlers_lock:
+            self._response_handlers[request_id] = request
+
+        async with request.cv:
+            await self._send_payload(make_request(method, request_id, params))
+            self._log(f"Waiting for asyncio condition for request {method} with params:\n{params}")
+            await request.cv.wait()
+        self._log(f"Finished waiting, processing result")
+        if isinstance(request.error, Error):
+            raise MultilspyException(f"Could not process request {method} with params:\n{params}.\n  Language server error: {request.error}") from request.error
+        self._log(f"Returning non-error result, which is:\n{request.result}")
+        return request.result
+
+
+    def _send_payload_sync(self, payload: StringDict) -> None:
+        """
+        Send the payload to the server by writing to its stdin synchronously
+        """
+        if not self.process or not self.process.stdin:
+            return
+        msg = create_message(payload)
+        if self.logger:
+            self.logger("client", "server", payload)
+
+        # Use lock to prevent concurrent writes to stdin that cause buffer corruption
+        with self._stdin_lock:
+            try:
+                self.process.stdin.writelines(msg)
+            except (BrokenPipeError, ConnectionResetError, OSError) as e:
+                # Log the error but don't raise to prevent cascading failures
+                if self.logger:
+                    self.logger("client", "logger", f"Failed to write to stdin: {e}")
+                return
+
+
+    async def _send_payload(self, payload: StringDict) -> None:
+        """
+        Send the payload to the server by writing to its stdin asynchronously.
+        """
+        if not self.process or not self.process.stdin:
+            return
+        self._log(payload)
+        msg = create_message(payload)
+
+        # Use lock to prevent concurrent writes to stdin that cause buffer corruption
+        with self._stdin_lock:
+            try:
+                self.process.stdin.writelines(msg)
+                await self.process.stdin.drain()
+            except (BrokenPipeError, ConnectionResetError, OSError) as e:
+                # Log the error but don't raise to prevent cascading failures
+                if self.logger:
+                    self.logger("client", "logger", f"Failed to write to stdin: {e}")
+                return
+
+
+    def on_request(self, method: str, cb) -> None:
+        """
+        Register the callback function to handle requests from the server to the client for the given method
+        """
+        self.on_request_handlers[method] = cb
+
+    def on_notification(self, method: str, cb) -> None:
+        """
+        Register the callback function to handle notifications from the server to the client for the given method
+        """
+        self.on_notification_handlers[method] = cb
+
+    async def _response_handler(self, response: StringDict) -> None:
+        """
+        Handle the response received from the server for a request, using the id to determine the request
+        """
+        with self._response_handlers_lock:
+            request = self._response_handlers.pop(response["id"])
+
+        if "result" in response and "error" not in response:
+            await request.on_result(response["result"])
+        elif "result" not in response and "error" in response:
+            await request.on_error(Error.from_lsp(response["error"]))
+        else:
+            await request.on_error(Error(ErrorCodes.InvalidRequest, ""))
+
+
+    async def _request_handler(self, response: StringDict) -> None:
+        """
+        Handle the request received from the server: call the appropriate callback function and return the result
+        """
+        method = response.get("method", "")
+        params = response.get("params")
+        request_id = response.get("id")
+        handler = self.on_request_handlers.get(method)
+        if not handler:
+            self.send_error_response(
+                request_id,
+                Error(
+                    ErrorCodes.MethodNotFound,
+                    "method '{}' not handled on client.".format(method),
+                ),
+            )
+            return
+        try:
+            self.send_response(request_id, await handler(params))
+        except Error as ex:
+            self.send_error_response(request_id, ex)
+        except Exception as ex:
+            self.send_error_response(request_id, Error(ErrorCodes.InternalError, str(ex)))
+
+    async def _notification_handler(self, response: StringDict) -> None:
+        """
+        Handle the notification received from the server: call the appropriate callback function
+        """
+        method = response.get("method", "")
+        params = response.get("params")
+        handler = self.on_notification_handlers.get(method)
+        if not handler:
+            self._log(f"unhandled {method}")
+            return
+        try:
+            await handler(params)
+        except asyncio.CancelledError:
+            return
+        except Exception as ex:
+            if (not self._received_shutdown) and self.logger:
+                self.logger(
+                    "client",
+                    "logger",
+                    str(
+                        {
+                            "type": MessageType.error,
+                            "message": str(ex),
+                            "method": method,
+                            "params": params,
+                        }
+                    ),
+                )

src/solidlsp/lsp_request.py

@@ -0,0 +1,435 @@
+from typing import List, Union
+from multilspy.lsp_protocol_handler import lsp_types
+
+
+class LspRequest:
+    def __init__(self, send_request):
+        self.send_request = send_request
+
+    async def implementation(
+            self, params: lsp_types.ImplementationParams
+    ) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
+        """A request to resolve the implementation locations of a symbol at a given text
+        document position. The request's parameter is of type [TextDocumentPositionParams]
+        (#TextDocumentPositionParams) the response is of type {@link Definition} or a
+        Thenable that resolves to such."""
+        return await self.send_request("textDocument/implementation", params)
+
+    async def type_definition(
+            self, params: lsp_types.TypeDefinitionParams
+    ) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
+        """A request to resolve the type definition locations of a symbol at a given text
+        document position. The request's parameter is of type [TextDocumentPositionParams]
+        (#TextDocumentPositionParams) the response is of type {@link Definition} or a
+        Thenable that resolves to such."""
+        return await self.send_request("textDocument/typeDefinition", params)
+
+    async def document_color(
+            self, params: lsp_types.DocumentColorParams
+    ) -> List["lsp_types.ColorInformation"]:
+        """A request to list all color symbols found in a given text document. The request's
+        parameter is of type {@link DocumentColorParams} the
+        response is of type {@link ColorInformation ColorInformation[]} or a Thenable
+        that resolves to such."""
+        return await self.send_request("textDocument/documentColor", params)
+
+    async def color_presentation(
+            self, params: lsp_types.ColorPresentationParams
+    ) -> List["lsp_types.ColorPresentation"]:
+        """A request to list all presentation for a color. The request's
+        parameter is of type {@link ColorPresentationParams} the
+        response is of type {@link ColorInformation ColorInformation[]} or a Thenable
+        that resolves to such."""
+        return await self.send_request("textDocument/colorPresentation", params)
+
+    async def folding_range(
+            self, params: lsp_types.FoldingRangeParams
+    ) -> Union[List["lsp_types.FoldingRange"], None]:
+        """A request to provide folding ranges in a document. The request's
+        parameter is of type {@link FoldingRangeParams}, the
+        response is of type {@link FoldingRangeList} or a Thenable
+        that resolves to such."""
+        return await self.send_request("textDocument/foldingRange", params)
+
+    async def declaration(
+            self, params: lsp_types.DeclarationParams
+    ) -> Union["lsp_types.Declaration", List["lsp_types.LocationLink"], None]:
+        """A request to resolve the type definition locations of a symbol at a given text
+        document position. The request's parameter is of type [TextDocumentPositionParams]
+        (#TextDocumentPositionParams) the response is of type {@link Declaration}
+        or a typed array of {@link DeclarationLink} or a Thenable that resolves
+        to such."""
+        return await self.send_request("textDocument/declaration", params)
+
+    async def selection_range(
+            self, params: lsp_types.SelectionRangeParams
+    ) -> Union[List["lsp_types.SelectionRange"], None]:
+        """A request to provide selection ranges in a document. The request's
+        parameter is of type {@link SelectionRangeParams}, the
+        response is of type {@link SelectionRange SelectionRange[]} or a Thenable
+        that resolves to such."""
+        return await self.send_request("textDocument/selectionRange", params)
+
+    async def prepare_call_hierarchy(
+            self, params: lsp_types.CallHierarchyPrepareParams
+    ) -> Union[List["lsp_types.CallHierarchyItem"], None]:
+        """A request to result a `CallHierarchyItem` in a document at a given position.
+        Can be used as an input to an incoming or outgoing call hierarchy.
+
+        @since 3.16.0"""
+        return await self.send_request("textDocument/prepareCallHierarchy", params)
+
+    async def incoming_calls(
+            self, params: lsp_types.CallHierarchyIncomingCallsParams
+    ) -> Union[List["lsp_types.CallHierarchyIncomingCall"], None]:
+        """A request to resolve the incoming calls for a given `CallHierarchyItem`.
+
+        @since 3.16.0"""
+        return await self.send_request("callHierarchy/incomingCalls", params)
+
+    async def outgoing_calls(
+            self, params: lsp_types.CallHierarchyOutgoingCallsParams
+    ) -> Union[List["lsp_types.CallHierarchyOutgoingCall"], None]:
+        """A request to resolve the outgoing calls for a given `CallHierarchyItem`.
+
+        @since 3.16.0"""
+        return await self.send_request("callHierarchy/outgoingCalls", params)
+
+    async def semantic_tokens_full(
+            self, params: lsp_types.SemanticTokensParams
+    ) -> Union["lsp_types.SemanticTokens", None]:
+        """@since 3.16.0"""
+        return await self.send_request("textDocument/semanticTokens/full", params)
+
+    async def semantic_tokens_delta(
+            self, params: lsp_types.SemanticTokensDeltaParams
+    ) -> Union["lsp_types.SemanticTokens", "lsp_types.SemanticTokensDelta", None]:
+        """@since 3.16.0"""
+        return await self.send_request("textDocument/semanticTokens/full/delta", params)
+
+    async def semantic_tokens_range(
+            self, params: lsp_types.SemanticTokensRangeParams
+    ) -> Union["lsp_types.SemanticTokens", None]:
+        """@since 3.16.0"""
+        return await self.send_request("textDocument/semanticTokens/range", params)
+
+    async def linked_editing_range(
+            self, params: lsp_types.LinkedEditingRangeParams
+    ) -> Union["lsp_types.LinkedEditingRanges", None]:
+        """A request to provide ranges that can be edited together.
+
+        @since 3.16.0"""
+        return await self.send_request("textDocument/linkedEditingRange", params)
+
+    async def will_create_files(
+            self, params: lsp_types.CreateFilesParams
+    ) -> Union["lsp_types.WorkspaceEdit", None]:
+        """The will create files request is sent from the client to the server before files are actually
+        created as long as the creation is triggered from within the client.
+
+        @since 3.16.0"""
+        return await self.send_request("workspace/willCreateFiles", params)
+
+    async def will_rename_files(
+            self, params: lsp_types.RenameFilesParams
+    ) -> Union["lsp_types.WorkspaceEdit", None]:
+        """The will rename files request is sent from the client to the server before files are actually
+        renamed as long as the rename is triggered from within the client.
+
+        @since 3.16.0"""
+        return await self.send_request("workspace/willRenameFiles", params)
+
+    async def will_delete_files(
+            self, params: lsp_types.DeleteFilesParams
+    ) -> Union["lsp_types.WorkspaceEdit", None]:
+        """The did delete files notification is sent from the client to the server when
+        files were deleted from within the client.
+
+        @since 3.16.0"""
+        return await self.send_request("workspace/willDeleteFiles", params)
+
+    async def moniker(
+            self, params: lsp_types.MonikerParams
+    ) -> Union[List["lsp_types.Moniker"], None]:
+        """A request to get the moniker of a symbol at a given text document position.
+        The request parameter is of type {@link TextDocumentPositionParams}.
+        The response is of type {@link Moniker Moniker[]} or `null`."""
+        return await self.send_request("textDocument/moniker", params)
+
+    async def prepare_type_hierarchy(
+            self, params: lsp_types.TypeHierarchyPrepareParams
+    ) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
+        """A request to result a `TypeHierarchyItem` in a document at a given position.
+        Can be used as an input to a subtypes or supertypes type hierarchy.
+
+        @since 3.17.0"""
+        return await self.send_request("textDocument/prepareTypeHierarchy", params)
+
+    async def type_hierarchy_supertypes(
+            self, params: lsp_types.TypeHierarchySupertypesParams
+    ) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
+        """A request to resolve the supertypes for a given `TypeHierarchyItem`.
+
+        @since 3.17.0"""
+        return await self.send_request("typeHierarchy/supertypes", params)
+
+    async def type_hierarchy_subtypes(
+            self, params: lsp_types.TypeHierarchySubtypesParams
+    ) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
+        """A request to resolve the subtypes for a given `TypeHierarchyItem`.
+
+        @since 3.17.0"""
+        return await self.send_request("typeHierarchy/subtypes", params)
+
+    async def inline_value(
+            self, params: lsp_types.InlineValueParams
+    ) -> Union[List["lsp_types.InlineValue"], None]:
+        """A request to provide inline values in a document. The request's parameter is of
+        type {@link InlineValueParams}, the response is of type
+        {@link InlineValue InlineValue[]} or a Thenable that resolves to such.
+
+        @since 3.17.0"""
+        return await self.send_request("textDocument/inlineValue", params)
+
+    async def inlay_hint(
+            self, params: lsp_types.InlayHintParams
+    ) -> Union[List["lsp_types.InlayHint"], None]:
+        """A request to provide inlay hints in a document. The request's parameter is of
+        type {@link InlayHintsParams}, the response is of type
+        {@link InlayHint InlayHint[]} or a Thenable that resolves to such.
+
+        @since 3.17.0"""
+        return await self.send_request("textDocument/inlayHint", params)
+
+    async def resolve_inlay_hint(
+            self, params: lsp_types.InlayHint
+    ) -> "lsp_types.InlayHint":
+        """A request to resolve additional properties for an inlay hint.
+        The request's parameter is of type {@link InlayHint}, the response is
+        of type {@link InlayHint} or a Thenable that resolves to such.
+
+        @since 3.17.0"""
+        return await self.send_request("inlayHint/resolve", params)
+
+    async def text_document_diagnostic(
+            self, params: lsp_types.DocumentDiagnosticParams
+    ) -> "lsp_types.DocumentDiagnosticReport":
+        """The document diagnostic request definition.
+
+        @since 3.17.0"""
+        return await self.send_request("textDocument/diagnostic", params)
+
+    async def workspace_diagnostic(
+            self, params: lsp_types.WorkspaceDiagnosticParams
+    ) -> "lsp_types.WorkspaceDiagnosticReport":
+        """The workspace diagnostic request definition.
+
+        @since 3.17.0"""
+        return await self.send_request("workspace/diagnostic", params)
+
+    async def initialize(
+            self, params: lsp_types.InitializeParams
+    ) -> "lsp_types.InitializeResult":
+        """The initialize request is sent from the client to the server.
+        It is sent once as the request after starting up the server.
+        The requests parameter is of type {@link InitializeParams}
+        the response if of type {@link InitializeResult} of a Thenable that
+        resolves to such."""
+        return await self.send_request("initialize", params)
+
+    async def shutdown(self) -> None:
+        """A shutdown request is sent from the client to the server.
+        It is sent once when the client decides to shutdown the
+        server. The only notification that is sent after a shutdown request
+        is the exit event."""
+        return await self.send_request("shutdown")
+
+    async def will_save_wait_until(
+            self, params: lsp_types.WillSaveTextDocumentParams
+    ) -> Union[List["lsp_types.TextEdit"], None]:
+        """A document will save request is sent from the client to the server before
+        the document is actually saved. The request can return an array of TextEdits
+        which will be applied to the text document before it is saved. Please note that
+        clients might drop results if computing the text edits took too long or if a
+        server constantly fails on this request. This is done to keep the save fast and
+        reliable."""
+        return await self.send_request("textDocument/willSaveWaitUntil", params)
+
+    async def completion(
+            self, params: lsp_types.CompletionParams
+    ) -> Union[List["lsp_types.CompletionItem"], "lsp_types.CompletionList", None]:
+        """Request to request completion at a given text document position. The request's
+        parameter is of type {@link TextDocumentPosition} the response
+        is of type {@link CompletionItem CompletionItem[]} or {@link CompletionList}
+        or a Thenable that resolves to such.
+
+        The request can delay the computation of the {@link CompletionItem.detail `detail`}
+        and {@link CompletionItem.documentation `documentation`} properties to the `completionItem/resolve`
+        request. However, properties that are needed for the initial sorting and filtering, like `sortText`,
+        `filterText`, `insertText`, and `textEdit`, must not be changed during resolve.
+        """
+        return await self.send_request("textDocument/completion", params)
+
+    async def resolve_completion_item(
+            self, params: lsp_types.CompletionItem
+    ) -> "lsp_types.CompletionItem":
+        """Request to resolve additional information for a given completion item.The request's
+        parameter is of type {@link CompletionItem} the response
+        is of type {@link CompletionItem} or a Thenable that resolves to such."""
+        return await self.send_request("completionItem/resolve", params)
+
+    async def hover(
+            self, params: lsp_types.HoverParams
+    ) -> Union["lsp_types.Hover", None]:
+        """Request to request hover information at a given text document position. The request's
+        parameter is of type {@link TextDocumentPosition} the response is of
+        type {@link Hover} or a Thenable that resolves to such."""
+        return await self.send_request("textDocument/hover", params)
+
+    async def signature_help(
+            self, params: lsp_types.SignatureHelpParams
+    ) -> Union["lsp_types.SignatureHelp", None]:
+        return await self.send_request("textDocument/signatureHelp", params)
+
+    async def definition(
+            self, params: lsp_types.DefinitionParams
+    ) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
+        """A request to resolve the definition location of a symbol at a given text
+        document position. The request's parameter is of type [TextDocumentPosition]
+        (#TextDocumentPosition) the response is of either type {@link Definition}
+        or a typed array of {@link DefinitionLink} or a Thenable that resolves
+        to such."""
+        return await self.send_request("textDocument/definition", params)
+
+    async def references(
+            self, params: lsp_types.ReferenceParams
+    ) -> Union[List["lsp_types.Location"], None]:
+        """A request to resolve project-wide references for the symbol denoted
+        by the given text document position. The request's parameter is of
+        type {@link ReferenceParams} the response is of type
+        {@link Location Location[]} or a Thenable that resolves to such."""
+        return await self.send_request("textDocument/references", params)
+
+    async def document_highlight(
+            self, params: lsp_types.DocumentHighlightParams
+    ) -> Union[List["lsp_types.DocumentHighlight"], None]:
+        """Request to resolve a {@link DocumentHighlight} for a given
+        text document position. The request's parameter is of type [TextDocumentPosition]
+        (#TextDocumentPosition) the request response is of type [DocumentHighlight[]]
+        (#DocumentHighlight) or a Thenable that resolves to such."""
+        return await self.send_request("textDocument/documentHighlight", params)
+
+    async def document_symbol(
+            self, params: lsp_types.DocumentSymbolParams
+    ) -> Union[
+        List["lsp_types.SymbolInformation"], List["lsp_types.DocumentSymbol"], None
+    ]:
+        """A request to list all symbols found in a given text document. The request's
+        parameter is of type {@link TextDocumentIdentifier} the
+        response is of type {@link SymbolInformation SymbolInformation[]} or a Thenable
+        that resolves to such."""
+        return await self.send_request("textDocument/documentSymbol", params)
+
+    async def code_action(
+            self, params: lsp_types.CodeActionParams
+    ) -> Union[List[Union["lsp_types.Command", "lsp_types.CodeAction"]], None]:
+        """A request to provide commands for the given text document and range."""
+        return await self.send_request("textDocument/codeAction", params)
+
+    async def resolve_code_action(
+            self, params: lsp_types.CodeAction
+    ) -> "lsp_types.CodeAction":
+        """Request to resolve additional information for a given code action.The request's
+        parameter is of type {@link CodeAction} the response
+        is of type {@link CodeAction} or a Thenable that resolves to such."""
+        return await self.send_request("codeAction/resolve", params)
+
+    async def workspace_symbol(
+            self, params: lsp_types.WorkspaceSymbolParams
+    ) -> Union[
+        List["lsp_types.SymbolInformation"], List["lsp_types.WorkspaceSymbol"], None
+    ]:
+        """A request to list project-wide symbols matching the query string given
+        by the {@link WorkspaceSymbolParams}. The response is
+        of type {@link SymbolInformation SymbolInformation[]} or a Thenable that
+        resolves to such.
+
+        @since 3.17.0 - support for WorkspaceSymbol in the returned data. Clients
+         need to advertise support for WorkspaceSymbols via the client capability
+         `workspace.symbol.resolveSupport`.
+        """
+        return await self.send_request("workspace/symbol", params)
+
+    async def resolve_workspace_symbol(
+            self, params: lsp_types.WorkspaceSymbol
+    ) -> "lsp_types.WorkspaceSymbol":
+        """A request to resolve the range inside the workspace
+        symbol's location.
+
+        @since 3.17.0"""
+        return await self.send_request("workspaceSymbol/resolve", params)
+
+    async def code_lens(
+            self, params: lsp_types.CodeLensParams
+    ) -> Union[List["lsp_types.CodeLens"], None]:
+        """A request to provide code lens for the given text document."""
+        return await self.send_request("textDocument/codeLens", params)
+
+    async def resolve_code_lens(
+            self, params: lsp_types.CodeLens
+    ) -> "lsp_types.CodeLens":
+        """A request to resolve a command for a given code lens."""
+        return await self.send_request("codeLens/resolve", params)
+
+    async def document_link(
+            self, params: lsp_types.DocumentLinkParams
+    ) -> Union[List["lsp_types.DocumentLink"], None]:
+        """A request to provide document links"""
+        return await self.send_request("textDocument/documentLink", params)
+
+    async def resolve_document_link(
+            self, params: lsp_types.DocumentLink
+    ) -> "lsp_types.DocumentLink":
+        """Request to resolve additional information for a given document link. The request's
+        parameter is of type {@link DocumentLink} the response
+        is of type {@link DocumentLink} or a Thenable that resolves to such."""
+        return await self.send_request("documentLink/resolve", params)
+
+    async def formatting(
+            self, params: lsp_types.DocumentFormattingParams
+    ) -> Union[List["lsp_types.TextEdit"], None]:
+        """A request to to format a whole document."""
+        return await self.send_request("textDocument/formatting", params)
+
+    async def range_formatting(
+            self, params: lsp_types.DocumentRangeFormattingParams
+    ) -> Union[List["lsp_types.TextEdit"], None]:
+        """A request to to format a range in a document."""
+        return await self.send_request("textDocument/rangeFormatting", params)
+
+    async def on_type_formatting(
+            self, params: lsp_types.DocumentOnTypeFormattingParams
+    ) -> Union[List["lsp_types.TextEdit"], None]:
+        """A request to format a document on type."""
+        return await self.send_request("textDocument/onTypeFormatting", params)
+
+    async def rename(
+            self, params: lsp_types.RenameParams
+    ) -> Union["lsp_types.WorkspaceEdit", None]:
+        """A request to rename a symbol."""
+        return await self.send_request("textDocument/rename", params)
+
+    async def prepare_rename(
+            self, params: lsp_types.PrepareRenameParams
+    ) -> Union["lsp_types.PrepareRenameResult", None]:
+        """A request to test and perform the setup necessary for a rename.
+
+        @since 3.16 - support for default behavior"""
+        return await self.send_request("textDocument/prepareRename", params)
+
+    async def execute_command(
+            self, params: lsp_types.ExecuteCommandParams
+    ) -> Union["lsp_types.LSPAny", None]:
+        """A request send from the client to the server to execute a command. The request might return
+        a workspace edit which the client will apply to the workspace."""
+        return await self.send_request("workspace/executeCommand", params)