第 13 章：LSP 协议与最小 Server

Content-Length: 85

{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"processId":null,"rootUri":null,"capabilities":{}}}

#!/usr/bin/env python3
import json
import sys


class LSPServer:
    def __init__(self):
        self.running = True

    def read_message(self):
        header = {}
        while True:
            line = sys.stdin.readline()
            if not line:
                return None
            line = line.strip()
            if not line:
                break
            key, value = line.split(":", 1)
            header[key.strip()] = value.strip()

        length = int(header.get("Content-Length", 0))
        body = sys.stdin.read(length)
        return json.loads(body)

    def send_message(self, payload):
        body = json.dumps(payload)
        sys.stdout.write(f"Content-Length: {len(body)}\r\n\r\n{body}")
        sys.stdout.flush()

    def handle(self, request):
        method = request.get("method")
        req_id = request.get("id")

        if method == "initialize":
            return self._ok(req_id, {
                "capabilities": {
                    "textDocumentSync": 1,
                    "hoverProvider": True,
                },
                "serverInfo": {"name": "minimal-lsp", "version": "0.1.0"}
            })

        if method == "initialized":
            return None  # 通知无需响应

        if method == "textDocument/hover":
            return self._ok(req_id, {
                "contents": {
                    "kind": "markdown",
                    "value": "**Minimal LSP**: This is a placeholder hover response."
                }
            })

        if method == "shutdown":
            return self._ok(req_id, None)

        if method == "exit":
            self.running = False
            return None

        return self._error(req_id, -32601, f"Method not found: {method}")

    def _ok(self, req_id, result):
        return {"jsonrpc": "2.0", "id": req_id, "result": result}

    def _error(self, req_id, code, message):
        return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}

    def run(self):
        while self.running:
            msg = self.read_message()
            if msg is None:
                break
            response = self.handle(msg)
            if response is not None:
                self.send_message(response)


if __name__ == "__main__":
    LSPServer().run()

python -c "
import json, sys
body = json.dumps({'jsonrpc':'2.0','id':1,'method':'initialize','params':{'processId':None,'rootUri':None,'capabilities':{}}})
sys.stdout.write(f'Content-Length: {len(body)}\r\n\r\n{body}')
" | python lsp_server.py

import * as readline from "readline";

function send(payload: unknown) {
  const body = JSON.stringify(payload);
  process.stdout.write(`Content-Length: ${body.length}\r\n\r\n${body}`);
}

const rl = readline.createInterface({ input: process.stdin });
let buffer = "";
let length = 0;

rl.on("line", (line) => {
  if (line.startsWith("Content-Length:")) {
    length = parseInt(line.split(":")[1].trim());
    buffer = "";
  } else if (line === "" && length > 0) {
    process.stdin.once("data", (chunk) => {
      const req = JSON.parse(chunk.toString().slice(0, length));
      if (req.method === "initialize") {
        send({ jsonrpc: "2.0", id: req.id, result: { capabilities: { hoverProvider: true } } });
      }
    });
  }
});