An everlasting objective of every developer is to code faster and more accurately.

Towards this goal, Microsoft created the Language Server Protocol (LSP). Nowadays, LSP client implementations are a must have for modern text editors. Those provide powerful language specific features for code editing, such as completion, error detection and refactoring.

This article explains the principles of the protocol, and follows the implementation of a basic client side binary.

Principles of the protocol

Concept

The LSP offers guidelines for communicating between a client and a server. In most scenarios, the client is a text editor or an IDE, and the server is a language specific binary. The two processes run simultaneously and exchange context dependent information.

The main application for this protocol is code analysis. Its most common features are code completion, error detection and refactoring. Some more specific features also exist for some languages such as symbol renaming or jumping to a symbol definition.

Origins

With the fast growing number of programming languages among the years comes the multiplication of text editors and IDEs, leading to each editor trying to integrate each language. Originally addressing this issue for the Microsoft Visual Studio Code IDE, Microsoft opened and standardized the LSP in 2016 with the help of Red Had and Codenvy. With this protocol, different code analysis features were aggregated and could be handled by a single program (i.e. one per programming language), avoiding the need for development tools to re-design their code interaction process for each programming language.

Technical overview

The LSP is based on the JSON-RPC protocol. In short, this implies all messages exchanged by the client and the server compose JSON objects. Additionally, client requests must always contain at least:

  • An ID
  • A method name
  • Parameters

And server responses include:

  • The same ID as the request
  • A result
  • An error indicator

A nice overview of a go to definition request and response is available on the protocol’s official website.

microsoft-overview-schema

For each feature of the language server exists a request and a response type available through an API, while notifications are cross-feature. Available features are usually listed on the language server’s documentation, but an editor can also send a known request and determine with the response whether this feature is implemented.

Additionally, depending on the language, servers can pre-compile code in the background in order to provide appropriate support. This is mostly the case for statically typed languages such as Java or C, as opposed to Javascript or Python.

Application

The following section explores the step by step implementation of a minimal LSP client in Python, capable of sending and receiving one request. Here, the clangd language server is used, but with a bit more tuning, this program could become generic.

Note that an official SDK to ease the creation of LSP clients and servers is provided as an NPM module on the LSP repository, but it will not be used in this article to avoid having too much abstraction.

Server initialization

In Python, the clangd binary can be launched as any other subprocess.

1
2
3
4

    server = subprocess.Popen('clangd',
                              stdin=subprocess.PIPE,
                              stdout=subprocess.PIPE,
                              stderr=sys.stderr)

At the end of the file, let’s also terminate the server properly.

1

    server.terminate()

Before being able to interact with code, the server has to be initialized with a specific request. The following snippet contains only the required headers for the initialization to be successful.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

    json_content = json.dumps({
        'jsonrpc': '2.0',
        'id': new_request_id(),
        'method': 'initialize',
        'params': {
            'processId': os.getpid(),
            'rootUri': pathlib.Path.cwd().as_uri(),
            'capabilities': {},
        },
    })

At this point, the content can be prefixed with its length and written to the server’s input. Note that by default, clangd is receiving and writing its IO from stdin and stdout.

1
2
3

    message = f'Content-Length: {len(json_content)}\r\n\r\n{json_content}'
    server.stdin.write(message.encode('utf-8'))
    server.stdin.flush()

According to clangd’s logs, the initialization is now correct.

1
2

I[20:11:38.475] <-- initialize(1)
I[20:11:38.477] --> reply:initialize(1) 1 ms

Finally, let’s refactor this logic into a class.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class Server():
    def __init__(self):
        self.request_id = 0
        self.process = subprocess.Popen('clangd',
                                        stdin=subprocess.PIPE,
                                        stdout=subprocess.PIPE,
                                        stderr=sys.stderr)
        self.send_request('initialize', {
            'processId': os.getpid(),
            'rootUri': pathlib.Path.cwd().as_uri(),
            'capabilities': {},
        })

    def __del__(self):
        self.process.terminate()

    def new_request_id(self) -> int:
        self.request_id += 1
        return self.request_id

    def send(self, message: dict):
        message['jsonrpc'] = '2.0'
        json_content = json.dumps(message)
        str_content = f'Content-Length: {len(json_content)}\r\n\r\n{json_content}'
        self.process.stdin.write(str_content.encode('utf-8'))
        self.process.stdin.flush()
        time.sleep(0.5)

    def send_request(self, method: str, params: dict):
        self.send({
            'id': self.new_request_id(),
            'method': method,
            'params': params,
            'message_type': 'request',
        })

Notifications

In order to interact with code, the client needs to inform the server that a file was opened. This can be done using notifications.

Let’s implement this by adding the following method to the Server class:

1
2
3
4
5
6

    def send_notification(self, method: str, params: dict):
        self.send({
            'method': method,
            'params': params,
            'message_type': 'notification',
        })

Then, after having added a state (an open_files variable) in the class keeping track of which files are currently open, open_file and close_file methods can be easily implemented:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

    def open_file(self, path: pathlib.Path, language_id: str):
        if path in self.open_files:
            return
        with open(path, encoding="utf-8") as f:
            self.send_notification('textDocument/didOpen', {
                'textDocument': {
                    'uri': path.absolute().as_uri(),
                    'languageId': language_id,
                    'version': 0,
                    'text': f.read(),
                },
            })
        self.open_files.add(path)

    def close_file(self, path: pathlib.Path):
        if path not in self.open_files:
            return
        self.send_notification('textDocument/didClose', {
            'textDocument': {
                'uri': path.absolute().as_uri(),
            },
        })
        self.open_files.remove(path)

Actual code request

Now that the basic bricks have been set up, it becomes possible to really interact with the code. This sections implements a simple go to definition request.

The following method can be added to the Server class.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

    def go_to_definition(self, path: pathlib.Path, location: tuple):
        if path not in self.open_files:
            return
        self.send_request('textDocument/definition', {
            'textDocument': {
                'uri': path.absolute().as_uri(),
            },
            'position': {
                'line': location[0],
                'character': location[1],
            },
        })

The last detail is getting the response from the language server. A function that reads on the process output according to the content length does the job. Beware to consume each response in order to find the right one for a given request.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

    def recv_response(self) -> str:
        buffer = bytearray()
        match = None
        while match is None:
            buffer += self.process.stdout.readline()
            match = re.search(b'Content-Length: [0-9]+\r\n\r\n', buffer)
        content_length = int(match.group(0).decode('utf-8').strip().split(':')[1])
        buffer = buffer[match.end():]
        while len(buffer) < content_length:
            buffer += self.process.stdout.readline(content_length - len(buffer))
        return json.loads(buffer.decode('utf-8'))

Full demonstration

The program implemented above should output a go to definition for the following C program.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

#include <stdio.h>

void hello(void)
{
    printf("Hello, world!\n");
}

int main(void)
{
    hello();
    return 0;
}

After adding a simple main function to call previous methods from the Server class, the expected result correctly returned is returned by the language server.

1
2
3
4
5
6
7
8

def main():
    server = Server()

    hello = pathlib.Path('hello.c')
    server.open_file(hello, 'c')
    server.recv_response()
    print(server.go_to_definition(hello, (9, 4)))
    server.close_file(hello)

1

{'id': 2, 'jsonrpc': '2.0', 'result': [{'range': {'end': {'character': 10, 'line': 2}, 'start': {'character': 5, 'line': 2}}, 'uri': 'file:///home/tristan/hello.c'}]}

Conclusion

The LSP allows many hard tasks regarding code edition and interactions to be aggregated into a single program. This massively eases client tasks.

The full example shown above is far from perfect: multiple parameters have been hardcoded, and it lacks proper interfacing with an editor. On another hand, it shows that implementing a client can be quite succinct after the basic communication aspects of the protocol have been handled.

Sources

Appendix

Full Python code used for the demonstration:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

import subprocess
import time
import sys
import pathlib
import json
import os
import re

class Server():
    def __init__(self):
        self.request_id = 0
        self.open_files = set()
        self.process = subprocess.Popen('clangd',
                                        stdin=subprocess.PIPE,
                                        stdout=subprocess.PIPE,
                                        stderr=sys.stderr)
        self.send_request('initialize', {
            'processId': os.getpid(),
            'rootUri': pathlib.Path.cwd().as_uri(),
            'capabilities': {},
        })
        self.recv_response()

    def __del__(self):
        self.process.terminate()

    def new_request_id(self) -> int:
        self.request_id += 1
        return self.request_id

    def send(self, message: dict):
        message['jsonrpc'] = '2.0'
        json_content = json.dumps(message)
        str_content = f'Content-Length: {len(json_content)}\r\n\r\n{json_content}'
        self.process.stdin.write(str_content.encode('utf-8'))
        self.process.stdin.flush()
        time.sleep(0.5)

    def send_request(self, method: str, params: dict):
        self.send({
            'id': self.new_request_id(),
            'method': method,
            'params': params,
            'message_type': 'request',
        })

    def send_notification(self, method: str, params: dict):
        self.send({
            'method': method,
            'params': params,
            'message_type': 'notification',
        })

    def recv_response(self) -> str:
        buffer = bytearray()
        match = None
        while match is None:
            buffer += self.process.stdout.readline()
            match = re.search(b'Content-Length: [0-9]+\r\n\r\n', buffer)
        content_length = int(match.group(0).decode('utf-8').strip().split(':')[1])
        buffer = buffer[match.end():]
        while len(buffer) < content_length:
            buffer += self.process.stdout.readline(content_length - len(buffer))
        return json.loads(buffer.decode('utf-8'))

    def open_file(self, path: pathlib.Path, language_id: str):
        if path in self.open_files:
            return
        with open(path, encoding="utf-8") as f:
            self.send_notification('textDocument/didOpen', {
                'textDocument': {
                    'uri': path.absolute().as_uri(),
                    'languageId': language_id,
                    'version': 0,
                    'text': f.read(),
                },
            })
        self.open_files.add(path)

    def close_file(self, path: pathlib.Path):
        if path not in self.open_files:
            return
        self.send_notification('textDocument/didClose', {
            'textDocument': {
                'uri': path.absolute().as_uri(),
            },
        })
        self.open_files.remove(path)

    def go_to_definition(self, path: pathlib.Path, location: tuple) -> str:
        if path not in self.open_files:
            return
        self.send_request('textDocument/definition', {
            'textDocument': {
                'uri': path.absolute().as_uri(),
            },
            'position': {
                'line': location[0],
                'character': location[1],
            },
        })
        return self.recv_response()


def main():
    server = Server()

    hello = pathlib.Path('hello.c')
    server.open_file(hello, 'c')
    server.recv_response()
    print(server.go_to_definition(hello, (9, 4)))
    server.close_file(hello)

if __name__ == '__main__':
    main()