pylsp_jsonrpc_server_with_hello_notification_and_add_request.py

1import threading
2from pylsp_jsonrpc.dispatchers import MethodDispatcher
3from pylsp_jsonrpc.endpoint import Endpoint
4from pylsp_jsonrpc.streams import JsonRpcStreamReader, JsonRpcStreamWriter
5
6class ExampleServer(MethodDispatcher):
7    def m_hello(self, name=None):
8        return {"message": f"Hello, {name}!" if name else "Hello!"}
9
10    def m_add(self, x, y):
11        return x + y
12
13def main():
14    # Use stdin/stdout for communication streams
15    import sys
16    reader = JsonRpcStreamReader(sys.stdin.buffer)
17    writer = JsonRpcStreamWriter(sys.stdout.buffer)
18
19    dispatcher = ExampleServer()
20    endpoint = Endpoint(dispatcher, writer.write)
21
22    # Start the reader loop in the main thread
23    reader.listen(endpoint.consume)
24
25if __name__ == "__main__":
26    main()