You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何通过pyright-langserver的stdio模式实现LSP标准JSONRPC交互以获取实时代码提示

如何通过pyright-langserver的stdio模式实现LSP标准JSONRPC交互以获取实时代码提示

我仔细看了你的问题和脚本,发现是几个不符合LSP规范的细节导致pyright-langserver没有正常响应你的请求。咱们一步步来解决这些问题:

问题1:初始化请求参数不完整

LSP的initialize请求有几个必填参数,你只传了rootUri,缺少processIdcapabilities这些关键字段,pyright会因为参数不合法而拒绝处理后续请求。

问题2:消息格式不符合LSP规范

你直接把JSON字符串加换行发送,但LSP要求每个消息必须以Content-Length头开头,后跟空行,再是JSON内容。你的脚本没有构造这个头,导致服务器无法解析你的请求。

问题3:未发送initialized通知

在收到initialize响应后,必须发送initialized通知告诉服务器初始化完成,否则它不会处理任何后续的文本文档相关请求(比如补全)。

问题4:未处理服务器的通知消息

服务器启动时会发送window/logMessage这类通知(没有id字段的JSONRPC消息),你的read_response函数只处理带id的响应,会卡在读取缓冲区的步骤,因为这些通知还没被消费。


下面是修正后的完整脚本,我注释了每一处修改:

import subprocess
import json
import time

process = subprocess.Popen(
    ["pyright-langserver", "--stdio"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE  # 加上stderr以便查看服务器错误输出
)

def construct_lsp_message(content):
    # 构造符合LSP规范的消息:Content-Length头 + 空行 + JSON内容
    json_str = json.dumps(content)
    length = len(json_str.encode('utf-8'))
    return f"Content-Length: {length}\r\n\r\n{json_str}".encode('utf-8')

def send_message(message):
    # 发送构造好的LSP消息
    process.stdin.write(message)
    process.stdin.flush()

def read_next_message():
    # 读取服务器发送的单个LSP消息(兼容通知和响应)
    buffer = b""
    # 先读取头部,直到找到空行分隔符
    while b"\r\n\r\n" not in buffer:
        chunk = process.stdout.read(1)
        if not chunk:
            # 服务器关闭了连接
            return None
        buffer += chunk
    # 拆分头部和内容
    headers_part, content_part = buffer.split(b"\r\n\r\n", 1)
    # 解析Content-Length
    content_length = None
    for header in headers_part.split(b"\r\n"):
        if header.startswith(b"Content-Length"):
            content_length = int(header.split(b":")[1].strip())
            break
    if not content_length:
        raise ValueError("Missing Content-Length header")
    # 读取剩余的内容直到达到指定长度
    while len(content_part) < content_length:
        content_part += process.stdout.read(content_length - len(content_part))
    # 解析JSON
    return json.loads(content_part.decode('utf-8'))

def consume_initial_notifications():
    # 消费服务器启动时发送的初始化通知(比如window/logMessage)
    print("Reading initial server notifications...")
    while True:
        msg = read_next_message()
        if not msg:
            break
        # 如果是带有id的响应,说明是initialize的响应,返回它
        if "id" in msg:
            return msg
        # 否则是通知,打印出来
        if msg.get("method") == "window/logMessage":
            print(f"Server log: {msg['params']['message']}")

# 1. 发送initialize请求(参数完整)
initialize_msg = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "processId": None,  # 可以传None或者当前进程ID
        "rootUri": "file:///Users/arpit/Programming/Python/test-project",
        "capabilities": {  # 至少传空的capabilities字段
            "textDocument": {},
            "workspace": {}
        }
    }
}
send_message(construct_lsp_message(initialize_msg))

# 2. 消费初始化通知,获取initialize响应
initialize_response = consume_initial_notifications()
print("Initialize response:", initialize_response)

# 3. 发送initialized通知,告诉服务器初始化完成
initialized_msg = {
    "jsonrpc": "2.0",
    "method": "initialized",
    "params": {}
}
send_message(construct_lsp_message(initialized_msg))

# 4. 先发送textDocument/didOpen通知,告诉服务器要处理的文件内容
did_open_msg = {
    "jsonrpc": "2.0",
    "method": "textDocument/didOpen",
    "params": {
        "textDocument": {
            "uri": "file:///Users/arpit/Programming/Python/test-project/main.py",
            "languageId": "python",
            "version": 1,
            # 必须传入文件的实际内容!否则服务器没有上下文,无法返回补全
            "text": open("/Users/arpit/Programming/Python/test-project/main.py", "r").read()
        }
    }
}
send_message(construct_lsp_message(did_open_msg))

# 5. 发送补全请求
completion_msg = {
    "jsonrpc": "2.0",
    "id": 2,
    "method": "textDocument/completion",
    "params": {
        "textDocument": {
            "uri": "file:///Users/arpit/Programming/Python/test-project/main.py"
        },
        "position": {"line": 7, "character": 20}
    }
}
send_message(construct_lsp_message(completion_msg))

# 读取补全响应
completion_response = read_next_message()
print("Completion response:", completion_response)

# 可选:如果文件内容变更,发送textDocument/didChange通知更新服务器上下文
# did_change_msg = {
#     "jsonrpc": "2.0",
#     "method": "textDocument/didChange",
#     "params": {
#         "textDocument": {
#             "uri": "file:///Users/arpit/Programming/Python/test-project/main.py",
#             "version": 2
#         },
#         "contentChanges": [{"text": "修改后的代码内容"}]
#     }
# }
# send_message(construct_lsp_message(did_change_msg))

关键修改说明

  1. construct_lsp_message函数:负责生成符合LSP规范的消息,自动添加Content-Length头和正确的分隔符,这是服务器能解析请求的核心。
  2. read_next_message函数:能读取任何类型的LSP消息(通知或响应),不再只等待带id的响应。
  3. consume_initial_notifications函数:先处理服务器启动时的日志通知,避免脚本卡住,同时捕获initialize的响应。
  4. 完整的初始化流程:严格遵循LSP的初始化步骤,确保服务器进入就绪状态。
  5. 添加textDocument/didOpen通知:必须告诉服务器当前打开的文件内容,否则它没有代码上下文,无法返回有效的补全建议。

备注:内容来源于stack exchange,提问作者Arpit Pandey

火山引擎 最新活动