如何从外部进程（如Vim）调用Jupyter Notebook API创建并执行单元格

刚好我之前研究过类似的需求，完全可以通过Jupyter Notebook的两套核心API来实现——REST API负责单元格的增删改操作，WebSocket API负责触发代码执行并让浏览器同步更新结果。下面给你一步步拆解具体实现方案：

一、先搞定Jupyter的认证信息 #

Jupyter默认开启了安全验证，所以第一步得拿到服务器地址和token：

• 运行终端命令 jupyter notebook list，会输出正在运行的服务器信息，比如：

  Currently running servers:
  http://localhost:8888/?token=abc123def456 :: /home/user/notebooks
  

  这里的http://localhost:8888是服务器地址，abc123def456就是你需要的认证token。
• 后续所有API请求都要带上这个token，要么在URL参数里加?token=xxx，要么在请求头里添加Authorization: token abc123def456。

二、用REST API操作单元格（增/改） #

假设你已经有一个打开的Notebook（比如test.ipynb），先通过REST API来修改它的内容：

1. 获取Notebook当前内容 #

发送GET请求到 http://<服务器地址>:<端口>/api/contents/<Notebook路径>，带上认证头。返回的JSON数据里，content.cells字段就是所有单元格的列表，每个单元格包含cell_type（代码/ markdown）、source（内容）等关键信息。

2. 添加新代码单元格 #

发送PUT请求到同一个地址，把修改后的Notebook内容作为JSON请求体传回去。比如要在Notebook末尾加一个代码单元格，构造的新cell结构如下：

{
  "cell_type": "code",
  "source": ["print('Hello from Vim!')", "import matplotlib.pyplot as plt", "plt.plot([1,2,3])", "plt.show()"],
  "execution_count": null,
  "outputs": [],
  "metadata": {}
}

把这个cell对象追加到content.cells列表的末尾，然后发送PUT请求即可更新Notebook。

3. 编辑现有单元格 #

同样用PUT请求，找到目标单元格在content.cells里的索引，替换它的source字段为你从Vim选中的代码，再发送更新请求就行。

三、用WebSocket API触发执行并同步浏览器 #

这一步是核心——REST API只能修改内容，要让代码执行并让浏览器自动同步结果，必须用WebSocket：

1. 获取Kernel ID #

每个Notebook都对应一个运行的Kernel，先拿到它的ID：

• 发送GET请求到 http://<服务器地址>:<端口>/api/sessions，遍历返回的会话列表，找到对应Notebook的条目，里面的kernel.id就是你需要的Kernel ID。

2. 建立WebSocket连接 #

连接地址格式为 ws://<服务器地址>:<端口>/api/kernels/<Kernel ID>/channels，记得带上token参数（比如?token=abc123def456）。

3. 发送执行请求 #

连接建立后，向WebSocket发送一个JSON格式的执行请求，示例如下：

{
  "header": {
    "msg_id": "随便生成一个唯一ID（比如用uuid）",
    "msg_type": "execute_request",
    "username": "你的用户名",
    "session": "随便生成的会话ID",
    "version": "5.3"
  },
  "parent_header": {},
  "metadata": {},
  "content": {
    "code": "你要执行的代码内容",
    "silent": false,
    "store_history": true,
    "user_expressions": {},
    "allow_stdin": false
  },
  "channel": "shell"
}

这里的code直接填你从Vim选中的代码即可。

4. 自动同步浏览器 #

WebSocket会实时返回执行相关的消息（比如执行状态、控制台输出、matplotlib图表数据），Jupyter的前端会自动接收这些消息并更新页面——所以你不用额外做任何操作，浏览器就能同步显示单元格的执行结果。

四、结合Vim插件的实现思路 #

在Vim插件里，你可以这么做：

• 让用户配置Jupyter服务器地址和token（比如存在g:jupyter_server和g:jupyter_token全局变量里）
• 监听用户的选中文本操作，把选中的代码行拼接成字符串
• 调用脚本（比如Python脚本）处理API请求：先通过REST API把代码添加/更新到Notebook，再通过WebSocket触发执行
• 可以在Vim状态栏显示执行状态（比如「正在执行...」「执行完成」）

五、简化方案：用现成工具库 #

如果不想自己写HTTP和WebSocket请求，可以用Python的jupyter_client库，它封装了Jupyter的所有API。你可以在Vim插件里调用Python脚本（比如用!python3 jupyter_execute.py <选中代码>）来处理所有逻辑，不用在VimScript里写复杂的网络请求。

举个简单的Python脚本示例（实现添加并执行单元格）：

import requests
import websocket
import json
import uuid

# 配置信息，可从Vim变量传入
JUPYTER_SERVER = "http://localhost:8888"
TOKEN = "abc123def456"
NOTEBOOK_PATH = "test.ipynb"
CODE = "\n".join([
    "print('Hello from Vim!')",
    "import matplotlib.pyplot as plt",
    "plt.plot([1,2,3])",
    "plt.show()"
])

# 1. 获取Notebook内容
headers = {"Authorization": f"token {TOKEN}"}
notebook_url = f"{JUPYTER_SERVER}/api/contents/{NOTEBOOK_PATH}"
notebook = requests.get(notebook_url, headers=headers).json()

# 2. 添加新代码单元格
new_cell = {
    "cell_type": "code",
    "source": CODE.split("\n"),
    "execution_count": None,
    "outputs": [],
    "metadata": {}
}
notebook["content"]["cells"].append(new_cell)
requests.put(notebook_url, headers=headers, json=notebook)

# 3. 获取Kernel ID
sessions = requests.get(f"{JUPYTER_SERVER}/api/sessions", headers=headers).json()
kernel_id = next(s["kernel"]["id"] for s in sessions if s["path"] == NOTEBOOK_PATH)

# 4. 建立WebSocket连接并执行代码
ws_url = f"ws://localhost:8888/api/kernels/{kernel_id}/channels?token={TOKEN}"
ws = websocket.create_connection(ws_url)

# 构造执行请求
execute_msg = {
    "header": {
        "msg_id": str(uuid.uuid4()),
        "msg_type": "execute_request",
        "username": "user",
        "session": str(uuid.uuid4()),
        "version": "5.3"
    },
    "parent_header": {},
    "metadata": {},
    "content": {
        "code": CODE,
        "silent": False,
        "store_history": True,
        "user_expressions": {},
        "allow_stdin": False
    },
    "channel": "shell"
}
ws.send(json.dumps(execute_msg))

# 接收并处理执行结果（可选，可在Vim里显示）
while True:
    msg = json.loads(ws.recv())
    if msg["header"]["msg_type"] == "execute_reply":
        print("✅ 执行完成")
        break
    elif msg["header"]["msg_type"] == "stream":
        print(f"📤 输出: {msg['content']['text']}")
    elif msg["header"]["msg_type"] == "display_data":
        print("📊 图表已生成，浏览器已更新")

ws.close()

内容的提问来源于stack exchange，提问作者eyeApps LLC